tencent cloud

CREATE TABLE `jdbc_sink_table` (
    `id` INT,
    `name` STRING
) WITH (
    -- Specify the options for database connection.
    'connector' = 'jdbcPG',
    'url' = 'jdbc:postgresql://10.1.28.93:3306/CDB?reWriteBatchedInserts=true&serverTimezone=Asia/Shanghai', -- Replace it with your PostreSQL database URL.
    'table-name' = 'my-table',  -- The name of JDBC table to connect.
    'username' = 'admin',       -- The username (with the INSERT permission required) for database access.
    'password' = 'MyPa$$w0rd',  -- The password for database access.
    'sink.buffer-flush.max-rows' = '200',  -- The max size of buffered records before flush.
    'sink.buffer-flush.interval' = '2s'    -- The flush interval.
);

CREATE TABLE `jdbc_upsert_sink_table` (
    `id` INT PRIMARY KEY NOT ENFORCED,
    `name` STRING
) WITH (
    -- Specify the options for database connection.
    'connector' = 'jdbcPG',
    'url' = 'jdbc:postgresql://10.1.28.93:3306/CDB?reWriteBatchedInserts=true&serverTimezone=Asia/Shanghai', -- Replace it with your PostreSQL connection URL.
    'table-name' = 'my-upsert-table', -- The name of JDBC table to connect.
    'username' = 'admin',             -- The username (with the INSERT permission required) for database access.
    'password' = 'MyPa$$w0rd',        -- The password for database access.
    'sink.buffer-flush.max-rows' = '200',  -- The max size of buffered records before flush.
    'sink.buffer-flush.interval' = '2s'    -- The flush interval.
);

CREATE TABLE jdbc_upsert_sink_table (
  id INT ,
  age INT,
  name STRING,
  PRIMARY KEY (id) NOT ENFORCED -- Define the primary key here if the database table to monitor has it defined.
) WITH (
    -- Specify the options for database connection.
    'connector' = 'jdbcPG',
    'url' = 'jdbc:postgresql://10.0.0.2:5436/postgres', 
    'table-name' = 'my-upsert-table', -- The name of JDBC table to connect.
    'username' = 'admin',             -- The username (with the INSERT permission required) for database access.
    'password' = 'password',        -- The password for database access.
    'sink.buffer-flush.max-rows' = '300',  -- The max size of buffered records before flush.
    'sink.buffer-flush.interval' = '100s'    -- The flush interval.
);


-- MySQL CDC connector as the source. Use "flink-connector-mysql-cdc" and this connector together. The MySQL version must be v5.7 or later.
For more information, see: https://www.tencentcloud.com/document/product/849/52698?from_cn_redirect=1.
CREATE TABLE mysql_cdc_source_table (
  id INT,
  age INT,
  name STRING,
  PRIMARY KEY (id) NOT ENFORCED -- Define the primary key here if the database table to monitor has it defined.
) WITH (
  'connector' = 'mysql-cdc',	  -- Here, it should be 'mysql-cdc'.
  'hostname' = '10.0.0.6',   -- IP of the MySQL database server.
  'port' = '3360',                --  Integer port number of the MySQL database server.
  'username' = 'root',        -- Name of the database to use when connecting to the MySQL database server (SHOW DATABASES, REPLICATION SLAVE, REPLICATION CLIENT, SELECT, and RELOAD permissions are required).
  'password' = 'password',    -- Password to use when connecting to the MySQL database server.
  -- 'scan.incremental.snapshot.enabled' = 'false' -- This option is required if the source table has no primary key. 
  'database-name' = 'database',   -- Database name of the MySQL server to monitor.
  'table-name' = 'table'      -- Table name of the MySQL database to monitor.
);


INSERT INTO jdbc_upsert_sink_table select * from mysql_cdc_source_table;