ToB企服应用市场:ToB评测及商务社交产业平台

标题: Flink CDC同步MySQL数据至Doris及遇到的问题踩坑 [打印本页]
作者: 愛在花開的季節    时间: 2024-12-30 00:13
标题: Flink CDC同步MySQL数据至Doris及遇到的问题踩坑
Flink CDC同步MySQL数据至Doris

1、情况准备

  1. Flink vesion: 1.17.2
  2. Doris version: 2.1.7
  4. jars:
  5.         flink-sql-connector-mysql-cdc-3.0.1.jar
  6.         mysql-connector-java-8.0.27.jar
  7.         flink-doris-connector-1.17-24.0.1.jar
复制代码
同步时需要在 $FLINK_HOME/lib 目次下添加对应的 Flink CDC 依赖，如上述。
下载链接: maven 仓库
  1. # 启动flink yarn session
  2. yarn-session -d
复制代码
2、编写CDC同步脚本

  1. cd $FLINK_HOME
  2. vim bin/mysql2doris.sh
  4. bin/flink run \
  5.     -Dexecution.checkpointing.interval=10s \
  6.     -Dparallelism.default=4 \
  7.     -Denv.java.opts="-Dfile.encoding=UTF-8" \
  8. #解决中文乱码问题
  9.     -c org.apache.doris.flink.tools.cdc.CdcTools \
  10.     lib/flink-doris-connector-1.17-24.0.1.jar \
  11.     mysql-sync-database \
  12.     --database doris_database_name\
  13.     --mysql-conf hostname=xxx \
  14.     --mysql-conf port=3306 \
  15.     --mysql-conf username=root\
  16.     --mysql-conf password=123456 \
  17.     --mysql-conf database-name=mysql_database_name \
  18.     --mysql-conf scan.incremental.snapshot.chunk.key-column=database.table1:column1,database.table2:column1 \
  19.     --including-tables "[a-z].*" \ #需要同步的 MySQL 表，可以使用 \| 分隔多个表，并支持正则表达式。比如--including-tables table1
  20.     --sink-conf fenodes=xxx \
  21.     --sink-conf username=root \
  22.     --sink-conf password=123456 \
  23.     --sink-conf jdbc-url=jdbc:mysql://xxx:9030 \
  24.     --sink-conf sink.label-prefix=label_1 \
  25.     --table-conf replication_num=3
  27. # 启动脚本
  28. sh bin/mysql2doris.sh
复制代码
3、验证


查看checkpoint情况

checkpoint完成，数据同步成功，去doris查看数据即可。
4、踩坑记载

cdc任务启动失败

1、 java.lang.NoSuchFieldError: SCAN_INCREMENTAL_CLOSE_IDLE _READER_ENABLED


错误原因：jar包使用错误，使用的是 flink-connector-mysql-cdc-${version}.jar
解决方案：必须使用 flink-sql-connector-mysql- ${version}.jar
2、Caused by: java.lang.ClassNotFoundException: com.ververica.cdc.connectors.mysql.source.MySqlsource



错误原因:由于之前为相识决第一个错误，下了很多个 flink-sql-connector-mysql- ${version}.jar包到lib目次下，逐个反编译各个jar包后发现 MySqlsource 是存在的。大概lib目次下存在多个jar包，存在冲突

解决方案：将多余的cdc jar包移出lib目次，并重启flink集群
3、写入字符范例数据时凌驾varchar长度限制，Doris写入失败



错误原因：查看flink-doris-connector后发现，connector自动建表时，默认将源表中的varchar范例长度*3，所以不存在超出长度限制问题。
又发现写入数据是中文，doris显示是乱码，大概存在中文乱码问题。

解决方案：
1、修改flink设置文件 flink-conf.yml ， 添加以下内容
  1. env.java.opts: -Dfile.encoding=UTF-8
复制代码
重启flink yarn session
2、在cdc脚本中添加
  1. -Denv.java.opts="-Dfile.encoding=UTF-8" \
复制代码
两种方案选一即可
4、scan.incremental.snapshot.chunk.key-column’ must be set when the table doesn’t have primary keys

错误原因：整库同步时，遇到部门了无主键的表
解决方案：
1、忽略这些表，不同步
脚本里添加
  1. --excluding-tables "table1|table2" # 不同表之间用 | 分隔
复制代码
2、设置 scan.incremental.snapshot.chunk.key-column，且只能选择非空范例的一个字段。
脚本里添加
  1. --mysql-conf scan.incremental.snapshot.chunk.key-column=database.table1:column1,database.table2:column1
  2. # 不同的库表列之间用 , 隔开。
复制代码
免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！更多信息从访问主页：qidao123.com:ToB企服之家，中国第一个企服评测及商务社交产业平台。



欢迎光临 ToB企服应用市场:ToB评测及商务社交产业平台 (https://dis.qidao123.com/) Powered by Discuz! X3.4