搞实时数据的人应该都遇到过这个灵魂拷问Flink的“精确一次Exactly-Once Semantics”到底是不是真的能兑现到下游数据库我最近正好完整做了个验证项目标题就叫“Flink EOS--整合MySQL验证2PC”把一条Kafka到MySQL的数据链路用Flink的两阶段提交协议完整跑了一遍包括正常消费、故障重启、事务回滚三种场景。结论是EOS确实能落库但它不是白给的需要Flink的Checkpoint机制、MySQL的XA事务、Sink端的提交逻辑三者严丝合缝地配合。这篇文章就把整个验证过程拆开讲先从原理说起然后是链路和表怎么设计、核心代码怎么写、我怎么设计实验去证明“不重不丢”最后是踩过的几个坑。适合正在搞实时数仓、想把Flink结果可靠落库的工程师参考也适合那些对“Flink端到端精确一次”半信半疑、想自己动手验证一把的同学。1. 先把“精确一次”这件事掰开揉碎1.1 从Checkpoint到端到端一致为什么不能只靠状态后端很多人以为Flink开个Checkpoint就算精确一次了这其实只覆盖了Flink内部的状态。Checkpoint的本质是分布式快照它通过Barrier对齐把每个算子Operator的State冻结下来存储在状态后端里。当作业故障恢复时Flink从最近一次成功的Checkpoint恢复State同时把Kafka的消费位点也恢复到那次Checkpoint对应的位置于是Source端重新消费这段数据整个DAG重新计算一遍。这套机制保证了“Flink自己内部的状态”不会重复累加也不丢数据。但问题出在下游。Flink把计算结果写到MySQL时用的是外部事务。如果Flink在写完一批数据、还没提交事务时挂了作业恢复后会从Kafka重放这一批数据。Sink函数收到重放的数据会再次执行INSERT如果上一次那批数据已经提交了就重复写入如果没有提交又会再写一遍。也就是说Checkpoint只能保证计算状态的一致性没法让MySQL一起“回到过去”。这就是端到端一致性的难点——下游系统必须参与到这个恢复过程中来。解决思路其实很朴素让Flink和MySQL在同一个“分布式事务”里协同提交通过两阶段提交2PC协议把“预提交”和“正式提交”拆开。这样Flink故障恢复时能够知道哪些事务已经预提交但还未正式提交从而要么提交要么回滚而不是盲目重放。1.2 2PC在Flink里的落地形态Flink内部把两阶段提交抽象成了一个接口类叫TwoPhaseCommitSinkFunction。这个类把Sink的生命周期拆成了五个阶段beginTransaction、preCommit、commit、abort以及两个恢复用的回调recoverAndCommit、recoverAndAbort。我用自己的话翻译一下这套流程beginTransaction在每条Checkpoint屏障到达Sink之前开启一个事务拿到一个事务句柄。preCommit当Checkpoint屏障穿过Sink时Sink会停止写入新数据先把自己缓存的这一批数据写入数据库但不提交只做“预提交”。在MySQL的场景下这一步对应XA PREPARE。commit等Checkpoint真正完成所有算子都确认快照成功后再通知数据库执行XA COMMIT这批数据才算真正生效。abort如果Checkpoint失败或者作业恢复后发现某个事务没预提交成功就把事务回滚。这样设计有一个精妙的地方Flink的提交动作以Checkpoint为粒度。一次Checkpoint对应一个事务也就是说“一批数据”和“一个事务”是对齐的。如果作业挂了Flink恢复后状态里有每个未完成事务的句柄它可以根据事务在MySQL里的实际状态来决定是继续提交还是回滚从而避免重放数据导致的重复写入。MySQL这边对应的支撑能力是XA事务。MySQL从5.7开始对XA的支持已经比较完整它本身就实现了两阶段提交协议的事务原语包括XA START、XA END、XA PREPARE、XA COMMIT、XA ROLLBACK。Flink的JDBC连接器底层的XAResource就是通过JDBC驱动把这些XA语句发给MySQL执行的。这个配合关系是整个验证项目的核心地基。搞懂了这个后面看代码就不会觉得抽象了。2. 验证项目设计环境、表结构、链路怎么搭2.1 整体链路与数据流设计我搭的验证链路是这样的Kafka作为输入源Flink消费Kafka里的订单事件做一次简单的聚合清洗然后写入MySQL的两张表。整体数据流可以用一条线描述Kafka订单事件 → Flink状态计算 → MySQL订单聚合表。为了让验证有意义我故意把计算逻辑设计成“非幂等”的如果重复执行结果就会变多。具体做法是把订单事件按用户ID聚合成消费金额累加值每次事件过来统计值就增加对应金额。这种累加逻辑天然对重复敏感如果Flink的EOS没生效同一批订单被重复消费时聚合金额会明显翻倍一眼就能看出来。Kafka Topic设置成3个分区Flink的并行度也对应设置这样每个并行子任务都能对应一个Kafka分区方便对齐Barrier也方便观察数据分发。2.2 表设计与语义约定MySQL这边我建了两张表一张是订单明细表一张是聚合统计表。订单明细表t_order_event存原始事件主键是订单ID。这张表的目的是记录Flink到底往MySQL写入了哪些数据用来做对账。实际上为了测试Flink的EOS下推写入的核心表是聚合统计表t_order_stat它的主键是用户ID字段包括总金额和总笔数。表结构大致如下CREATE TABLE t_order_event ( order_id BIGINT PRIMARY KEY, user_id BIGINT NOT NULL, amount DECIMAL(12,2) NOT NULL, event_time DATETIME NOT NULL, source_ts TIMESTAMP NOT NULL ) ENGINEInnoDB; CREATE TABLE t_order_stat ( user_id BIGINT PRIMARY KEY, total_amount DECIMAL(14,2) NOT NULL, order_count BIGINT NOT NULL, update_time DATETIME NOT NULL ) ENGINEInnoDB;核心语义是把同一user_id的订单金额累加到t_order_stat.total_amount上。这个聚合结果如果在故障恢复后重复累加total_amount会虚高合同对不上就是铁证。选择用户ID作为主键还有一个好处是方便后续用INSERT INTO ... ON DUPLICATE KEY UPDATE这种幂等写入方式做对照组——虽然2PC不用依赖这个但可以作为对比实验。2.3 组件选型为什么是Kafka MySQL而不是其他用Kafka做Source是因为它天然适合验证EOSKafka自带消费位点配合Flink Checkpoint可以精确恢复消费位置。如果Source换成Socket或者普通消息队列位点恢复就没这么方便验证结论会打折扣。用MySQL做Sink是因为MySQL的XA事务是验证2PC最好的“试验田”。它成熟、稳定、可观测能通过information_schema里的XA事务状态来确认事务是否预提交成功。而且JDBC驱动和Flink连接器都有现成的XA支持不需要自己去实现分布式事务协议。这里我提一个重要判断如果你想验证“Flink 2PC这个机制本身”不要选那些不支持XA或者不暴露事务状态的存储。我之前试过用某个纯KV型数据库做Sink它只支持单行事务根本不支持跨语句的预提交最后只能自己模拟一套事务标记验证出来的东西说服力不强。MySQL的XA支持是关系型数据库里最容易观察的。我在环境准备上用的版本是Flink 1.17版本MySQL 8.0Kafka 3.4JDBC驱动用的MySQL Connector/J 8.0.33。这几个版本之间没有兼容性大坑如果有人想复现直接用这套组合最省心。3. 核心代码实现从Source到2PC Sink3.1 Source端Kafka消费要同步配合Flink的CheckpointSource端我用的是Flink的KafkaSource API而不是老的FlinkKafkaConsumer。关键点在于关闭Kafka自身的自动提交让位点完全交给Flink Checkpoint管理。如果Kafka这边自己提交位点一旦Flink作业从Checkpoint恢复并重置位点到更早的位置Kafka的位点和Flink的位点就会不一致后续对账会出问题。KafkaSourceOrderEvent source KafkaSource.OrderEventbuilder() .setBootstrapServers(localhost:9092) .setTopics(order_topic) .setGroupId(flink-eos-verify) .setStartingOffsets(OffsetsInitializer.latest()) .setValueOnlyDeserializer(new OrderEventDeserialization()) // 关闭自动提交位点交给Flink的Checkpoint .setProperty(enable.auto.commit, false) .build();Source端还有一个细节deserialization的水位线时间戳。我在解析订单事件时把Kafka的消息时间戳也透传到了记录里方便后面做延迟统计。这个字段在处理常见故障时不参与一致性判断但能帮我确认“同一批数据是否被重放”。重放的数据在source_ts上会有明显的时间特征。3.2 核心一用JdbcExactlyOnceSink直接落MySQL如果不想自己从零实现2PCFlink新版本的JDBC连接器里已经有现成的JdbcExactlyOnceSink它内部就是基于TwoPhaseCommitSinkFunction和XA协议封装的。我第一版验证就是先用它把链路跑通确认整个机制是通的再去做自定义实现。用法比较直接JdbcExactlyOnceSinkOrderAgg sink JdbcExactlyOnceSink .OrderAggbuilder() .setDataSource(dataSource) .setDialect(new MySqlDialect()) .setExecutionOptions(JdbcExecutionOptions.builder() .withBatchSize(1000) .withMaxRetries(3) .build()) .setJdbcStatementBuilder((ps, agg) - { ps.setLong(1, agg.userId); ps.setBigDecimal(2, agg.totalAmount); ps.setLong(3, agg.orderCount); ps.setTimestamp(4, new Timestamp(System.currentTimeMillis())); }) .build();这里有两个配置容易踩坑。第一个是batchSize它控制的是JDBC批量提交的批量大小不是事务大小。JdbcExactlyOnceSink的事务边界是Checkpoint不是batch。如果你以为batchSize设成100事务就会每100条提交一次那就理解错了实际事务只会跟着Checkpoint走。第二个是setMaxRetries这个重试次数针对的是单条SQL执行失败的重试不会改变事务边界。如果你用的是老版本Flink或者想深入理解机制建议自己实现一个TwoPhaseCommitSinkFunction这也是我这个项目里最值得写代码的部分。3.3 核心二手工实现TwoPhaseCommitSinkFunction我实现自定义Sink时使用的不是XAConnection那一套抽象接口而是直接操作MySQL的XA语句。这样每一步在做什么都看得清清楚楚排查问题也更容易。实现的关键点如下。首先是事务句柄类用来保存XIDpublic class XaTransactionHandle { private final String xid; private final Connection conn; // 构造时创建连接并执行 XA START }然后是Sink主体继承自TwoPhaseCommitSinkFunctionpublic class XaJdbc2PcSink extends TwoPhaseCommitSinkFunctionOrderAgg, XaTransactionHandle, XaTransactionHandle { public XaJdbc2PcSink() { super(new ListStateDescriptor(xa-tx-handle, XaTransactionHandle.class), new ListStateDescriptor(xa-tx-pending, XaTransactionHandle.class)); } Override protected XaTransactionHandle beginTransaction() throws Exception { return new XaTransactionHandle(UUID.randomUUID().toString()); } Override protected void preCommit(XaTransactionHandle tx) throws Exception { tx.execute(XA END tx.getXid() ); tx.execute(XA PREPARE tx.getXid() ); } Override protected void commit(XaTransactionHandle tx) throws Exception { tx.execute(XA COMMIT tx.getXid() ); tx.close(); } Override protected void abort(XaTransactionHandle tx) throws Exception { tx.execute(XA ROLLBACK tx.getXid() ); tx.close(); } Override protected void invoke(XaTransactionHandle tx, OrderAgg value, Context context) throws Exception { tx.execute(INSERT INTO t_order_stat(user_id, total_amount, order_count, update_time) VALUES ( value.userId , value.totalAmount , value.orderCount , NOW()) ON DUPLICATE KEY UPDATE total_amount total_amount value.totalAmount , order_count order_count value.orderCount); } }这套代码有个非常关键的点就是invoke方法里执行SQL用的是“事务句柄里的同一条连接”。如果这里不小心用了连接池里的其他连接事务就散架了2PC根本无从谈起。我在每个XaTransactionHandle内部持有一个独立的Connection这条连接的生命周期和checkpoint事务绑定作业里同时活跃的checkpoint数量有多少就需要多少条这样的连接。recoverAndCommit和recoverAndAbort两个方法我没展开写但它们是故障恢复的核心。作业重启后Flink会扫描checkpointedState里的所有未决事务句柄。对每一个句柄先查MySQL里的XA事务状态如果还在PREPARED状态就提交如果没有就回滚。这套逻辑保证了“数据不会因重放而重复”。3.4 关键配置超时、重试、隔离级别用2PC做Sink时Flink的Checkpoint配置要特别上心。我用的配置如下Configuration config new Configuration(); // 相关配置代码 StreamExecutionEnvironment env StreamExecutionEnvironment.getExecutionEnvironment(); env.enableCheckpointing(10_000, CheckpointingMode.EXACTLY_ONCE); env.getCheckpointConfig().setMinPauseBetweenCheckpoints(5_000); env.getCheckpointConfig().setCheckpointTimeout(60_000); env.getCheckpointConfig().setMaxConcurrentCheckpoints(1); env.getCheckpointConfig().setTolerableCheckpointFailureNumber(3);我把Checkpoint间隔设成10秒。这个间隔决定了事务的粒度间隔越大单个事务批次越大数据库锁占用的时间越长间隔太小XA PREPARE和COMMIT的频次太高反而浪费资源。10秒在验证项目里是比较平衡的选择。setMaxConcurrentCheckpoints(1)是我特别强调的配置。2PC的多个并发Checkpoint意味着同时有多个XA事务处于PREPARED状态MySQL对PREPARED但未COMMIT的事务是持有行锁的。并发事务一多很容易互相等锁最后全部超时。强制串行Checkpoint事务之间就不会互相干扰。隔离级别方面MySQL的默认隔离级别REPEATABLE READ就够用。XA事务本身有自己的隔离性保证不需要特意调整。但要注意事务内的写入要避免跨多个非必要的行否则长时间持锁会拖垮业务库。验证时如果发现数据库锁等待指标飙升大概率就是事务批次太大或者Checkpoint间隔太长。4. 验证实验设计怎么证明它真的精确一次4.1 场景A无故障跑批验证幂等去重第一个实验最简单往Kafka里灌1万条订单事件Flink作业在线跑完然后查MySQL两张表看最终结果和预期是否一致。这个场景没有故障它的核心目的是验证“正常路径下2PC不会丢失数据”。具体做法是先用脚本生成有序事件每条事件的user_id离散分布在100个用户上amount随机。然后启动Flink作业消费完这批数据后从UI上观察Checkpoint数量增加数据全部写入。最后执行对账SQLSELECT user_id, total_amount, order_count FROM t_order_stat ORDER BY user_id;和事先算好的期望值逐行对比如果完全一致说明正常路径通过了。这个实验看似简单但很有价值。很多2PC实现的bug在正常路径下就会暴露比如事务提交后连接没有归还、commit方法里没有真正执行XA COMMIT、又或者事务边界和Checkpoint没对齐。正常跑一遍是底线跑不过就不用谈后面的故障实验了。4.2 场景B手动制造故障验证事务回滚验证EOS最有说服力的实验是制造故障让作业在“某个Checkpoint提交前”挂掉然后观察恢复后的结果。我用的故障注入方式是在Flink UI上手动取消作业。操作步骤是这样的作业正常跑起来等到第一二个Checkpoint成功后我向Kafka里灌入一批新数据。等这批数据已经进入Sink的invoke方法、但还没触发下一个Checkpoint时我手动取消作业。这时候这批数据已经被写进MySQL的XA事务里处于未提交状态。然后我从最近的一次Checkpoint恢复作业重新消费那批数据。关键来了正常情况下从Checkpoint恢复会让Flink重新计算并重新执行invoke把同样的数据再写一遍。但因为旧事务还挂在PREPARED状态Flink的恢复逻辑会先执行recoverAndCommit或recoverAndAbort。如果我是手动取消的Checkpoint没有完成那么旧事务应该被回滚同时新一版事务会把数据重新写入并提交最终结果不能有重复累加。我跑这个实验时有一次特别明显的对照我故意关掉recoverAndAbort的恢复逻辑让旧事务永久挂在PREPARED状态。结果恢复后MySQL里出现了一个一直挂着的PREPARED事务行锁不释放后续查询直接锁等待。这个现象直观地说明了如果2PC恢复逻辑缺失轻则数据多写重则拖垮数据库。所以Chekpoint恢复路径上的恢复回调是端到端一致性的胜负手。4.3 场景CKafka消费位点与MySQL写入的对账第三个实验我用了更严格的方式把Kafka的位点信息和MySQL的写入情况做交叉对账。具体做法是Flink消费Kafka时把当前Kafka分区位点透传到MySQL表里记录下来然后对比“MySQL里实际写入的订单ID集合”和“Kafka里理论上暴露的订单ID集合”。这个实验的操作细节是在invoke方法里额外执行一条INSERT到t_order_event明细表把order_id、user_id、amount都存下来。明细表不加任何聚合逻辑纯粹记录“Flink已经拿到了哪些数据”。然后故障恢复后通过一个离线的比对脚本把t_order_event里的order_id集合和Kafka消息里的order_id集合做差集差集里如果有数据说明丢失如果出现了原本Kafka里没有的order_id说明重复。这个场景能同时验证“不丢”和“不重”两个维度比只看整表金额更严谨。实际跑下来两张表的数据完全吻合。4.4 对账结论怎么量化验证项目不能只看“看起来对了”要把指标量化出来才有说服力。我统计了三个指标数据重复率、数据丢失率、对账偏差。数据重复率等于重复写入的订单条数除以总订单条数数据丢失率等于丢失条数除以总条数对账偏差等于MySQL聚合值减去预期聚合值除以预期聚合值。在开启EOS并正确实现2PC的条件下三个指标全部为0。对照组里我关掉Checkpoint或者用普通JdbcSink代替2PC Sink数据重复率直接飙到40%以上因为一批重放数据被重复写入。这个对照非常说明问题也让我对Flink的EOS边界有了更清晰的认识——它只保证“做了Checkpoint的那批数据和事务是对齐的”不保证“没做Checkpoint的数据不会乱”。5. 踩坑记录与问题排查实录5.1 MySQL驱动版本与XA支持问题一开始我用的是MySQL Connector/J 5.1.49结果在执行XA PREPARE语句时一直报XAER_RMERR错误。后来查到原因老版本驱动对XA事务的支持不完整尤其是XA RECOVER的元数据读取能力不行。Flink的TwoPhaseCommitSinkFunction在恢复未决事务时要调用XAResource.recover方法驱动不支持就恢复不了。解决方案是升级到8.0.33版本并设置连接参数useInformationSchematrue。升级后XA事务的所有操作都稳定了。这里提醒不要只看驱动版本新旧要确认它支持18.0版本的XA RECOVER语义否则故障恢复路径会出问题。5.2 二阶段提交超时导致作业卡死实验过程中我遇到过一次作业整体卡死。现象是Checkpoint一直不成功任务运行了半小时也不见进度。排查后发现是MySQL的锁等待有一个XA事务在PREPARED状态后没有提交持有大量行锁后续一批数据写入时要更新同一批用户的行结果全部阻塞。触发原因是当时我把Checkpoint最大并发设成了3三个事务同时PREPARED互相锁冲突。解决办法是设置并发Checkpoint为1并且把MySQL的事务隔离级别在必要时调低。还有一个小技巧处理这种问题时要看information_schema.innodb_trx和performance_schema.events_statements_current能直接看到是哪个XA事务持有锁。5.3 binlog行格式和主键对Exactly-Once的影响验证时我还发现binlog_format对结果有影响。如果MySQL的binlog格式设置为STATEMENTXA PREPARE之后的binlog记录可能不完整导致下游Canal或DRS解析出来的数据出现漂移。我干脆把binlog_format设置成ROW保证每条记录都有完整的前后镜像这样即使后续接数据同步也不会出问题。另一个关键点是表必须有明确主键。如果t_order_stat没有主键那么即使Flink端2PC正确MySQL端做INSERT时也没法对应到同一行数据重复率会居高不下。现在很多“Flink写入MySQL出现重复”的经典案例一大半都是因为目标表缺主键。5.4 关于“精确一次”的最常见误区最后必须说一个我踩过的思维误区。一开始我以为把Sink加上2PC之后无论什么故障都能不重不丢。实际情况是2PC只解决“Flink恢复后外部存储的提交一致性”它不能代替幂等设计。如果事务已经成功提交但Flink在Checkpoint完成前挂了那这批数据在MySQL里已经生效作业恢复后Flink不会回滚已提交的事务只会把这段数据重新写入新事务。这时候必须依赖要么事务回滚要么业务侧的幂等设计来兜底。这也是为什么我在表设计时给t_order_stat加了ON DUPLICATE KEY UPDATE逻辑的原因。2PC负责跨进程的事务一致性幂等键负责在极端情况下兜底两者配合才能构建真正可靠的数据链路。别把2PC当成银弹它是机制的一环不是全部。这个项目整体做下来我个人最大的体会是纸上谈兵聊一百遍“EOS原理”不如亲手把MySQL的PREPARED事务挂在那边再去重启一次Flink作业看得直观。2PC协议看着复杂真正落地的时候就几个关键判断——事务边界跟没跟Checkpoint对齐、恢复时有没有处理未决事务、目标表能不能承受PREPARED状态的锁。把这几个点验证透了后面再去做其他数据库的Sink就一通百通了。如果你也想复现建议不要一上来就上生产链路先用一条Kafka Topic加一张表把这个闭环跑通。我最后还有一个扩展建议在Sink之外再挂一个用于幂等对账的Redis表或明细表可以进一步验证“多个下游同时写入”场景下的2PC效果那是另一个更刺激的话题了。