📚 课程目录第3章 Flink核心概念

Flink核心概念：从API到时间语义再到状态管理

说实话，Flink 的核心概念说多不多，说少不少。但如果你想把流批一体真正落地，有几个东西是绕不过去的——DataStream API 和 Table API 怎么选、事件时间和处理时间到底差在哪、Watermark 机制到底怎么玩、状态后端又该怎么配。这一章，我就把这些核心概念掰开揉碎讲清楚。

3.1 DataStream API 与 Table API：两条路，各有千秋

Flink 提供了两套主流 API：DataStream API 和 Table API / SQL。很多人刚接触时会纠结选哪个。我的建议是——别纠结，看场景。

3.1.1 DataStream API：底层控制力

DataStream API 是 Flink 的“亲儿子”。它提供了对数据流的精细控制，你能直接操作每个元素、每个状态、每个时间戳。适合做复杂事件处理、自定义算子、需要精细调优的场景。

// 一个简单的 DataStream 示例
DataStream<String> source = env.addSource(new FlinkKafkaConsumer<>(...));
DataStream<Event> parsed = source
    .map(line -> JsonUtil.parse(line, Event.class))
    .keyBy(Event::getUserId)
    .window(TumblingEventTimeWindows.of(Time.hours(1)))
    .process(new MyRichProcessFunction());

我在项目中遇到过几次，用 DataStream API 处理复杂的乱序数据流，Watermark 和状态管理都能精确控制。但代价是代码量偏大，开发效率不如 SQL。

3.1.2 Table API / SQL：声明式高效

Table API 和 SQL 是 Flink 面向分析场景的“快车道”。你只需要声明“我要什么”，Flink 帮你优化执行计划。对于 ETL、报表、批处理场景，SQL 的简洁性无可替代。

// Table API 示例
Table orders = tableEnv.fromDataStream(orderStream, $("id"), $("amount"), $("ts").rowtime());
Table result = orders
    .window(Tumble.over(lit(1).hour()).on($("ts")).as("w"))
    .groupBy($("w"))
    .select($("w").start(), $("amount").avg());

我的经验： 如果团队里 SQL 能力强的同学多，优先用 Table API。但遇到复杂的状态逻辑或自定义窗口，还是得回到 DataStream API。两条路不是互斥的，Flink 允许你混合使用。

3.1.3 如何选择？一张表说清楚

维度	DataStream API	Table API / SQL
控制粒度	细，可操作每个元素	粗，基于关系代数
开发效率	低，代码量大	高，声明式
状态管理	灵活，自定义状态	内置，有限定制
复杂事件处理	强，CEP 库支持	弱，需 UDF
批处理兼容	需额外处理	天然统一
适用场景	实时风控、规则引擎	实时报表、ETL

3.2 事件时间与处理时间：时间语义的选择

时间语义是流计算的灵魂。Flink 支持三种时间：事件时间（Event Time）、处理时间（Processing Time）、摄入时间（Ingestion Time）。实际项目中，最常用的是前两种。

3.2.1 处理时间：简单但不可靠

处理时间就是机器当前的时间。你想想看，如果数据从 Kafka 消费到 Flink 处理，中间有网络延迟、背压、重启，处理时间跟事件实际发生的时间可能差很远。适合对时效性要求不高、允许结果有偏差的场景。

避坑： 我曾经在一个实时大屏项目里用了处理时间，结果数据源短暂故障恢复后，所有窗口都按恢复时刻的时间重新计算，大屏数据直接“穿越”了。从那以后，只要涉及业务统计，我坚决用事件时间。

3.2.2 事件时间：真实但复杂

事件时间是数据本身携带的时间戳，比如用户点击页面的时间、传感器采集的时间。它不受系统时钟影响，能真实反映业务发生顺序。但代价是——数据可能乱序、迟到，你需要 Watermark 机制来处理。

// 设置事件时间
env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);

// 提取时间戳并生成 Watermark
DataStream<Event> withTimestamps = stream
    .assignTimestampsAndWatermarks(
        WatermarkStrategy.<Event>forBoundedOutOfOrderness(Duration.ofSeconds(5))
            .withTimestampAssigner((event, timestamp) -> event.getTimestamp())
    );

3.3 Watermark 机制：乱序数据的“交通警察”

