但是查看正式目录下相关文件的信息，我们发现 task 2、3 两个 task 并没有 Checkpoint 4608 的文件（文件名含有 task id 和 Checkpoint id 信息，所以可以根据正式目录下的文件名知道其是哪个 task 在哪个 Checkpoint 期间创建的）。故初步确定的原因是某些文件被误删造成数据丢失。Task 2/3/6/7 在文件删除后由于没有文件的写入和关闭操作，task 正常运行；而 task 0/1/4/5 在文件删除后还有文件的写入和关闭操作，造成 task 失败。 

下一步就要去排查文件丢失的原因。我们通过 HDFS trace 记录表（ HDFS trace记录表记录着用户和系统调用行为，以达到分析和运维的目的）查看 task 2 Checkpoint 4608 临时目录操作记录，对应的路径为/xx/_DUMP_TEMPORARY/cp-4608/task-2。

为了解决这个问题，我们首先想到的是 HDFS 保证删除操作的幂等性，这样即使删除操作重复执行也不会影响后续写入的问题，进而可以保证数据的准确性。但是咨询 HDFS 后，HDFS 表示 HDFS在现有架构下无法保证删除的幂等性。

参考 DDIA (Designing Data-Intensive Applications) 第 9 章中关于因果关系的定义：因果关系对事件施加了一种顺序——因在果之前。对应于MQ dump 流程中删除操作是因，发生在写入数据之前。我们需要保证这两个关系的因果关系。而根据其解决因果问题的方法，一种解决思路是 HDFS 在每个client 请求中都带上序列号顺序，进而在HDFS NameNode 上可以保证单个client的请求因果性。跟HDFS 讨论后发现这个方案的实现成本会比较大。

了解 HDFS 难以保证操作的幂等性后，我们想是否可以将写入前的删除操作去除，也就是说在写入 HDFS 之前不清理文件夹而是直接写入数据到文件，这样就不需要有因果性的保证。

如果我们知道临时文件夹中哪些文件是我们需要的，在重命名阶段就可以直接将需要的文件重命名到正式目录而忽略临时文件夹中的脏文件，这样在写入之前就不需要删除文件夹。故我们的解决方案是将写入的文件路径存储到 Flink state 中，从而确保在 commit 阶段以及恢复阶段可以将需要的文件移动到正式目录。

最终，我们选择了方案二解决该问题，使用文件 state 前后处理流程对比如下图所示：

目前文件 state 已经在线上使用了，下面先介绍一下实现中碰到的相关问题，然后再描述一下上线后效果。

可观测性

• 创建文件的数量：state 中所有文件的数量，也就是当前 Checkpoint 处理数据阶段创建的所有文件数量。

• 重命名成功文件的数量：NotifyCheckpointComplete 阶段将临时文件成功移动到正式目录下的文件数量。

• 忽略重命名文件的数量：NotifyCheckpointComplete 阶段忽略移动到正式目录下下的文件数量。也就是临时文件夹中不存在但是正式目录存在的文件。这种情况通常发生在任务有 Failover 的情况。Failover 后任务从 Checkpoint 中恢复，失败前已经重命名成功的文件在当前阶段会忽略重命名。

• 重命名失败的文件数量：临时目录以及正式目录下都不存在文件的数量。这种情况通常是由于任务发生了异常造成数据的丢失。目前线上比较常见的一个 case 是任务在关闭一段时间后再开启。由于 HDFS TTL 的设置小于任务关闭的时长，临时目录中写入的文件被 HDFS TTL 策略清除。这个结果实际是符合预期的。

前向兼容性

• 第一期写入数据前保留了删除操作

• 第二期删除了写入数据前的删除操作

• HDFS 集群正常切主

• HDFS 集群主节点失败超过10分钟

字节跳动流式数据集成仍在不断发展中，未来主要关注以下几方面：

1. 功能增强，增加简单的数据转换逻辑，缩短流式数据处理链路，进而减少处理时延

2. 架构升级，离线集成和实时数据集成架构统一

3. 支持 auto scaling 功能，在业务高峰和低峰自动扩缩容，提高资源利用率，减少资源浪费

本文中介绍的《字节跳动流式数据集成基于Flink Checkpoint两阶段提交的实践和优化》，目前已通过火山引擎数据产品大数据研发治理套件 DataLeap 向。

• 字节跳动基于Flink的MQ-Hive实时数据集成

• 字节跳动单点恢复功能及 Regional CheckPoint 优化实践

• Designing Data-Intensive Applications
• Stateful Stream Processing