4、实时计算引擎：Apache Flink核心概念、时间语义与Watermark、状态管理与容错

好，咱们进入实时推荐平台最核心的引擎层——Apache Flink。

说实话，在 Flink 出现之前，做实时流处理是一件挺痛苦的事。Storm 延迟低但吞吐上不去，Spark Streaming 本质上是微批，延迟做不到纯实时。直到 Flink 把「真正的流处理」带到工业界，我才觉得实时推荐这事终于有了靠谱的底座。

这一章我会把 Flink 最关键的三个概念讲透：核心抽象、时间语义与 Watermark、状态管理与容错。这些都是你写生产级 Flink 作业必须理解的东西。

4.1 Flink 核心概念：DataStream 与算子

Flink 把一切数据都看作流。你写代码时面对的核心抽象就是 DataStream。

说白了，DataStream 就是一个可能无限大的、带时间戳的事件序列。你在上面做 map、filter、keyBy、window 这些操作，Flink 会帮你把算子分布到集群的 TaskManager 上去并行执行。

我刚开始用 Flink 时犯过一个低级错误：在 map 函数里创建了数据库连接，结果每个元素都 new 一次连接，直接把数据库打挂了。后来才意识到，应该用 RichMapFunction 的 open() 方法来做初始化。

// 错误示范：每个元素都创建连接
dataStream.map(new MapFunction<Event, String>() {
    @Override
    public String map(Event value) throws Exception {
        Connection conn = DriverManager.getConnection(url); // 千万别这么干！
        // ...
    }
});

// 正确做法：使用 RichFunction
dataStream.map(new RichMapFunction<Event, String>() {
    private transient Connection conn;

    @Override
    public void open(Configuration parameters) throws Exception {
        conn = DriverManager.getConnection(url); // 只初始化一次
    }

    @Override
    public String map(Event value) throws Exception {
        // 复用 conn
    }
});

4.2 时间语义：Event Time、Ingestion Time、Processing Time

时间在流处理里是个绕不开的话题。Flink 支持三种时间语义，我直接给你说人话：

时间语义	含义	推荐场景
Event Time	事件实际发生的时间（嵌入在数据里）	推荐日志、交易记录、IoT 传感器
Ingestion Time	数据进入 Flink 的时间	数据本身没时间戳，但需要有序处理
Processing Time	算子本地时钟时间	实时监控、简单聚合、对延迟不敏感

做实时推荐，我几乎只用 Event Time。为什么？因为用户的行为日志从客户端发出，经过网络、消息队列、可能还有重试，到达 Flink 时可能已经乱序了。如果你用 Processing Time，那窗口统计就完全取决于数据到达的时机，结果根本不准。

举个例子：用户 A 在 10:00:01 点击了商品，10:00:03 又点了一次。但因为网络抖动，第二条日志先到了 Flink。如果你用 Processing Time 开 5 秒窗口，这两条可能被分到不同窗口，推荐特征就错了。

4.3 Watermark：乱序世界的时钟

Event Time 虽好，但有个问题：Flink 怎么知道某个时间戳的数据都到齐了？它没法无限等下去。这时候就需要 Watermark。

Watermark 本质上是一个「时间进度标记」。它告诉 Flink：当前时间戳小于 Watermark 的数据，应该已经全部到达了。如果还有迟到的数据，Flink 默认会丢弃，或者你可以用 allowedLateness 来兜底。

我个人习惯把 Watermark 理解成「水位线」。水位线涨到窗口的结束时间时，窗口就触发计算。

// 生产级 Watermark 生成策略
DataStream<Event> streamWithWatermarks = inputStream
    .assignTimestampsAndWatermarks(
        WatermarkStrategy.<Event>forBoundedOutOfOrderness(Duration.ofSeconds(5))
            .withTimestampAssigner((event, timestamp) -> event.getEventTime())
    );

这段代码的意思是：允许最多 5 秒的乱序。Watermark = 当前收到的最大事件时间 - 5 秒。你想想看，如果网络抖动导致数据晚到 3 秒，这个策略完全能兜住。

