杰霸霸博客

主题

MySQL到Doris的StreamingETL实现 ​

1. 环境准备 ​

1.1 安装FlinkCDC ​

sh
[jack@hadoop102 software]$ tar -xvf flink-cdc-3.3.0-bin.tar.gz -C ../module/

2. 同步变更 ​

2.1 编写MySQL到Doris的同步变更配置文件 ​

在FlinkCDC目录下创建job文件夹，并在该目录下创建同步变更配置文件mysql-to-doris.yaml

yml
source:
  type: mysql
  hostname: hadoop103
  port: 3306
  username: root
  password: "root"
  tables: test.\.*
  server-id: 5400-5404
  server-time-zone: UTC+8

sink:
  type: doris
  fenodes: hadoop102:7030
  username: root
  password: "root"
  table.create.properties.light_schema_change: true
  table.create.properties.replication_num: 1

pipeline:
  name: Sync MySQL Database to Doris
  parallelism: 1

2.2 启动任务并测试 ​

开启Flink集群，注意：在Flink配置信息中打开CheckPoint

sh
## 启动FlinkCDC同步变更任务
bin/flink-cdc.sh job/mysql-to-doris.yaml

3. 路由变更 ​

  1. 编写MySQL到Doris的路由变更配置文件
yml
source:
  type: mysql
  hostname: hadoop103
  port: 3306
  username: root
  password: "000000"
  tables: test_route.\.*
  server-id: 5400-5404
  server-time-zone: UTC+8

sink:
  type: doris
  fenodes: hadoop102:7030
  benodes: hadoop102:7040
  username: root
  password: "000000"
  table.create.properties.light_schema_change: true
  table.create.properties.replication_num: 1

route:
  - source-table: test_route.t1
    sink-table: doris_test_route.doris_t1
  - source-table: test_route.t2
    sink-table: doris_test_route.doris_t1
  - source-table: test_route.t3
    sink-table: doris_test_route.doris_t3

pipeline:
  name: Sync MySQL Database to Doris
  parallelism: 1