flinkcdc同步完全量数据就不同步增量数据了-摩杜云开发者社区

flinkcdc同步完全量数据就不同步增量数据了

使用flinkcdc同步mysql数据，使用的是全量采集模型
startupOptions(StartupOptions.earliest())
全量阶段同步完成之后，发现并不开始同步增量数据，原因有以下两个：

原因1：

1.mysql中对应的数据库没有开启binlog
在/etc/my.cnf配置文件中，在[ mysqld ]添加以下内容

[mysqld]

#数据库id
server-id=1
##启动Binlog，该参数的值会作为binlog的文件名前缀
log-bin=mysql-bin
##binlog类型，maxwell要求为row类型
binlog_format=row
##启动binlog的数据库，需根据实际情况修改配置
binlog-do-db=student
binlog-do-db=easypm

然后重启数据库，执行命令

sudo systemctl restart mysqld

原因2

和checkpoint有关
1.没有开启ck
2.ck的时间过长
测试阶段，可以把ck的时间设置的非常短

env.enableCheckpointing(1000L, CheckpointingMode.EXACTLY_ONCE);//头和头的之间
env.getCheckpointConfig().setMinPauseBetweenCheckpoints(1000L);//头和尾

3.ck的路径收到损坏，删除对应的ck路径下的内容即可。

        MySqlSource<String> source = MySqlSource.<String>builder()
                .hostname("127.0.0.1")
                .port(3306)
                .username("root")
                .password("root__")
                .databaseList("bigscreen")
                .tableList("bigscreen.sys_user")
                .startupOptions(StartupOptions.initial())
                .deserializer(new JsonDebeziumDeserializationSchema())
                .includeSchemaChanges(true)
                .build();

.startupOptions(StartupOptions.initial())

StartupOptions的

flinkcdc同步完全量数据就不同步增量数据了_mysql

// 默认全量
StartupOptions.initial();

// 从最开始的binlog 开始
StartupOptions.earliest();

// 从最后的binlog开始
StartupOptions.latest();

// 从指定时间戳开始
StartupOptions.timestamp(long startupTimestampMillis);

// 从指定GTID开始
StartupOptions.specificOffset(String gtidSet)

// 从构建的binlogOffset构建的位置开始
StartupOptions.specificOffset(BinlogOffset binlogOffset)

// 从指定的binlog文件名和位置开始
StartupOptions.specificOffset(String specificOffsetFile, Long specificOffsetPos)

MySQLSource.builder()
    .startupOptions(StartupOptions.earliest()) // 从最早位点启动
    .startupOptions(StartupOptions.latest()) // 从最晚位点启动
    .startupOptions(StartupOptions.specificOffset("mysql-bin.000003", 4L) // 从指定 binlog 文件名和位置启动
    .startupOptions(StartupOptions.specificOffset("24DA167-0C0C-11E8-8442-00059A3C7B00:1-19")) // 从 GTID 集合启动
    .startupOptions(StartupOptions.timestamp(1667232000000L) // 从时间戳启动
    ...
    .build()