flink 与cdc 版本使用搭配:
flink1.13.6 + flink mysql cdc 1.4.0
flink 1.16.0 + flink mysql cdc 2.3.0
flink 1.16.0 + flink mysql cdc 2.4.0
flink 1.16.0 + flink postgresql cdc 2.3.0
flink 1.13.6 + flink mysql cdc 2.3.0 : 没有报错,没有数据,估计是兼容有问题
flink cdc
参数说明
1、调整chunck大小 : scan.incremental.snapshot.chunk.size
2、设置cdc模式:scan.startup.mode【initial(默认)、latest-offset】
3、支持chunk key 列设置,默认是第一个字段:scan.incremental.snapshot.chunk.key-column
官网:https://ververica.github.io/flink-cdc-connectors/master/content/connectors/postgres-cdc.html
原理分析
1、cdc mysql 全量快照阶段split sql :SELECT * FROM cdc_db
.tablename
WHERE id >= ? AND NOT (id = ?) AND id <= ?;
备注:id 是主键id
(DBLog)无锁算法论文
链接地址:https://arxiv.org/pdf/2010.12597.pdf , 对此算法感兴趣的可以看这位大佬的分享:https://zhuanlan.zhihu.com/p/600303844
论文部分摘要理解:
-
全量阶段:
1、flink cdc 任务启动后按设置的chunk size切分数据,sql如下:
(sql:SELECT * FROMcdc_db
.tablename
WHERE id >= ? AND NOT (id = ?) AND id <= ?; )
2、同时会启动读取binlog任务,读取chunk对应的binlog,通过binlog对 select chunk的数据做合并操作,此操作是合并在期间执行了update、delete操作,保证insert-only -
增量阶段:
1、不断追加数据
mysql cdc
cdc api 动态加表
1、启动任务,复制checkpoint路径
2、新增监听的表到tableList(可以使用同一个jar包,在外部传参动态加表)
3、从checkpoint初重启任务即可
flink cdc sql 性能压测
1、cdc mysql sink to kafka :一个takmanager , 4个slot , source 并发度4,sink kafka 并发度1 ,最高写入2.8W条/s
flink cdc api 性能压测
1、cdc mysql sink to kafka :一个takmanager , 4个slot , source 并发度4,sink kafka 并发度1 ,最高写入2.8W条/s
PostgreSqlCDC
执行更新语句,会出现 2 种情况
1、若更新字段包含(部分)主键字段,会先发送一条删除之前主键的记录。op = d , after = null ; 然后再发送一条新主键记录,op = c,且before = null 。
2、若仅更新非主键主键,只会发送一条记录,op = u , before = null。
主体代码如下:
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
// 必须开启 checkpoint ,因为Flink Postgres CDC 只会在 checkpoint 完成的时候更新 Postgres slot 中的 LSN,否则磁盘使用率会一直很高
env.enableCheckpointing(1000);
//监听 postgresql wal 日志
DebeziumSourceFunction<String> sourceFunction = PostgreSQLSource.<String>builder()
.hostname(host)
.port(port)
.username(userName)
.password(passWord)
.database(dbName)
.tableList(tableList)
.deserializer(new JsonDebeziumDeserializationSchema())
.slotName(slotName)
.build();
DataStreamSource<String> dataStreamSource = env.addSource(sourceFunction);
dataStreamSource.print(">>>").setParallelism(1);
env.execute();
cdc sink to kafka
AT_LEAST_ONCE 模型要配置 acks = 1
报错
mysql时区错误,The server time zone value ‘EDT’ is unrecognized or represents
登录mysql并查询当前时区:show variables like “%time_zone%”;
执行以下命令修改时区:
set global time_zone = '+8:00'; ##修改mysql全局时区为北京时间,即我们所在的东8区
set time_zone = '+8:00'; ##修改当前会话时区
flush privileges; #立即生效
java.lang.NoClassDefFoundError: io/debezium/connector/mysql/MySqlConnectorConfig
缺包,引入 debezium-connector-mysql-1.6.4.Final.jar包会报Could not initialize class io.debezium.connector.mysql.MySqlConnectorConfig,看 flink cdc 社区群反馈可能是
Cannot discover a connector using option: ‘connector’=‘mysql-cdc’
删除pom.xml文件flink-connector-mysql-cdc依赖报下的provided
Could not instantiate the executor. Make sure a planner module is on the classpath
包冲突原因。注释或删除jar:flink-table-planner-loader-1.16.0.jar文章来源:https://www.toymoban.com/news/detail-604123.html
(source 算子 )The TaskExecutor is shutting down.
加大心跳间隔时间,默认是30s,‘heartbeat.interval’ = ‘60s’文章来源地址https://www.toymoban.com/news/detail-604123.html
到了这里,关于flink Mysql CDC(动态加表)、postgresqlCDC 和 CDC无锁算法的文章就介绍完了。如果您还想了解更多内容,请在右上角搜索TOY模板网以前的文章或继续浏览下面的相关文章,希望大家以后多多支持TOY模板网!