Dinky运行前开启全局变量，以支持使用：  'sink.sink.label-prefix' = '${idUtil.simpleUUID()}'

Mysql同步Doris - testMysqlCdcDoris：

EXECUTE CDCSOURCE demo_doris WITH ('connector' = 'mysql-cdc','hostname' = '172.xxx','port' = '3306','username' = 'xxx','password' = 'xxx','checkpoint' = '10000','scan.startup.mode' = 'initial','parallelism' = '1','database-name' = 'test','table-name' = 'test\.student,','sink.connector' = 'doris','sink.fenodes' = '172.xxx:8130','sink.username' = 'xxx','sink.password' = 'xxx','sink.doris.batch.size' = '1000','sink.sink.max-retries' = '1','sink.sink.db' = 'test','sink.sink.enable-delete' = 'true','sink.sink.properties.format' ='json','sink.sink.properties.read_json_by_line' ='true','sink.table.prefix' = 'test_','sink.table.identifier' = '#{schemaName}.#{tableName}','sink.sink.label-prefix' = '${idUtil.simpleUUID()}'
);

读取Doris - testDorisRead:

CREATE TABLE flink_doris_source (aggregate_id int,replace_data string,max_data string,agg_item int,max_item int,min_item int
) 
WITH ('connector' = 'doris','fenodes' = '172.xxx:8130','table.identifier' = 'test.aggregate_table','username' = 'xxx','password' = 'xxx'
);select * from flink_doris_source

Doris同步Doris - testDorisCdcDoris:

-- doris source
CREATE TABLE flink_doris_source (aggregate_id int,replace_data string,max_data string,agg_item int,max_item int,min_item int
) 
WITH ('connector' = 'doris','fenodes' = '172.xxx:8130','table.identifier' = 'test.aggregate_table','username' = 'xxx','password' = 'xxx'
);-- enable checkpoint
SET 'execution.checkpointing.interval' = '10s';-- doris sink
CREATE TABLE flink_doris_sink (aggregate_id int,replace_data string,max_data string,agg_item int,max_item int,min_item int) WITH ('connector' = 'doris','fenodes' = '172.xxx:8030','table.identifier' = 'test.test_aggregate_table','username' = 'xxx','password' = 'xxx','sink.label-prefix' = '${idUtil.simpleUUID()}'
);-- submit insert job
INSERT INTO flink_doris_sink select aggregate_id, replace_data, max_data, agg_item, max_item, min_item from flink_doris_source

参考

Flink Doris Connector - Apache Doris

Doris + Flink + DolphinScheduler + Dinky 构建开源数据平台_dinky dolphinscheduler flink-CSDN博客

整库同步概述 | Dinky