开发者社区> 问答> 正文

如何合并 binlog stream 和 table stream?

我在用 Debezium 从数据库读取最新的 binlog 写入 Kafka,比如对于 mysql_server.test.tableA 有一个 topic “mysql_server.test.tableA”,我需要在 Flink 里实现这样的逻辑:

  1. 先连接上 Kafka 开始消费 topic “mysql_server.test.tableA”,确保连接成功,记为 binlog-stream,但是要暂停消费 Kafka;
  2. 用 JDBCInputFormat 读取 test.tableA 到一个 DataStream 里,记为 table-stream;
  3. 合并两个 streams,消费完 table-stream 后再开始消费 binlog-stream,这样可以确保 binlog 是 应用到某个快照表上。

问题是我怎么能暂停消费 binlog-stream 呢? 我目前想到的办法是用 flink state 做一个全局状态 startBinlog,初始值为 false:

binlog-stream -> waitOperator -> sinkOperator table-stream -> notifyOperator -> sinkOperator

两个流被合并输出到 sinkOperator,waitOperator() 会 while loop阻塞式的检查全局状态, 等 table-stream 消费完(不知道怎么判断消费完了。。。), notifyOperator 修改全局状态,这样 binlog-stream 就能被继续消费了。

但由于 kafka consumer 如果长期阻塞不 ack 的话,kafka consumer 会被断开,所以这个做法应该是不行的。

请教怎么破?

谢谢!*来自志愿者整理的FLINK邮件归档

展开
收起
玛丽莲梦嘉 2021-12-03 18:24:29 912 0
1 条回答
写回答
取消 提交回答
  • 你这里的合并是用join 来做么? 这样的话,会比较耗性能。

    一种做法是先消费 jdbc table, 消费完后,再切换到 kafka 上。这种要求 binlog 是幂等操作的,因为会有多处理一部分的 binlog,没法做到 精确地切换到 kafka offset 上。

    另外你也可以参考下 StreamSQL 的 bootstrap 的做法: https://streamsql.io/blog/from-apache-kafka-to-apache-pulsar*来自志愿者整理的FLINK邮件归档

    2021-12-03 21:17:52
    赞同 展开评论 打赏
问答排行榜
最热
最新

相关电子书

更多
PolarDB-X 2.0 全局 Binlog 与备份恢复能 立即下载
低代码开发师(初级)实战教程 立即下载
阿里巴巴DevOps 最佳实践手册 立即下载