云数据仓库 TCHouse-P

CREATE TABLE `jdbc_sink_table` (
    `id` INT,
    `name` STRING
) WITH (
    -- 指定数据库连接参数
    'connector' = 'jdbcPG',
    'url' = 'jdbc:postgresql://10.1.28.93:3306/CDB?reWriteBatchedInserts=true&serverTimezone=Asia/Shanghai', -- 请替换为您的实际 PG 连接参数
    'table-name' = 'my-table',  -- 需要写入的数据表
    'username' = 'admin',       -- 数据库访问的用户名（需要提供 INSERT 权限）
    'password' = 'MyPa$$w0rd',  -- 数据库访问的密码
    'sink.buffer-flush.max-rows' = '200',  -- 批量输出的条数
    'sink.buffer-flush.interval' = '2s'    -- 批量输出的间隔
);

CREATE TABLE `jdbc_upsert_sink_table` (
    `id` INT PRIMARY KEY NOT ENFORCED,
    `name` STRING
) WITH (
    -- 指定数据库连接参数
    'connector' = 'jdbcPG',
    'url' = 'jdbc:postgresql://10.1.28.93:3306/CDB?reWriteBatchedInserts=true&serverTimezone=Asia/Shanghai', -- 请替换为您的实际 PG 连接参数
    'table-name' = 'my-upsert-table', -- 需要写入的数据表
    'username' = 'admin',             -- 数据库访问的用户名（需要提供 INSERT 权限）
    'password' = 'MyPa$$w0rd',        -- 数据库访问的密码
    'sink.buffer-flush.max-rows' = '200',  -- 批量输出的条数
    'sink.buffer-flush.interval' = '2s'    -- 批量输出的间隔
);

CREATE TABLE jdbc_upsert_sink_table (
  id INT ,
  age INT,
  name STRING,
  PRIMARY KEY (id) NOT ENFORCED -- 如果要同步的数据库表定义了主键, 则这里也需要定义
) WITH (
    -- 指定数据库连接参数
    'connector' = 'jdbcPG',
    'url' = 'jdbc:postgresql://10.0.0.2:5436/postgres', 
    'table-name' = 'my-upsert-table', -- 需要写入的数据表
    'username' = 'admin',             -- 数据库访问的用户名（需要提供 INSERT 权限）
    'password' = 'password',        -- 数据库访问的密码
    'sink.buffer-flush.max-rows' = '300',  -- 批量输出的条数
    'sink.buffer-flush.interval' = '100s'    -- 批量输出的间隔
);
?
?
-- MySQL CDC Source。配合 flink-connector-mysql-cdc 使用。要求 MySQL 版本 >= 5.7
-- 参见 /document/product/849/52698
CREATE TABLE mysql_cdc_source_table (
  id INT,
  age INT,
  name STRING,
  PRIMARY KEY (id) NOT ENFORCED -- 如果要同步的数据库表定义了主键, 则这里也需要定义
) WITH (
  'connector' = 'mysql-cdc',	  -- 固定值 'mysql-cdc'
  'hostname' = '10.0.0.6',   -- 数据库的 IP
  'port' = '3360',                -- 数据库的访问端口
  'username' = 'root',        -- 数据库访问的用户名（需要提供 SHOW DATABASES、REPLICATION SLAVE、REPLICATION CLIENT、SELECT 和 RELOAD 权限）
  'password' = 'password',    -- 数据库访问的密码
  -- 'scan.incremental.snapshot.enabled' = 'false' -- 如果 source 表没有设置 PRIMARY Key，需要启用该设置 
  'database-name' = 'database',   -- 需要同步的数据库
  'table-name' = 'table'      -- 需要同步的数据表名
);
?
?
INSERT INTO jdbc_upsert_sink_table select * from mysql_cdc_source_table;