流式计算 年末活动

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.api.common.functions.MapFunction;

public class SalesAnalysis {
    public static void main(String[] args) throws Exception {
        final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        DataStream<String> salesData = env.socketTextStream("localhost", 9999);

        DataStream<Sale> parsedSales = salesData.map(new MapFunction<String, Sale>() {
            @Override
            public Sale map(String value) {
                String[] fields = value.split(",");
                return new Sale(fields[0], Double.parseDouble(fields[1]), Integer.parseInt(fields[2]));
            }
        });

        parsedSales.keyBy("productId")
                   .timeWindow(Time.minutes(5))
                   .sum("amount")
                   .print();

        env.execute("Sales Analysis Job");
    }

    public static class Sale {
        public String productId;
        public double amount;
        public int quantity;

        // Constructor, getters, and setters
    }
}