实时计算 双十一活动

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;

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

        FlinkKafkaConsumer<String> kafkaConsumer = new FlinkKafkaConsumer<>("sales-topic", new SimpleStringSchema(), properties);
        DataStream<String> stream = env.addSource(kafkaConsumer);

        stream.map(new SaleMapper())
              .keyBy("productId")
              .timeWindow(Time.minutes(5))
              .aggregate(new SaleAggregator())
              .print();

        env.execute("Real-time Sales Processing");
    }
}