使用 Flink 和 Kafka 构建实时连续查询：入门与实践-java教程-PHP中文网

使用 flink 和 kafka 构建实时连续查询：入门与实践

本文提供了一份关于如何使用 Apache Flink 和 Apache Kafka 构建实时连续查询的全面指南。文章详细介绍了如何将 Kafka 作为持续数据源集成到 Flink 应用中，并利用 Flink 强大的窗口处理功能进行基于时间的事件聚合，旨在帮助初学者快速掌握核心概念并实践流处理解决方案。

1. Flink 与 Kafka：实时流处理的黄金组合

在现代数据架构中，实时数据处理变得越来越重要。Apache Kafka 作为分布式流平台，以其高吞吐量、持久性和可扩展性成为数据管道的核心。Apache Flink 则是一个强大的流处理框架，能够对无限数据流进行有状态计算。将 Kafka 作为 Flink 的数据源，可以构建出高效、可靠且低延迟的实时连续查询系统。这种组合非常适合需要实时监控、分析或响应事件的场景。

2. 集成 Kafka 作为 Flink 数据源

要使用 Kafka 作为 Flink 应用程序的连续数据源，需要利用 Flink 提供的 Kafka 连接器。该连接器能够可靠地从 Kafka 主题中消费数据，并将其转换为 Flink 的数据流（DataStream）。

2.1 添加必要的依赖

首先，在项目的 pom.xml 文件中添加 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版本调整 -->
    <scope>provided</scope>
</dependency>
<dependency>
    <groupId>org.apache.flink</groupId>
    <artifactId>flink-clients</artifactId>
    <version>1.17.1</version> <!-- 请根据实际使用的Flink版本调整 -->
    <scope>provided</scope>
</dependency>

登录后复制

2.2 构建 Kafka 数据源

使用 KafkaSource 构建器可以配置 Kafka 消费者。以下是一个简单的示例，演示如何从 Kafka 主题读取字符串消息：

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.environment.StreamExecutionEnvironment;

public class FlinkKafkaSourceExample {
    public static void main(String[] args) throws Exception {
        // 1. 获取 Flink 执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1); // 为简化示例，设置并行度为1

        // 2. 配置 KafkaSource
        KafkaSource<String> kafkaSource = KafkaSource.<String>builder()
                .setBootstrapServers("localhost:9092") // Kafka Broker 地址
                .setTopics("my-input-topic") // 要消费的 Kafka 主题
                .setGroupId("my-flink-consumer-group") // 消费者组ID
                .setStartingOffsets(OffsetsInitializer.earliest()) // 从最早的偏移量开始消费
                .setValueOnlyDeserializer(new SimpleStringSchema()) // 指定消息值的反序列化器
                .build();

        // 3. 将 KafkaSource 添加到 Flink 环境中，并打印接收到的消息
        env.fromSource(kafkaSource, WatermarkStrategy.noWatermarks(), "Kafka Source")
                .print("Kafka Message"); // 打印每个接收到的消息

        // 4. 执行 Flink 作业
        env.execute("Flink Kafka Source Connector Example");
    }
}

登录后复制

在上述代码中：

setBootstrapServers：指定 Kafka 集群的地址。
setTopics：指定要消费的一个或多个主题。
setGroupId：定义消费者组，用于管理消费偏移量。
setStartingOffsets：配置消费者启动时从哪个偏移量开始消费（例如，earliest() 从头开始，latest() 从最新消息开始）。
setValueOnlyDeserializer：指定如何反序列化 Kafka 消息的值。SimpleStringSchema 适用于简单的字符串消息。如果消息是更复杂的数据结构（如JSON、Avro），需要使用相应的反序列化器。
WatermarkStrategy.noWatermarks()：在这个基本示例中，我们暂不处理事件时间，因此使用无水印策略。在后续的窗口操作中，事件时间处理将变得重要。

3. 使用窗口处理实现连续查询

连续查询通常涉及对数据流进行聚合，例如计算在特定时间段内发生的事件数量、平均值或总和。Flink 的窗口（Window）机制是实现这一功能的强大工具。窗口可以将无限的数据流划分为有限的、可管理的“桶”，然后对这些桶中的数据进行计算。

