Flink 与 Kafka 集成：实现流式数据连续查询教程

<dependencies>
    <!-- Flink 核心流处理 API -->
    <dependency>
        <groupId>org.apache.flink</groupId>
        <artifactId>flink-streaming-java</artifactId>
        <version>1.17.1</version> <!-- 请替换为您的 Flink 版本 -->
        <scope>provided</scope>
    </dependency>
    <!-- Flink 客户端，用于提交作业 -->
    <dependency>
        <groupId>org.apache.flink</groupId>
        <artifactId>flink-clients</artifactId>
        <version>1.17.1</version> <!-- 请替换为您的 Flink 版本 -->
        <scope>provided</scope>
    </dependency>
    <!-- Flink Kafka 连接器 -->
    <dependency>
        <groupId>org.apache.flink</groupId>
        <artifactId>flink-connector-kafka</artifactId>
        <version>1.17.1</version> <!-- 请替换为您的 Flink 版本 -->
    </dependency>
</dependencies>

import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.connector.kafka.source.KafkaSource;
import org.apache.flink.connector.kafka.source.enumerator.initializer.OffsetsInitializer;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;

public class KafkaFlinkContinuousQuery {

    public static void main(String[] args) throws Exception {
        // 1. 获取流处理执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        // 设置并行度，此处为简单示例，生产环境可根据需求调整
        env.setParallelism(1); 
        // 启用检查点，保证故障恢复和精确一次语义（生产环境强烈推荐）
        // env.enableCheckpointing(60 * 1000L); // 每60秒触发一次检查点

        // 2. 配置 Kafka Source
        // 假设 Kafka 运行在 localhost:9092，并且有一个名为 'my-input-topic' 的主题
        KafkaSource<String> source = KafkaSource.<String>builder()
                .setBootstrapServers("localhost:9092") // Kafka 集群地址
                .setTopics("my-input-topic") // 要消费的 Kafka 主题
                .setGroupId("my-flink-consumer-group") // 消费者组ID
                .setStartingOffsets(OffsetsInitializer.earliest()) // 从最早的偏移量开始消费
                .setValueOnlyDeserializer(new SimpleStringSchema()) // 使用 SimpleStringSchema 反序列化字符串
                .build();

        // 3. 从 Kafka 源创建数据流
        // WatermarkStrategy.noWatermarks() 适用于处理时间窗口，如果需要事件时间处理，请使用 WatermarkStrategy.forBoundedOutOfOrderness
        DataStream<String> kafkaStream = env.fromSource(source, WatermarkStrategy.noWatermarks(), "Kafka Source");

        // 4. 应用窗口处理逻辑：每5秒统计一次消息数量
        DataStream<Tuple2<String, Integer>> processedStream = kafkaStream
                // 将每条消息映射为一个Tuple2<String, Integer>，例如 <"message", 1>
                .map(message -> new Tuple2<>("total_messages", 1))
                // 按键分组，这里使用一个常量字符串作为键，使得所有消息进入同一个逻辑组，方便后续窗口操作
                .keyBy(value -> value.f0) 
                // 应用一个 5 秒的滚动事件时间窗口
                // 注意：由于上面使用了 WatermarkStrategy.noWatermarks()，这里实际上是处理时间窗口
                // 如果需要严格的事件时间窗口，需要正确生成 Watermark
                .window(TumblingEventTimeWindows.of(Time.seconds(5)))
                // 在每个窗口内，对消息数量进行累加
                .reduce((value1, value2) -> new Tuple2<>(value1.f0, value1.f1 + value2.f1));

        // 5. 将处理结果打印到控制台
        processedStream.print("Windowed Count");

        // 6. 启动 Flink 作业
        env.execute("Flink Kafka Continuous Query Example");
    }
}