跳到主要内容
版本: 1.1

MySQLCDC 整库到 Doris

前置准备

  • 请确保已经在 Flink/lib 和 dinky/extends 目录下放置了 Doris 的 Flink connector jar。如果提交模式为 Application/Per-Job,请确保 Doris connector jar 已经放置在 HDFS 中
  • 请确保已经在 Flink/lib 和 dinky/extends 目录下放置了 MySQL CDC 的 Flink connector jar。 如果提交模式为 Application/Per-Job,请确保 MySQL CDC connector jar 已经放置在 HDFS 中
  • 如在两方启动后才进行放置上述 jar 包,请重启 Flink 和 Dinky 服务,或者使用 Dinky 中提供的 ADD CUSTOMJAR 功能进行加载。

示例

注意事项:

  • 该示例是将 mysql 整库同步到 Doris 表,且写入名为 ods 的库,目标表名前缀取 test__ 并转小写。
  • 该示例参数中的 #{tableName} 为占位符,实际执行时会替换为实际表名,如 ods_productsods_orders 等。
  • 该示例参数中的 #{schemaName} 为占位符,实际执行时会替换为实际库名,如 test 等。
  • 该示例 sink 中的各个参数均可根据实际情况进行调整,请按照 Doris 连接器官方文档进行配置。并请遵守整库同步的规范.

普通同步

Doris 的 Flink 连接器参数随版本变化较大,以下为 Doris 1.2.0 版本的参数配置。

每次提交作业都需要手动修改 'sink.sink.label-prefix' = '#{schemaName}_#{tableName}_1' 的值,比如改变尾部的数值。

EXECUTE CDCSOURCE demo_doris WITH (
'connector' = 'mysql-cdc',
'hostname' = '127.0.0.1',
'port' = '3306',
'username' = 'root',
'password' = '123456',
'checkpoint' = '10000',
'scan.startup.mode' = 'initial',
'parallelism' = '1',
'table-name' = 'test\.student,test\.score',
'sink.connector' = 'doris',
'sink.fenodes' = '127.0.0.1:8030',
'sink.username' = 'root',
'sink.password' = '123456',
'sink.doris.batch.size' = '1000',
'sink.sink.max-retries' = '1',
'sink.sink.batch.interval' = '60000',
'sink.sink.db' = 'test',
'sink.sink.properties.format' ='json',
'sink.sink.properties.read_json_by_line' ='true',
'sink.table.identifier' = '#{schemaName}.#{tableName}',
'sink.sink.label-prefix' = '#{schemaName}_#{tableName}_1'
);

字段模式演变

自动同步列新增和删除列,库表名需要与源库相同。

EXECUTE CDCSOURCE demo_doris_schema_evolution WITH (
'connector' = 'mysql-cdc',
'hostname' = '127.0.0.1',
'port' = '3306',
'username' = 'root',
'password' = '123456',
'checkpoint' = '10000',
'scan.startup.mode' = 'initial',
'parallelism' = '1',
'table-name' = 'test\.student,test\.score',
'sink.connector' = 'datastream-doris-schema-evolution',
'sink.fenodes' = '127.0.0.1:8030',
'sink.username' = 'root',
'sink.password' = '123456',
'sink.doris.batch.size' = '1000',
'sink.sink.max-retries' = '1',
'sink.sink.batch.interval' = '60000',
'sink.sink.db' = 'test',
'sink.table.identifier' = '#{schemaName}.#{tableName}'
);