Doris实时数仓dim层构建-编程学习网

FlinkCDC接入多表或整库

前言

flink cdc实时同步mysql维表
本文是基于flink 1.16

一、导入flink-doris-connector jar包

将 flink-doris-connector-1.16-1.4.0.jar 上传至 $FLINK_HOME/lib目录下
可至 maven官网下载

二、cdc脚本

--database 同步到Doris的数据库名。--table-prefix Doris表前缀名，例如 --table-prefix ods_。--table-suffix 同上，Doris表的后缀名。--including-tables 需要同步的MySQL表，可以使用"|" 分隔多个表，并支持正则表达式。 比如--including-tables table1|tbl.*就是同步table1和所有以tbl开头的表。--excluding-tables 不需要同步的表，用法同上。--mysql-conf MySQL CDCSource 配置，例如--mysql-conf hostname=127.0.0.1 ，您可以在这里查看所有配置MySQL-CDC，其中hostname/username/password/database-name 是必需的。--sink-conf Doris Sink 的所有配置，可以在这里查看完整的配置项。--table-conf Doris表的配置项，即properties中包含的内容。 例如 --table-conf replication_num=1--ignore-default-value 关闭同步mysql表结构的默认值。适用于同步mysql数据到doris时，字段有默认值，但实际插入数据为null情况。--use-new-schema-change 新的schema change支持同步mysql多列变更、默认值。# 若要接入整库，则 including-tables 和 excluding-tables不填

三、脚本配置

vim cdc.sh# 写入如下内容export HADOOP_CLASSPATH=`hadoop classpath`# 我部署的是yarn session模式，以application形式启动程序，这可自行更改$FLINK_HOME/bin/flink run-application -t yarn-application \    -Djobmanager.memory.process.size=700m \    -Dtaskmanager.memory.process.size=1024m \#必须设置checkpoint    -Dexecution.checkpointing.interval=10s \    -Drest.bind-port=8082-8087 \    -Dparallelism.default=1 \    -Denv.java.opts="-Dfile.encoding=UTF-8" \ #程序入口及jar包位置    -c org.apache.doris.flink.tools.cdc.CdcTools \      $FLINK_HOME/lib/flink-doris-connector-1.16-1.4.0.jar \#以下是cdc配置    mysql-sync-database \    --database doris_test --mysql-conf hostname=127.0.0.1 --mysql-conf username=root --mysql-conf password=123456 --mysql-conf database-name=cc_test --including-tables "tbl1|test.*" --sink-conf fenodes=127.0.0.1:8030 --sink-conf username=root --sink-conf password=000000 --sink-conf jdbc-url=jdbc:mysql://127.0.0.1:9030 --sink-conf sink.label-prefix=test_1# 同步tbl1和所有以test开头的表