实时计算 11.11活动

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> salesData = env.addSource(kafkaConsumer);

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

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