流式计算 促销

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 PromotionAnalytics {
    public static void main(String[] args) throws Exception {
        final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        DataStream<String> promotions = env.addSource(new PromotionSource());

        DataStream<PromotionEvent> parsedPromotions = promotions.map(new MapFunction<String, PromotionEvent>() {
            @Override
            public PromotionEvent map(String value) throws Exception {
                return PromotionEvent.fromJson(value);
            }
        });

        parsedPromotions.keyBy("productId")
                        .timeWindow(Time.minutes(5))
                        .aggregate(new PromotionAggregator())
                        .addSink(new PromotionSink());

        env.execute("Promotion Analytics");
    }
}