Watermark 是 Flink 处理乱序数据的核心机制。说白了，它就是一个“时间截止线”——告诉 Flink：“在这个时间戳之前的数据，我基本都收到了，你可以触发窗口计算了。”

3.3.1 Watermark 的工作原理

Watermark 是一个单调递增的时间戳。当 Watermark 达到窗口的结束时间时，窗口就会被触发。比如一个 10:00-11:00 的窗口，当 Watermark 达到 11:00 时，窗口开始计算。

为什么会需要 Watermark？因为数据可能迟到。比如事件发生在 10:59，但网络延迟导致 11:02 才到。如果没有 Watermark，窗口在 11:00 就关闭了，这条数据就丢了。Watermark 允许你设置一个“等待时间”，比如 5 秒，那么 Watermark 会延迟到 11:00:05 才触发窗口。

核心公式： Watermark = 当前最大事件时间 - 最大允许延迟时间

3.3.2 三种 Watermark 生成策略

策略	说明	适用场景
forMonotonousTimestamps	假设数据严格有序，Watermark 等于当前时间戳	数据源本身有序（如某些 IoT 设备）
forBoundedOutOfOrderness	允许一定范围的乱序，设置固定延迟	大部分生产环境，延迟可控
自定义 WatermarkStrategy	完全自定义生成逻辑，比如基于业务规则	特殊场景，如根据用户等级动态调整延迟

// 自定义 Watermark 生成器
public class MyWatermarkGenerator implements WatermarkGenerator<Event> {
    private long maxTimestamp = Long.MIN_VALUE;
    private final long maxDelay = 5000; // 5秒延迟

    @Override
    public void onEvent(Event event, long eventTimestamp, WatermarkOutput output) {
        maxTimestamp = Math.max(maxTimestamp, eventTimestamp);
    }

    @Override
    public void onPeriodicEmit(WatermarkOutput output) {
        output.emitWatermark(new Watermark(maxTimestamp - maxDelay));
    }
}

个人习惯： 我一般设置延迟时间为业务可接受的最大等待时间。比如实时大屏可以接受 10 秒延迟，我就设 10 秒。别设太大，否则窗口迟迟不触发，内存压力会很大。

3.4 状态后端：状态存哪、怎么存

Flink 的状态后端决定了状态数据存哪里、怎么存、怎么容错。选对了，性能翻倍；选错了，OOM 等着你。

3.4.1 三种状态后端对比

状态后端	存储位置	适用场景	优缺点
MemoryStateBackend	TaskManager 内存 + JobManager 堆内存（Checkpoint）	开发测试、小状态	快但容量小，易 OOM
FsStateBackend	TaskManager 内存 + 外部文件系统（Checkpoint）	生产环境，中等状态	容量大，但恢复时需从文件系统加载
RocksDBStateBackend	RocksDB（本地磁盘） + 外部文件系统（Checkpoint）	超大状态（TB 级）	支持增量 Checkpoint，但读写性能略低于内存

3.4.2 我如何选型

嗯，这里直接说结论：

状态小于 1GB： FsStateBackend 就够了，简单高效。
状态在 1GB 到 100GB： RocksDBStateBackend 是首选，增量 Checkpoint 能大幅减少快照时间。
状态超过 100GB： 必须用 RocksDBStateBackend，并且要调优 RocksDB 的配置，比如 block cache 大小、写缓冲等。

// 配置 RocksDB 状态后端
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
env.setStateBackend(new RocksDBStateBackend("hdfs://namenode:8020/flink/checkpoints", true));

// 增量 Checkpoint 配置
env.getCheckpointConfig().enableExternalizedCheckpoints(
    CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION
);

我曾经踩过的坑： 有次把 MemoryStateBackend 用在了生产环境，状态大小才 500MB，看起来没问题。结果一次大促流量暴增，状态膨胀到 3GB，TaskManager 直接 OOM 挂掉。从那以后，生产环境我至少用 FsStateBackend，大状态必上 RocksDB。

3.5 本章知识体系总览

下面这张图把本章的核心概念串起来了，方便你对照理解。

这张图把 API、时间语义、状态管理三条线串起来了。你从上往下看，Flink 在最顶层，下面分三大块。API 层分 DataStream 和 Table API；时间语义分事件时间和处理时间，而 Watermark 是连接事件时间和窗口触发的桥梁；状态后端则决定了状态存哪里、怎么存。三者共同构成了 Flink 流批一体的核心骨架。