导读 本文主要分享 Apache Doris 是如何一键实现MySQL万表整库同步的。
全文目录：
环境信息
功能介绍
前提概要
快速体验
常见问题
1. 硬件信息
CPU：48C
内存：256G
2. 软件信息
系统：CentOS
Apache Doris版本：2.0.2
MySQL版本：5.7
Flink版本：1.17.1
Flink-Doris-Connector版本：1.5
Mysql万表存量+增量同步至Doris/SelectDB，主要是通过Flink计算引擎、基于Mysql CDC Connector和Doris社区研发的Flink Doris Connector 实现的。
无需提前建表，当Flink任务启动后，Flink Doris Connector 会自动识别对应的Doris表是否存在，不存在则自动创建Doris表，如果存在则直接启动同步任务。
并且能够支持Mysql源表Schema增删列自动变更至Doris，不需要手动变更Doris表的Schema或重启Job，便利之极。
1. 安装部署
Doris/Flink/Mysql的安装部署流程可以根据版本自行Google。
① Doris下载地址：
https://doris.apache.org/zh-CN/download
② Flink下载地址：
https://dlcdn.apache.org/flink/flink-1.17.1
③ MySQL包下载地址：
https://downloads.mysql.com/archives/installer
2. JAR包准备
3. 脚本模版
run \-Dexecution.checkpointing.interval=10s \-Dparallelism.default=1 \-c org.apache.doris.flink.tools.cdc.CdcTools \lib/flink-doris-connector-1.17-1.5.0-SNAPSHOT.jar \mysql-sync-database \--database doris_db \--mysql-conf hostname= 127.0.0.1 \--mysql-conf port= 3306 \--mysql-conf username= root \--mysql-conf password= root \--mysql-conf database-name= mysql_db \--mysql-conf scan.startup.mode=initial \--including-tables "test.*" \--sink-conf fenodes= 127.0.0.1:8030 \--sink-conf username=root \--sink-conf password=root \--sink-conf jdbc-url=jdbc:mysql://127.0.0.1:9030 \--sink-conf sink.label-prefix=label \--table-conf replication_num=1 \
–job-name Flink任务名称, 非必需。
-Dexecution.checkpointing.interval checkpoint时间间隔，实际task中是需要checkpoint后才执行数据写入。
–database 同步到Doris的数据库名。
–mysql-conf MySQL CDC Source 配置，
例如–mysql-conf hostname=127.0.0.1 ，可以在这里查看所有配置MySQL-CDC：
https://ververica.github.io/flink-cdc-connectors/release-2.4/content/connectors/mysql-cdc.html
其中hostname/username/password/database-name 是必需的。
–including-tables 需要同步的MySQL表，可以使用"|" 分隔多个表，并支持正则表达式。比如–including-tables “test.*” 就是同步所有以test开头的表。
–sink-conf Doris Sink 的所有配置，可以在这里查看完整的配置项：
https://doris.apache.org/zh-CN/docs/dev/ecosystem/flink-doris-connector/#%E9%80%9A%E7%94%A8%E9%85%8D%E7%BD%AE%E9%A1%B9
–table-conf Doris表的配置项，即properties中包含的内容。例如 --table-conf replication_num=1
1. 单表同步
例如单表同步MySQL中Star Schema Benchmark的part零件信息表，共1600000条记录。
run \-Dexecution.checkpointing.interval=10s \-Dparallelism.default=1 \-c org.apache.doris.flink.tools.cdc.CdcTools \lib/flink-doris-connector-1.17-1.5.0-SNAPSHOT.jar \mysql-sync-database \--database ssb_test \--mysql-conf hostname=172.21.16.12 \--mysql-conf port=3306 \--mysql-conf username=root \--mysql-conf password=root \--mysql-conf database-name=ssb_test \--mysql-conf scan.startup.mode=initial \--mysql-conf scan.incremental.snapshot.chunk.key-column=ssb_test.part:p_partkey \--including-tables "part" \--sink-conf fenodes=127.0.0.0:8030 \--sink-conf username=root \--sink-conf password=root \--sink-conf jdbc-url=jdbc:mysql://127.0.0.0:9030 \--sink-conf sink.label-prefix=label \--table-conf replication_num=1 \
在FLINK_HOME执行上述命令后，WEB UI中会新增相应Job。
查看taskmanager日志可以发现，task会先对mysql主键做切分，然后再根据ckp的时间间隔按批写入commit（默认2pc提交）。
默认环境10秒ckp下，单并发写入只需3min左右即可同步完160W数据，且单副本数据压缩率高达90%！
2. 整库同步
-- 创建测试库create database ssb_test_part_all;-- 创建测试表 part01~3CREATE TABLE `part03` (`p_partkey` int(11) DEFAULT NULL,`p_name` varchar(23) DEFAULT NULL,`p_mfgr` varchar(7) DEFAULT NULL,`p_category` varchar(8) DEFAULT NULL,`p_brand` varchar(10) DEFAULT NULL,`p_color` varchar(12) DEFAULT NULL,`p_type` varchar(26) DEFAULT NULL,`p_size` int(11) DEFAULT NULL,`p_container` varchar(11) DEFAULT NULL) ENGINE=InnoDB DEFAULT CHARSET=utf8-- 数据写入 part01~3insert into ssb_test_part_all.part01select * from ssb_test.part
run \-Dexecution.checkpointing.interval=10s \-Dparallelism.default=1 \-c org.apache.doris.flink.tools.cdc.CdcTools \lib/flink-doris-connector-1.17-1.5.0-SNAPSHOT.jar \mysql-sync-database \--database ssb_test_part_all \--mysql-conf hostname=127.0.0.0.12 \--mysql-conf port=3306 \--mysql-conf username=root \--mysql-conf password=root \--mysql-conf database-name=ssb_test_part_all \--mysql-conf scan.startup.mode=initial \--mysql-conf scan.incremental.snapshot.chunk.key-column=ssb_test_part_all.part01:p_partkey,ssb_test_part_all.part02:p_partkey,ssb_test_part_all.part03:p_partkey \--including-tables ".*" \--sink-conf fenodes=127.0.0.0:8030 \--sink-conf username=root \--sink-conf password=root \--sink-conf jdbc-url=jdbc:mysql://127.0.0.0:9030 \--sink-conf sink.label-prefix=label \--table-conf replication_num=1 \
1. MySQL通信异常
可能是因为mysql信息填写错误或驱动包不兼容导致。
2. MySQL无Key表同步异常
mysql源表没有设置主键，可以在脚本中指定–mysql-conf scan.incremental.snapshot.chunk.key-column=database.table:column,database.table1.column… 解决。
3. CKP ON HDFS 连接异常
使用hadoop作为checkpoint时报如下异常：
Caused by: java.net.ConnectException:Call From hadoop1/ip to hadoop1:8020failed on connection exception:java.net.ConnectException: Connection refused;For more details see: http://wiki.apache.org/hadoop/ConnectionRefused
这类异常一般是由于网络原因或端口配置错误导致，可以先进行telnet ip port看是否通，再进行调整处理。
4. CKP ON HDFS 权限异常
使用默认root跑脚本时，如果出现如下异常：
Caused by: org.apache.hadoop.security.AccessControlException:Permission denied: user=root, access=WRITE, inode="/":hadoop:supergroup:drwxr-xr-x
这类异常一般切换至对应用户或者授权即可，这个case可以切换至hadoop用户起 或 将root用户加到hdfs用户组中即可。
5. 官方【Flink Doris Connector】QA
其它问题可以参考 官方【Flink Doris Connector】QA 或 私 或 在开源社区群交流。
至此，【Apache Doris】一键实现MySQL万表整库同步 分享结束，查阅过程中若遇到问题欢迎留言交流。
最后，欢迎关注Flink Forward Asia 2023 FFA 2023 ，Doris Committer 吴迪分享：《Flink 到 Doris 实时写入实践：基于 Flink CDC 构建更实时高效的数据集成链路》
往期推荐
