Flink CDC中我使用datastream的方式消费mysql的binlog，正常消费没有问题？

问题1:Flink CDC中我使用datastream的方式消费mysql的binlog，正常消费没有问题，但是指定时间戳消费就是消费不到数据。目前的mysql-cdc的版本是2.4.0，flink的版本是1.15.3，也没看到有什么报错信息，求指点？你们有使用datastream，flink的mysql-cdc的从指定时间戳消费数据的吗？有成功的案例吗？问题2:我目前测试从最开始的消费也就5条数据。我测试的时候先是inital模式，从头消费一遍，然后改为最早的时间，比如当前时间是10:27，我改为10:24，然后还是没有数据进来呢有的呢，{"op":"r","after":{"id":9,"f1":"字符串4","f2":1687516440000,"f3":20.2},"source":{"server_id":0,"version":"1.9.7.Final","file":"","connector":"mysql","pos":0,"name":"mysql_binlog_source","row":0,"ts_ms":0,"snapshot":"false","db":"changancar","table":"t_hubing_demo"},"ts_ms":1688091808608} 2023-06-30 10:23:28 {"op":"r","after":{"id":8,"f1":"字符串3","f2":1687516200000,"f3":20.34},"source":{"server_id":0,"version":"1.9.7.Final","file":"","connector":"mysql","pos":0,"name":"mysql_binlog_source","row":0,"ts_ms":0,"snapshot":"false","db":"changancar","table":"t_hubing_demo"},"ts_ms":1688091808607} 2023-06-30 10:23:28 {"op":"r","after":{"id":7,"f1":"字符串2","f2":1684652903000,"f3":10.23},"source":{"server_id":0,"version":"1.9.7.Final","file":"","connector":"mysql","pos":0,"name":"mysql_binlog_source","row":0,"ts_ms":0,"snapshot":"false","db":"changancar","table":"t_hubing_demo"},"ts_ms":1688091808607} 2023-06-30 10:23:28 {"op":"r","after":{"id":6,"f1":"字符串1","f2":1684739303000,"f3":1.23},"source":{"server_id":0,"version":"1.9.7.Final","file":"","connector":"mysql","pos":0,"name":"mysql_binlog_source","row":0,"ts_ms":0,"snapshot":"false","db":"changancar","table":"t_hubing_demo"},"ts_ms":1688091808606} {"op":"c","after":{"id":10,"f1":"字符串5","f2":1688120820000,"f3":21.2},"source":{"thread":12304031,"server_id":2884211128,"version":"1.9.7.Final","file":"mysql-bin.002400","connector":"mysql","pos":23506761,"name":"mysql_binlog_source","gtid":"bcb4812b-f07a-11ec-92fb-0c42a13e2cb8:286796453","row":0,"ts_ms":1688092055000,"snapshot":"false","db":"changancar","table":"t_hubing_demo"},"ts_ms":1688092056001}这就是我测试的几条数据，指定的时间戳就是：1688091808606理论上讲，这5条数据应该都需要重新消费一遍进来才对

展开

收起

真的很搞笑 2023-07-02 17:42:38 188 0

3 条回答

写回答

取消提交回答

算精通

北京阿里云ACE会长

是的，您可以使用 Flink DataStream API 来消费 MySQL Binlog，实现实时数据同步和处理。具体来说，您可以使用 Flink CDC 提供的 FlinkCDCSource 类，将 MySQL Binlog 转换为 Flink DataStream，并进行后续的数据处理和分析。
使用 FlinkCDCSource 来消费 MySQL Binlog 的代码示例如下：
java
Copy
Properties props = new Properties();
props.setProperty("scan.startup.mode", "latest-offset");
FlinkCDCSource source = FlinkCDCSource.builder()
.hostname("localhost")
.port(3306)
.databaseList("test")
.tableList("test_table")
.username("root")
.password("root")
.deserializer(new StringDebeziumDeserializationSchema())
.debeziumProperties(props)
.build();
DataStreamSource stream = env.addSource(source);
stream.print();
env.execute("Flink CDC Example");
在上面的示例中，使用 FlinkCDCSource 构建了一个 Flink DataStreamSource 对象，并将其作为输入流进行后续的数据处理和分析。需要注意的是，StringDebeziumDeserializationSchema 用于将 Binlog 数据转换为 String 类型，您可以根据实际需求选择适合的反序列化器。

2023-07-30 09:36:39

赞同展开评论打赏
Star时光

问题1：在使用 Flink CDC 的 DataStream 方式消费 MySQL 的 binlog 时，正常消费没有问题，但是指定时间戳消费却无法获取到数据。你使用的 MySQL CDC 版本是2.4.0，Flink 版本是1.15.3，并且没有看到任何报错信息。你想请教是否有人成功地使用 DataStream 和 Flink 的 MySQL CDC 在指定时间戳消费数据的案例。

回答1：关于指定时间戳消费数据的问题，确保你正确配置了时间戳参数，并且所指定的时间戳不是太接近当前时间。从时间戳消费本质上是从 earliest（最早）开始消费，然后根据时间戳过滤掉不符合时间区间的数据。这可能需要一定的等待时间才能消费到新的数据。你可以观察 CPU 和内存的使用情况来判断是否正在进行消费。

回答2：对于指定时间戳消费的问题，时间戳是指 binlog 中的变更时间。请确认在你指定的时间之后是否有数据写入 MySQL 数据库。如果在指定的时间戳之后没有数据写入，那么就不会有数据被消费。

需要注意的是，如果数据写入频率较低，可能需要等待一段时间才能消费到指定时间戳之后的数据。另外，还可以检查相关日志以获取更多的调试信息，例如 MySQL 的 binlog 日志和 Flink CDC 的日志等。

2023-07-30 09:39:46

赞同展开评论打赏
芯在这

回答1:你指定时间戳的配置是什么？是否给的是一个离当前时间很近的时间。从时间戳消费本质上是从earliest消费，然后过滤掉不符合时间区间的数据，因此可能等的时间比较长。你可以观察cpu和内存来判断是不是在消费了回答2:时间戳指的是binlog的变更时间。mysql在10:24之后有数据写入吗？，此回答整理自钉群“Flink CDC 社区”

2023-07-02 17:47:38

赞同展开评论打赏