Flink 与 Kafka：实现实时数据流的连续查询与窗口处理

<dependency>
    <groupId>org.apache.flink</groupId>
    <artifactId>flink-connector-kafka</artifactId>
    <version>1.17.1</version> <!-- 请根据您使用的 Flink 版本调整 -->
</dependency>
<dependency>
    <groupId>org.apache.flink</groupId>
    <artifactId>flink-streaming-java</artifactId>
    <version>1.17.1</version> <!-- Flink核心流处理API -->
    <scope>provided</scope>
</dependency>
<dependency>
    <groupId>org.apache.flink</groupId>
    <artifactId>flink-clients</artifactId>
    <version>1.17.1</version> <!-- Flink客户端，用于提交作业 -->
    <scope>provided</scope>
</dependency>

import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
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 java.time.Duration;
import java.util.Collections;

public class FlinkKafkaSourceExample {

    public static void main(String[] args) throws Exception {
        // 1. 获取流处理执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1); // 生产环境中应根据资源调整并行度

        // 2. 构建 Kafka Source
        KafkaSource<String> kafkaSource = KafkaSource.<String>builder()
                .setBootstrapServers("localhost:9092") // Kafka 集群的地址
                .setTopics(Collections.singletonList("my_input_topic")) // 要消费的主题列表
                .setGroupId("my_consumer_group") // 消费者组ID
                .setStartingOffsets(OffsetsInitializer.earliest()) // 从最早的偏移量开始消费
                .setValueOnlyDeserializer(new SimpleStringSchema()) // 指定反序列化器，这里使用简单的字符串反序列化
                .build();

        // 3. 将 Kafka Source 添加到 Flink 环境中，生成 DataStream
        // WatermarkStrategy.noWatermarks() 适用于处理时间语义，或在后续步骤中单独分配时间戳
        DataStream<String> kafkaStream = env.fromSource(
                kafkaSource,
                WatermarkStrategy.noWatermarks(), // 初始不分配水印，后续根据业务逻辑分配
                "Kafka Source"
        );

        // 4. 对数据流进行打印（仅用于演示）
        kafkaStream.print();

        // 5. 启动 Flink 作业
        env.execute("Flink Kafka Source Demo");
    }
}

import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.ReduceFunction;
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;

import java.time.Duration;
import java.util.Collections;

public class FlinkKafkaContinuousQuery {

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1); // 生产环境中应根据资源调整并行度

        // 1. 配置 Kafka Source
        KafkaSource<String> kafkaSource = KafkaSource.<String>builder()
                .setBootstrapServers("localhost:9092")
                .setTopics(Collections.singletonList("my_input_topic"))
                .setGroupId("flink_continuous_query_group")
                .setStartingOffsets(OffsetsInitializer.earliest())
                .setValueOnlyDeserializer(new SimpleStringSchema())
                .build();

        // 2. 从 Kafka 读取数据并分配事件时间戳和水印
        DataStream<String> kafkaStream = env.fromSource(
                kafkaSource,
                // 配置水印策略：允许5秒的乱序，并从数据中提取时间戳
                WatermarkStrategy.<String>forBoundedOutOfOrderness(Duration.ofSeconds(5))
                        .withTimestampAssigner((element, recordTimestamp) -> {
                            // 假设数据格式为 "key,value,timestamp_ms"
                            try {
                                String[] parts = element.split(",");
                                return Long.parseLong(parts[2]); // 提取第三部分作为事件时间戳
                            } catch (Exception e) {
                                // 错误处理，例如记录日志或返回当前时间
                                System.err.println("Failed to parse timestamp from: " + element + " - " + e.getMessage());
                                return System.currentTimeMillis();
                            }
                        }),
                "Kafka Source with Event Time"
        );

        // 3. 解析数据并进行 KeyBy 分组
        // 将原始字符串解析为 Tuple2<String, Long>，其中 f0 是 key，f1 是 value
        DataStream<Tuple2<String, Long>> parsedStream = kafkaStream
                .map(line -> {
                    String[] parts = line.split(",");
                    if (parts.length == 3) {
                        return Tuple2.of(parts[0], Long.parseLong(parts[1]));
                    } else {
                        System.err.println("Malformed record: " + line);
                        return Tuple2.of("unknown", 0L); // 默认值或错误处理
                    }
                })
                .keyBy(value -> value.f0); // 按 key (Tuple2 的 f0 字段) 进行分组

        // 4. 应用翻滚事件时间窗口并进行聚合
        // 每1分钟计算一次，对每个 key 在该窗口内的 value 进行求和
        DataStream<Tuple2<String, Long>> resultStream = parsedStream
                .window(TumblingEventTimeWindows.of(Time.minutes(1))) // 1分钟的翻滚窗口
                .reduce(new ReduceFunction<Tuple2<String, Long>>() {
                    @Override
                    public Tuple2<String, Long> reduce(Tuple2<String, Long> value1, Tuple2<String, Long> value2) throws Exception {
                        // 对相同 key 的 value 进行求和
                        return Tuple2.of(value1.f0, value1.f1 + value2.f1);
                    }
                });

        // 5. 打印结果到控制台
        resultStream.print("Windowed Sum by Key");

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