Dinky FlinkSQL Doris读取写入

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

参考