字节跳动流式数据集成基于Flink Checkpoint两阶段提交的实践和优化（1）

目前字节跳动中国区 MQ dump 例行任务数巨大，日均处理流量在 PB 量级。巨大的任务量和数据量对 MQ dump 的稳定性以及准确性带来了极大的挑战。

本文主要介绍 DTS MQ dump 在极端场景中遇到的数据丢失问题的排查与优化，最后介绍了上线效果。本文分两次连载，第一篇主要介绍Flink Checkpoint 以及 MQ dump 写入流程。

HDFS 集群某个元数据节点由于硬件故障宕机。在该元数据节点终止半小时后，HDFS 手动运维操作将 HDFS 切主到 backup 节点后，HDFS 恢复服务。故障恢复后用户反馈 MQ dump 在故障期间有数据丢失，产出的数据与 MQ 中的数据不一致。
收到反馈后我们立即进行故障的排查。下面先简要介绍一下 Flink Checkpoint 以及 MQ dump 写入流程。

Flink Checkpoint 简介

Flink 基于 Chandy-Lamport 分布式快照算法实现了 Checkpoint 机制，能够提供 Exactly Once 或者 At Least Once 语义。

Flink 通过在数据流中注入 barriers 将数据拆分为一段一段的数据，在不终止数据流处理的前提下，让每个节点可以独立创建 Checkpoint 保存自己的快照。每个 barrier 都有一个快照 ID ，在该快照 ID 之前的数据都会进入这个快照，而之后的数据会进入下一个快照。

Checkpoint 对 Operator state 进行快照的流程可分为两个阶段：

Snapshot state 阶段：对应 2PC 准备阶段。Checkpoint Coordinator 将 barries 注入到 Source Operator 中。Operator 接收到输入 Operator 所有并发的 barries 后将当前的状态写入到 state 中，并将 barries 传递到下一个 Operator。
Notify Checkpoint 完成阶段：对应 2PC 的 commit 阶段。Checkpoint Coordinator 收到 Sink Operator 的所有 Checkpoint 的完成信号后，会给 Operator 发送 Notify 信号。Operator 收到信号以后会调用相应的函数进行 Notify 的操作。

而在任务失败后，任务会从上一个 Checkpoint state 中进行恢复，进而实现 Exactly Once 或者 At Least Once 语义。

MQ dump 写入流程梳理

MQ dump 利用 Flink Checkpoint 机制和 2PC（Two-phase Commit）机制实现了 Exactly Once 语义，数据可以做到不重不丢。

根据 Flink Checkpoint 的流程，MQ dump 整个写入过程可以分为四个不同的流程：

数据写入阶段
SnapshotState 阶段
Notify Checkpoint 完成阶段
Checkpoint 恢复阶段

整个流程可以用下面的流程图表示：

详细介绍上面各个阶段的主要操作。假设 Flink 任务当前 Checkpoint id 为 n，当前任务的 task id 为x。

数据写入阶段

写入阶段就主要有以下两个操作：

如果是当前 Checkpoint 第一次写入（transaction），先清理要写入临时文件夹 /tmp/cp-n/task-x
在临时文件夹中建立文件并写入数据
注意在写入数据之前我们会先清理临时目录。执行这个操作的原因是我们需要保证最终数据的准确性：

假设任务 x 在 Checkpoint n 写入阶段失败了（将部分数据写入到临时文件夹/tmp/cp-n/task-x），那么任务会从上一个 Checkpoint n-1 恢复，下一个写入的 Checkpoint id 仍然为 n。如果写入前不清理临时目录，失败前遗留的部分脏文件就会保留，在 Checkpoint 阶段就会将脏文件移到正式目录中。