杰霸霸博客

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
最后更新于: 2025-03-01 15:54

评论