MySQL到Doris的StreamingETL实现（Flink CDC 3.0）

MySQL到Doris的StreamingETL实现（Flink CDC 3.0）

1 情况准备

1）安装FlinkCDC

1. [root@hadoop1 software]$ tar -zxvf flink-cdc-3.0.0-bin.tar.gz -C /opt/module/

复制代码

2）拖入MySQL以及Doris依靠包
将flink-cdc-pipeline-connector-doris-3.0.0.jar以及flink-cdc-pipeline-connector-mysql-3.0.0.jar防止在FlinkCDC的lib目次下
2 同步变动

2）编写MySQL到Doris的同步变动设置文件
在FlinkCDC目次下创建job文件夹，并在该目次下创建同步变动设置文件

1. [root@hadoop1 flink-cdc-3.0.0]$ mkdir job/
3. [root@hadoop1 flink-cdc-3.0.0]$ cd job
5. [root@hadoop1 job]$ vim mysql-to-doris.yaml
9. source:
11. type: mysql
13. hostname: hadoop3
15. port: 3306
17. username: root
19. password: "123456"
21. tables: test.\.
23. server-id: 5400-5404
25. server-time-zone: UTC+8
29. sink:
31. type: doris
33. fenodes: hadoop1:7030
35. username: root
37. password: "123456"
39. table.create.properties.light_schema_change: true
41. table.create.properties.replication_num: 1
45. pipeline:
47. name: Sync MySQL Database to Doris
49. parallelism: 1

复制代码

3）启动任务并测试
（1）开启Flink集群，注意：在Flink设置信息中打开CheckPoint

1. [root@hadoop3 flink-1.18.0]$ vim conf/flink-conf.yaml

复制代码

添加如下设置信息

1. execution.checkpointing.interval: 5000

复制代码

分发该文件至其他Flink呆板并启动Flink集群

1. [root@hadoop3 flink-1.18.0]$ bin/start-cluster.sh

复制代码

（2）开启Doris FE

1. [root@hadoop1 fe]$ bin/start_fe.sh

复制代码

（3）开启Doris BE

1. [root@hadoop1 be]$ bin/start_be.sh

复制代码

（4）在Doris中创建test数据库

1. [root@hadoop3 doris]$ mysql -uroot -p123456 -P9030 -hhadoop1
3. mysql: [Warning] Using a password on the command line interface can be insecure.
5. Welcome to the MySQL monitor.  Commands end with ; or g.
7. Your MySQL connection id is 0
9. Server version: 5.7.99 Doris version doris-1.2.4-1-Unknown
13. Copyright (c) 2000, 2022, Oracle and/or its affiliates.
17. Oracle is a registered trademark of Oracle Corporation and/or its
19. affiliates. Other names may be trademarks of their respective
21. owners.
25. Type 'help;' or 'h' for help. Type 'c' to clear the current input statement.
29. mysql> create database test;
31. Query OK, 1 row affected (0.00 sec)
35. mysql> create database doris_test_route;
37. Query OK, 1 row affected (0.00 sec)

复制代码

（5）启动FlinkCDC同步变动任务

1. [root@hadoop3 flink-cdc-3.0.0]$ bin/flink-cdc.sh job/mysql-to-doris.yaml

复制代码

（6）刷新Doris中test数据库观察结果
（7）在MySQL的test数据中对应的几张表举行新增、修改数据以及新增列操作，并刷新Doris中test数据库观察结果
3 路由变动

1）编写MySQL到Doris的路由变动设置文件

1. source:
3. type: mysql
5. hostname: hadoop3
7. port: 3306
9. username: root
11. password: "123456"
13. tables: test_route.\.
15. server-id: 5400-5404
17. server-time-zone: UTC+8
21. sink:
23. type: doris
25. fenodes: hadoop1:7030
27. benodes: hadoop1:7040
29. username: root
31. password: "123456"
33. table.create.properties.light_schema_change: true
35. table.create.properties.replication_num: 1
39. route:
41. - source-table: test_route.t1
43.   sink-table: doris_test_route.doris_t1
45. - source-table: test_route.t2
47.   sink-table: doris_test_route.doris_t1
49. - source-table: test_route.t3
51.   sink-table: doris_test_route.doris_t3
55. pipeline:
57. name: Sync MySQL Database to Doris
59. parallelism: 1

复制代码

2）启动任务并测试

1. [root@hadoop3 flink-cdc-3.0.0]$ bin/flink-cdc.sh job/mysql-to-doris-route.yaml

复制代码

3）刷新Doris中test_route数据库观察结果
4）在MySQL的test_route数据中对应的几张表举行新增、修改数据操作，并刷新Doris中doris_test_route数据库观察结果

免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！更多信息从访问主页：qidao123.com:ToB企服之家，中国第一个企服评测及商务社交产业平台。