如何构建批流一体数据融合平台的一致性语义保证？(15)_：如何构建批流一体数据融合平台的一致

第二，一致性处理

事务性消息最早就是为 Kafka Streams 设计和准备的。可以写一段 Kafka Streams 应用，从 Kafka 里读取数据，然后完成转化逻辑，进而将结果再输出回 Kafka。Sink 端再从 Kafka 中消费数据，写入目的地。

数据一致性写入

之前简要谈了一下二阶段提交协议的原理，DataPipeline 实现的方式不算很深奥，基本是业界的一种统一方式。其中最核心的点是，我们将 consumer offset 管理从 Kafka Connect 框架中独立出来，实现事务一致性提交。另外，在 Sink 端封装了一个类似于 Flink 的 TwoPhaseCommitSinkFunction 方式，其定义了 Sink 若要实现一个二阶段提交所必须要实现的一些功能。

DataPipeline 将 Sink Connector 分为两类，一类是 Connector 本身具备了事务能力，比如绝大部分的关系型数据库，只需将 offset 跟数据同时持久化到目的地即可。额外的可能需要有一张 offset 表来记录提交的 offset。还有一类 Sink 不具备事务性能力，类似像 FTP、OSS 这些对象存储，我们需要去实现一个二阶段提交协议，最终才能保证 Sink 端的数据能够达到一致性写入。