3.1 窗口类型

Flink 支持多种窗口类型：

时间窗口（Time Windows）：基于时间划分数据，如每分钟、每小时。
- 滚动窗口（Tumbling Windows）：固定大小、不重叠的窗口。
- 滑动窗口（Sliding Windows）：固定大小、可以重叠的窗口。
- 会话窗口（Session Windows）：基于活动间隔划分，当一段时间没有新事件时窗口关闭。
计数窗口（Count Windows）：基于事件数量划分数据。

对于实时连续查询，时间窗口是最常用的。

Mootion

Mootion是一个革命性的3D动画创作平台，利用AI技术来简化和加速3D动画的制作过程。

177

查看详情

3.2 示例：基于处理时间的滚动窗口计数

为了简化入门，我们首先演示基于处理时间（Processing Time）的滚动窗口，它使用 Flink 机器的系统时间来划分窗口。

import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
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.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.assigners.TumblingProcessingTimeWindows; // 注意这里使用ProcessingTimeWindows
import org.apache.flink.streaming.api.windowing.time.Time;

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

        KafkaSource<String> kafkaSource = KafkaSource.<String>builder()
                .setBootstrapServers("localhost:9092")
                .setTopics("my-input-topic")
                .setGroupId("my-flink-consumer-group-window")
                .setStartingOffsets(OffsetsInitializer.earliest())
                .setValueOnlyDeserializer(new SimpleStringSchema())
                .build();

        env.fromSource(kafkaSource, WatermarkStrategy.noWatermarks(), "Kafka Source")
                .map(new MapFunction<String, Long>() {
                    @Override
                    public Long map(String value) throws Exception {
                        // 假设每条消息代表一个事件，我们将其转换为计数1
                        return 1L;
                    }
                })
                .windowAll(TumblingProcessingTimeWindows.of(Time.seconds(5))) // 每5秒一个滚动窗口，基于处理时间
                .reduce(new ReduceFunction<Long>() {
                    @Override
                    public Long reduce(Long a, Long b) throws Exception {
                        // 对窗口内的所有1进行求和，得到消息总数
                        return a + b;
                    }
                })
                .print("Window Count"); // 打印每个窗口的计数结果

        env.execute("Flink Kafka Processing Time Window Example");
    }
}

登录后复制

在这个例子中：

我们首先将每条 Kafka 消息映射为 1L，表示一个事件。
windowAll(TumblingProcessingTimeWindows.of(Time.seconds(5)))：将所有进入的数据流分配到每 5 秒一个的滚动窗口中，窗口的划分基于处理时间。windowAll 操作会把所有数据发送到一个并行度，在生产环境中应尽量避免，除非数据量很小或聚合逻辑必须全局进行。对于有键（keyed）的数据流，应使用 keyBy() 之后再 window()。
reduce()：对窗口内的所有 1L 进行累加，计算出该窗口内接收到的消息总数。

3.3 事件时间与水印（Watermarks）

在实际的流处理应用中，通常更倾向于使用事件时间（Event Time）而不是处理时间。事件时间是指事件实际发生在其源头设备上的时间。由于网络延迟、乱序到达等问题，事件时间处理需要更复杂的机制，即水印（Watermarks）。

水印是 Flink 用来衡量事件时间进度的特殊标记。它告诉 Flink “到目前为止，所有事件时间戳小于或等于此水印的事件都应该已经到达”。这使得 Flink 即使在乱序数据流中也能正确地进行窗口聚合。

使用事件时间的步骤：

定义数据结构并提取时间戳： Kafka 消息通常是字符串或字节数组，需要反序列化并从中提取事件时间戳。
配置 WatermarkStrategy： 告诉 Flink 如何生成水印以及如何从事件中提取时间戳。

import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.common.serialization.DeserializationSchema;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.connector.kafka.source.KafkaSource;
import org.apache.flink.connector.kafka.source.enumerator.initializer.OffsetsInitializer;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows; // 注意这里使用EventTimeWindows
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.util.Collector;

