tencent cloud

CREATE TABLE `source_1`
(
    `f_sequence`   INT,
    `f_random`     INT,
    `f_random_str` VARCHAR,
    PRIMARY KEY (`f_sequence`) NOT ENFORCED
) WITH (
      'connector' = 'mysql-cdc' ,
      'hostname' = 'ip1',
      'port' = '3306',
      'username' = 'xxx',
      'password' = 'xxx',
      'database-name' = 'db1',
      'table-name' = 'source_1'
);

CREATE TABLE `source_2`
(
    `f_sequence`   INT,
    `f_random`     INT,
    `f_random_1`     INT,
    `f_random_str` VARCHAR,
    `f_random_str_1` VARCHAR,
    PRIMARY KEY (`f_sequence`) NOT ENFORCED
) WITH (
      'connector' = 'mysql-cdc' ,
      'hostname' = 'ip1',
      'port' = '3306',
      'username' = 'xxx',
      'password' = 'xxx',
      'database-name' = 'db2',
      'table-name' = 'source_2'
);


CREATE TABLE `sink_1`
(
    `f_sequence`   INT,
    `f_random`     INT,
    `f_random_str` VARCHAR,
    PRIMARY KEY (`f_sequence`) NOT ENFORCED
) WITH (
      'connector' = 'logger'
);

CREATE TABLE `sink_2`
(
    `f_sequence`   INT,
    `f_random`     INT,
    `f_random_1`     INT,
    `f_random_str` VARCHAR,
    `f_random_str_1` VARCHAR,
    PRIMARY KEY (`f_sequence`) NOT ENFORCED
) WITH (
      'connector' = 'logger'
);

insert into sink_1 select * from source_1;
insert into sink_2 select * from source_2;

SET table.optimizer.mysql-cdc-source.merge.enabled=true;

SET table.optimizer.mysql-cdc-source.merge.enabled=true;
SET table.optimizer.mysql-cdc-source.merge.default-group.splits=2;

insert into sink_1 select * from source_1;
insert into sink_2 select * from source_2;
insert into sink_3 select * from source_3;
insert into sink_4 select * from source_4;