开发 SQL 作业

--mysql cdc 源表Data_In
CREATE TABLE `Data_In` (
  `Id`     INT NOT NULL,
  `Str_A`   VARCHAR,
  `Str_B`   VARCHAR,
  PRIMARY KEY (`Id`) NOT ENFORCED
) WITH (
  'connector' = 'mysql-cdc',           -- 固定值 'mysql-cdc'
  'hostname' = 'YourHostName',         -- 数据库的 IP
  'port' = '3306',                     -- 数据库的访问端口
  'username' = 'YourUserName',         -- 数据库访问的用户名（需要提供 SHOW DATABASES、REPLICATION SLAVE、REPLICATION CLIENT、SELECT 和 RELOAD 权限）
  'password' = 'YourPassword',         -- 数据库访问的密码
  'database-name' = 'YourDatabase',    -- 需要同步的数据库
  'table-name' = 'YourTable'           -- 需要同步的数据表名     
);

--写入doris表Data_Out
CREATE TABLE `Data_Out` (
  `Id`       INT,
  `Str_A`     STRING,
  `Str_B`      STRING,
  PRIMARY KEY (`Id`) NOT ENFORCED
) WITH (
  'connector' = 'doris',                    -- 固定值 'doris'
  'fenodes' = 'FE_IP:FE_RESFUL_PORT',       -- Doris FE http 地址
  'table.identifier' = 'dbName.tableName',  -- Doris 表名 格式：db.tbl
  'username' = 'YourUserName',              -- 访问Doris的用户名，拥有库的写权限
  'password' = 'YourPassword',              -- 访问Doris的密码
  'sink.batch.size' = '500',                -- 单次写BE的最大行数
  'sink.batch.interval' = '1s'              -- flush 间隔时间，超过该时间后异步线程将 缓存中数据写入BE。 默认值为1秒，支持时间单位ms、s、min、h和d。设置为0表示关闭定期写入。
);

insert into Data_Out
select Id,Str_A,Str_B from Data_In;

1,Str_A_1,Str_B_1
2,Str_A_2,Str_B_2
3,Str_A_3,Str_B_3
4,Str_A_4,Str_B_4
5,Str_A_5,Str_B_5