Flink-1.14-cdc

基于flink1.14、flink-cdc2.2.1版本，支持mysql-cdc、oracle-cdc、sqlserver-cdc、tidb-cdc

mysql-cdc

示例1(原生sql)

CREATE TABLE sink (
    test_int INT,
  test_varchar VARCHAR,
  test_boolean BOOLEAN,
  test_tinyint TINYINT,
  test_smallint SMALLINT,
  test_bigint BIGINT,
  test_float FLOAT,
  test_double DOUBLE,
  PRIMARY KEY(test_int) NOT ENFORCED
) WITH (
    'connector' = 'jdbc', -- required: specify this table type is jdbc
    'url' = 'jdbc:mysql://localhost:3306/qa_slothdata', -- required: JDBC DB url
    'table-name' = 'alltype1',  -- required: jdbc table name
    'username' = 'root', -- required: jdbc user name and password
    'password' = 'root'
);

CREATE TABLE source (
  test_int INT,
  test_varchar VARCHAR,
  test_boolean BOOLEAN,
  test_tinyint TINYINT,
  test_smallint SMALLINT,
  test_bigint BIGINT,
  test_float FLOAT,
--  testDecimal DECIMAL(38, 18),
  test_double DOUBLE,
  PRIMARY KEY(test_int) NOT ENFORCED
) WITH (
    'connector' = 'mysql-cdc',
    'hostname' = 'localhost',
    'port' = '3306',
    'database-name' = 'qa_slothdata',
    'table-name' = 'alltype',  -- required: jdbc table name
    'username' = 'root', -- required: jdbc user name and password
    'password' = 'root'
);

INSERT INTO sink select * FROM source;

示例1参数

示例2(无ddl,metahub方式)

isCDC参数为必填项,其余参数可以根据需要自行补充，参考示例1参数配置项

set 'alltype.isCDC' = 'true'; -- required
set 'alltype.scan.startup.mode' = 'latest-offset';  -- optional
set 'alltype1.primary.keys'='test_int';

INSERT INTO sloth_test_qa_mysql_167.qa_slothdata.alltype1 select * FROM sloth_test_qa_mysql_167.qa_slothdata.alltype;

oracle-cdc

示例1(原生sql)

create table source (
    id  decimal(38,0),
    col_char    string,
    col_varchar2    string,
    col_nchar    string,
    col_nvarchar2    string,
    col_number    decimal(10,5),
    col_number_bigint    decimal(19,0),
    col_date    timestamp,
    col_decimal    decimal(20,5),
    col_smallint    decimal(22,0),
    col_integer    decimal(22,0),
    col_binary_float    float,
    col_binary_double    double,
    col_real    float,
    col_float    float,
    col_timestamp    timestamp,
    col_decimal_bigint    decimal(19,0),
  PRIMARY KEY(id) NOT ENFORCED
) WITH (
    'connector' = 'oracle-cdc', -- required: specify this table type is oracle-cdc
    'hostname' = 'localhost',
    'port' = '11521',
    'table-name' = 'ALL_FIELD',  -- required: jdbc table name
    'username' = 'root', -- required: jdbc user name and password
    'password' = 'password',
    'database-name' = 'orcl',
    'schema-name' = 'NDC',
    'debezium.database.tablename.case.insensitive' = 'false',
    'debezium.database.connection.adapter' = 'logminer',
    'debezium.log.mining.strategy' = 'online_catalog',
    'debezium.log.mining.continuous.mine' = 'true'
);

create table target (
    id    decimal(38,0),
    col_char    string,
    col_varchar2    string,
    col_nchar    string,
    col_nvarchar2    string,
    col_number    decimal(10,5),
    col_number_bigint    decimal(19,0),
    col_date    timestamp,
    col_decimal    decimal(20,5),
    col_smallint    decimal(22,0),
    col_integer    decimal(22,0),
    col_binary_float    float,
    col_binary_double    double,
    col_real    float,
    col_float    float,
    col_timestamp    timestamp,
    col_decimal_bigint    decimal(19,0),
    PRIMARY KEY(id) NOT ENFORCED
) WITH (
  'connector' = 'kafka',
  'topic' = 'oracle_xstream_all_field',
  'properties.bootstrap.servers' = 'localhost:9092',
  'format' = 'canal-json'
);

