dsimile / flink-cdc Goto Github PK

概述：修改源代码中 mysql 连接器 StartupOptions.initial 全量同步无主键表异常

问题：开源cdc 2.2.0 中 同步mysql 采用了全新的chunk 算法 https://flink-learning.org.cn/article/detail/3ebe9f20774991c4d5eeb75a141d9e1e,此算法全量同步表时包含两个步骤 全量读取chunk 以及合并读取chunk 时 binlog 的lowWatermark 和 highWatermark 的增量部分，在全量读取完成后会与增量部分的binlog合并，由于表没有主键 所以增量部分的binlog 无法替换全量中的数据（针对同一条数据而言），在源码中ChunkUtils.getChunkKeyColumn

if (primaryKeys.isEmpty()) { throw new ValidationException( String.format( "Incremental snapshot for tables requires primary key," + " but table %s doesn't have primary key.", table.id())); }

如果表不存在主键则直接抛出异常

解决思路：将 binlog 的lowWatermark 和 highWatermark 的增量binlog 发出，在下游去重。

使用方法： 将项目clone 本地，打maven 包 到本地，在自己项目中引用 MySqlSource mySqlSource = MySqlSource.builder() .hostname("") .port(3306) .databaseList("") // set captured database .tableList("") // set captured table .username("") .password("") .startupOptions(StartupOptions.initial()) .deserializer(new JsonDebeziumDeserializationSchema()) // converts SourceRecord to JSON String .supportNoMysqlPrimaryKey(true) .build();

supportNoMysqlPrimaryKey(true) 开启无主键同步功能，不设置默认false