大数据消息处理双11促销活动

// Kafka消费者配置
Properties properties = new Properties();
properties.setProperty("bootstrap.servers", "localhost:9092");
properties.setProperty("group.id", "testGroup");
properties.setProperty("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
properties.setProperty("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");

KafkaConsumer<String, String> consumer = new KafkaConsumer<>(properties);
consumer.subscribe(Arrays.asList("sales-topic"));

// Flink流处理作业
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
DataStream<String> stream = env.addSource(new FlinkKafkaConsumer<>("sales-topic", new SimpleStringSchema(), properties));

stream.map(new MapFunction<String, SaleEvent>() {
    @Override
    public SaleEvent map(String value) throws Exception {
        // 解析消息并转换为SaleEvent对象
        return parseSaleEvent(value);
    }
}).keyBy("productId")
 .timeWindow(Time.seconds(10))
 .aggregate(new SaleAggregator())
 .print();

env.execute("Sales Event Processing");