insert into target select * from source;

示例1参数

示例2(无ddl语句,metahub方式)

isCDC参数为必填项,其余参数可以根据需要自行补充，参考示例1参数配置项

-- 源端oracle-cdc
set 'WH_TABLE1.isCDC' = 'true';  -- required
set 'WH_TABLE1.schema-name' = 'NDC';
set 'WH_TABLE1.scan.startup.mode' = 'latest-offset';
set 'WH_TABLE1.debezium.database.tablename.case.insensitive' = 'false';
set 'WH_TABLE1.debezium.database.connection.adapter' = 'logminer';
set 'WH_TABLE1.debezium.log.mining.strategy' = 'online_catalog';
set 'WH_TABLE1.debezium.log.mining.continuous.mine' = 'true';

-- 目标端kafka 配置
set 'target_table.topic' = 'ingestion_wanghua_01';
set 'target_table.key.format' = 'csv';
set 'target_table.key.fields' = 'ID';
set 'target_table.format' = 'canal-json';

INSERT INTO poc_kafka_x_catalog.mem.target_table select * FROM dts.NDC.WH_TABLE1;

sqlserver-cdc

示例1(原生sql)

create table source (
    id  bigint,
    name string,
    age int,
  PRIMARY KEY(id) NOT ENFORCED
) WITH (
    'connector' = 'sqlserver-cdc', -- required: specify this table type is sqlserver-cdc
    'hostname' = 'localhost',
    'port' = '1433',
    'table-name' = 'table_9',  -- required: jdbc table name
    'username' = 'root', -- required: jdbc user name and password
    'password' = 'root',
    'database-name' = 'dtstest',
    'schema-name' = 'dbo'
);

create table target (
     id  bigint,
    name string,
    age int
) WITH (
  'connector' = 'kafka',
  'topic' = 'lyb1',
  'properties.bootstrap.servers' = 'localhost:9092',
  'format' = 'canal-json'
);

insert into target select * from source;

示例1参数

示例2(无ddl语句,metahub方式)

isCDC参数为必填项,其余参数可以根据需要自行补充，参考示例1参数配置项

set 'table_polling_1.isCDC' = 'true'; -- required
set 'table_polling_1.schema-name' = 'dbo';
set 'table_polling_1.scan.startup.mode' = 'latest-offset';

set 'target_table.topic' = 'lyb1';
set 'target_table.key.format' = 'csv';
set 'target_table.key.fields' = 'id';
set 'target_table.format' = 'canal-json';


INSERT INTO poc_kafka_x_catalog.mem.target_table select * FROM poc_sqlserver.dbo.table_polling_1;

tidb-cdc

示例1(原生sql)

CREATE TABLE sink (
  id INT,
  age INT,
  name VARCHAR,
  score double,
  PRIMARY KEY(id) NOT ENFORCED
) WITH (
    'connector' = 'jdbc',
    'url' = 'jdbc:mysql:/localhost:4000/test', -- required: JDBC DB url
    'table-name' = 'easyds_test1',  -- required: jdbc table name
    'username' = 'root', -- required: jdbc user name and password
    'password' = 'root'
);

CREATE TABLE source (
  id INT,
  age INT,
  name VARCHAR,
  score double,
  PRIMARY KEY(id) NOT ENFORCED
) WITH (
    'connector' = 'tidb-cdc', -- required: specify this table type is tidb-cdc
    'pd-addresses' = localhost:2390', -- required: pd-addresses
    'table-name' = 'root',  -- required:  table name
    'database-name' = 'root' -- required:  database
);

INSERT INTO sink select * FROM source;

示例1参数

示例2(无ddl语句,metahub方式)

• isCDC参数为必填项,其余参数可以根据需要自行补充，参考示例1参数配置项
• methhub方式在登记tidb数据源时候，需要在自定义参数中增加 "pd-addresses"参数，否则该方式无法正常使用

set 'easyds_test.isCDC' = 'true';
set 'tikv.grpc.timeout_in_ms' = '20000';
set 'easyds_test1.primary.keys'='id';

INSERT INTO test_tidb.test.easyds_test1 select * FROM test_tidb.test.easyds_test;