import java.io.IOException;
import java.time.Duration;

// 假设Kafka消息是简单的JSON字符串，包含一个"timestamp"字段和一个"value"字段
class Event {
    public long timestamp; // 事件时间戳，毫秒
    public String value;

    public Event() {}
    public Event(long timestamp, String value) {
        this.timestamp = timestamp;
        this.value = value;
    }

    @Override
    public String toString() {
        return "Event{" + "timestamp=" + timestamp + ", value='" + value + '\'' + '}';
    }
}

// 自定义反序列化器，用于将JSON字符串转换为Event对象
class EventDeserializer implements DeserializationSchema<Event> {
    @Override
    public Event deserialize(byte[] message) throws IOException {
        // 实际应用中会使用Jackson或Gson进行JSON解析
        String jsonString = new String(message);
        // 简单模拟解析，假设格式为 {"timestamp":1678886400000,"value":"test"}
        try {
            long ts = Long.parseLong(jsonString.substring(jsonString.indexOf("timestamp\":") + 11, jsonString.indexOf(",\"value\"")));
            String val = jsonString.substring(jsonString.indexOf("value\":\"") + 8, jsonString.lastIndexOf("\"}"));
            return new Event(ts, val);
        } catch (Exception e) {
            System.err.println("Failed to parse event: " + jsonString + ", error: " + e.getMessage());
            return new Event(System.currentTimeMillis(), "parse_error"); // 错误处理
        }
    }

    @Override
    public boolean is='true';
    public TypeInformation<Event> get='producedType' {
        return TypeInformation.of(Event.class);
    }
}


public class FlinkKafkaEventTimeWindowExample {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        // 设置事件时间语义
        env.getConfig().setAutoWatermarkInterval(1000); // 每秒生成一次水印

        KafkaSource<Event> kafkaSource = KafkaSource.<Event>builder()
                .setBootstrapServers("localhost:9092")
                .setTopics("my-input-topic")
                .setGroupId("my-flink-consumer-group-eventtime")
                .setStartingOffsets(OffsetsInitializer.earliest())
                .setValueOnlyDeserializer(new EventDeserializer()) // 使用自定义反序列化器
                .build();

        env.fromSource(kafkaSource,
                // 配置水印策略：允许5秒的乱序，并从Event对象的timestamp字段提取事件时间
                WatermarkStrategy.<Event>forBoundedOutOfOrderness(Duration.ofSeconds(5))
                        .withTimestampAssigner((event, recordTimestamp) -> event.timestamp),
                "Kafka Event Source")
                .map(new MapFunction<Event, Long>() {
                    @Override
                    public Long map(Event value) throws Exception {
                        return 1L; // 计数
                    }
                })
                .windowAll(TumblingEventTimeWindows.of(Time.seconds(10))) // 每10秒一个滚动窗口，基于事件时间
                .reduce(new ReduceFunction<Long>() {
                    @Override
                    public Long reduce(Long a, Long b) throws Exception {
                        return a + b;
                    }
                })
                .print("Event Time Window Count");

        env.execute("Flink Kafka Event Time Window Example");
    }
}

登录后复制

注意： 上述 EventDeserializer 中的 JSON 解析是一个非常简化的示例，仅用于演示概念。在生产环境中，应使用成熟的 JSON 库（如 Jackson 或 Gson）进行健壮的解析。

在这个事件时间窗口的例子中：

我们定义了一个 Event 类来表示包含时间戳和值的事件。
EventDeserializer 负责将 Kafka 消息（假设是 JSON 字符串）反序列化为 Event 对象。
WatermarkStrategy.forBoundedOutOfOrderness(Duration.ofSeconds(5))：配置了一个允许 5 秒乱序的水印策略。这意味着 Flink 会等待 5 秒，以确保大多数乱序事件都能到达。
.withTimestampAssigner((event, recordTimestamp) -> event.timestamp)：指定 Flink 从 Event 对象的 timestamp 字段中提取事件时间戳。
TumblingEventTimeWindows.of(Time.seconds(10))：现在窗口是基于事件时间进行划分的，每 10 秒一个滚动窗口。