下面这张图展示了 Watermark 如何驱动窗口计算：

你看，Watermark 涨到 10:07 时，窗口 [10:00, 10:05) 就触发了。e5 虽然来晚了，但如果设置了 allowedLateness，它还能被归到正确的窗口里。

4.4 状态管理与容错：Checkpoint 与 Savepoint

Flink 跟普通流处理框架最大的区别是什么？它有状态。

你在窗口里做的聚合、你在 KeyedProcessFunction 里维护的计数器、你在 RichMapFunction 里缓存的数据——这些都是状态。Flink 把状态分为两类：

• Operator State：绑定到算子的并行实例上，比如 Kafka 的 offset。
• Keyed State：按 key 分区，每个 key 有自己的状态。最常用的是 ValueState、ListState、MapState。

状态这么重要，万一机器挂了怎么办？Flink 的容错机制靠的是 Checkpoint。

Checkpoint 的原理其实不复杂：JobManager 定期往数据流里插入一个「屏障」（Barrier）。每个算子收到屏障后，把当前状态快照保存到持久化存储（比如 HDFS）。如果任务失败，Flink 从最近一次成功的 Checkpoint 恢复所有算子的状态，然后重新消费数据。

// 开启 Checkpoint 的配置
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
env.enableCheckpointing(60000); // 每 60 秒做一次 Checkpoint
env.getCheckpointConfig().setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);
env.getCheckpointConfig().setMinPauseBetweenCheckpoints(30000);
env.getCheckpointConfig().setCheckpointTimeout(120000);
env.getCheckpointConfig().setMaxConcurrentCheckpoints(1);

这里有几个参数我解释一下：

参数	值	说明
enableCheckpointing	60000	Checkpoint 间隔。太频繁影响性能，太慢恢复时间变长
setMinPauseBetweenCheckpoints	30000	上一个 Checkpoint 完成后，至少等 30 秒才做下一个
setCheckpointTimeout	120000	如果 Checkpoint 超过 2 分钟没完成，就丢弃
setMaxConcurrentCheckpoints	1	同一时间只允许一个 Checkpoint 在进行

除了 Checkpoint，还有 Savepoint。它本质上是手动触发的 Checkpoint，通常用于作业升级、代码变更、集群迁移。你可以理解成「数据库的备份」，而 Checkpoint 是「数据库的 WAL 日志」。Savepoint 需要你手动触发，而且会保留更长时间。

# 手动触发 Savepoint
flink savepoint <jobId> [targetDirectory]

# 从 Savepoint 恢复
flink run -s <savepointPath> -c MainClass myJob.jar

嗯，这里有个小技巧：升级作业时，如果状态结构变了（比如新增了一个 ValueState），Flink 默认会报错。你需要用 State TTL 或者自定义 StateSerializer 来做兼容。我一般会在开发环境先跑一遍从 Savepoint 恢复的流程，确认没问题再上线。

4.5 状态后端：内存 vs RocksDB

状态存哪里？Flink 提供了几种状态后端：

• MemoryStateBackend：状态存在 TaskManager 的 JVM 堆内存里。适合开发调试，生产环境别用。
• FsStateBackend：状态存在 TaskManager 内存，Checkpoint 写到文件系统。适合状态量不大的场景。
• RocksDBStateBackend：状态存在 RocksDB（本地磁盘），Checkpoint 写到文件系统。适合超大状态。

做实时推荐，用户画像和特征数据通常很大，我几乎只用 RocksDBStateBackend。它把状态序列化后写到磁盘，内存只做缓存，所以能支撑 TB 级别的状态。

// 生产环境推荐配置
env.setStateBackend(new RocksDBStateBackend("hdfs:///flink/checkpoints", true));

第二个参数 true 表示开启增量 Checkpoint。RocksDB 的增量 Checkpoint 只上传变化的部分，而不是全量快照。对于大状态作业，这个优化能省下 90% 的 Checkpoint 时间。