基于flink 1.12 如何通过Flink SQL 实现 多sink端 在同一个事务中？
hi all:
我目前基于flink 1.12 sql 来开发功能， 目前遇到一个问题， 我现在想实现 在一个事务里 先将kafka 源的数据写入到一张msyql 表中， 写入成功后再写入一张kafka的表里。如果写入db失败，则不写入kafka 。 语句类似这种： insert into db_table_sink select * from kafka_source_table; insert into kafka_table_sink select * from kafka_source_table;
请问flink SQL 有实现方式吗？ 目前经过测试，发现如果db写入失败，但是kafka依旧写入，同时flink 程序没有挂掉。*来自志愿者整理的flink邮件归档
版权声明：本文内容由阿里云实名注册用户自发贡献，版权归原作者所有，阿里云开发者社区不拥有其著作权，亦不承担相应法律责任。具体规则请查看《阿里云开发者社区用户服务协议》和《阿里云开发者社区知识产权保护指引》。如果您发现本社区中有涉嫌抄袭的内容，填写侵权投诉表单进行举报，一经查实，本社区将立刻删除涉嫌侵权内容。
在Apache Flink 1.12中，直接通过SQL来实现跨数据源的事务性操作（即保证MySQL和Kafka两个写入操作的原子性）并不直接支持。Flink SQL的事务控制主要是针对单个sink，而不是跨多个sink。这意味着如果你需要确保先写MySQL成功后再写Kafka，并且在MySQL写入失败时回滚Kafka的操作，你可能需要采用更复杂的处理方式。
一种可行的方案是利用Flink的两阶段提交（2PC, Two-Phase Commit）思想或者自定义Sink来实现这个需求，但这通常涉及到编写用户自定义函数（UDF/UDTF）或自定义sink，而非直接使用SQL完成。
自定义Sink方法
-
创建一个自定义Sink：你可以创建一个自定义的Sink，该Sink内部实现对MySQL和Kafka的写入逻辑，并且管理它们之间的事务一致性。这要求你在Sink内部实现必要的事务控制逻辑，比如尝试写MySQL，如果成功则继续写Kafka，如果任一环节失败则进行回滚。
-
使用两阶段提交模式：在自定义Sink中模拟两阶段提交过程。首先，预写（Prepare）阶段，尝试将数据写入到MySQL和临时的Kafka主题（或标记为待确认状态）。然后，在提交（Commit）阶段，确认MySQL写入成功后，才正式写入到目标Kafka主题；如果发现MySQL写入失败，则执行回滚操作，从临时Kafka主题移除数据或取消待确认状态。
-
考虑使用Flink的Exactly Once语义：虽然这不直接解决跨sink的事务问题，但确保每个sink自身的事务处理正确对于整体的一致性至关重要。确保你的Kafka sink配置为幂等写入，并且MySQL sink也支持事务处理。
注意事项
- 实现上述方案可能较为复杂，需要深入理解Flink的运行机制和编程模型。
- 考虑性能影响，两阶段提交可能会增加延迟。
- 确保所有操作具备幂等性，以防止重试时产生重复数据。
由于直接通过Flink SQL实现这一需求存在局限，建议评估是否可以接受一定程度的数据不一致风险，或者是否有业务流程上的调整空间，例如通过外部协调服务（如ZooKeeper）来辅助实现更高级别的事务一致性。
