整库同步（SQL）能力

CREATE DATABASE IF NOT EXISTS <target_database>
[COMMENT database_comment] 
[WITH (key1=val1, key2=val2, ...)] -- 指明写入目标库的参数
AS DATABASE <source_catalog>.<source_database> 	-- source_database 是被同步的源数据库
INCLUDING { ALL TABLES | TABLE 'table_name' }
-- INCLUDING ALL TABLES 表示同步数据库中的所有表
-- INCLUDING TABLE 'table' 表示同步数据库中特定的表，支持正则表达式，如 'order_.*';
-- 同步多张表时，可以写成 INCLUDING TABLE 'tableA|tableB|tableC'的格式
[EXCLUDING TABLE 'table_name']
-- EXCLUDING TABLE 'table' 表示不同步数据库中特定的表，支持正则表达式，如 'order_.*';
-- 排除多张表时，可以写成 EXCLUDING TABLE 'tableA|tableB|tableC'的格式
[/*+ `OPTIONS`('key1'='val1', 'key2'='val2', ... ) */] 
-- （可选，指明读取source的参数，如指定source serverId的范围，解析debezium时间戳字段类型等）

create catalog my_mysql with(...);
create database if not exists sink_db
with (
  'connector' = 'doris',
  'table.identifier' = 'db1.$tableName_doris'
  ...
) 
including all tables
/*+ `OPTIONS`('server-time-zone' = 'Asia/Shanghai') */; -- 声明解析timestamp字段的时区类型

-- 必须设置 table.optimizer.deterministic-operator-uid-for-cdas=true 才可以启用无损加表能力
SET table.optimizer.deterministic-operator-uid-for-cdas=true;

-- 可选：开启多 Source 复用，降低资源占用，提升稳定性
SET table.optimizer.mysql-cdc-source.merge.enabled=true;

-- 请替换成实际的 CDAS 语句，但至少要保留 'scan.newly-added-table.enabled' = 'true' 的 OPTIONS 选项
create catalog my_mysql with(...);
create database if not exists sink_db
with (
  ...
) 
including all tables
/*+ `OPTIONS`('scan.newly-added-table.enabled' = 'true','scan.lastchunk.optimize.enable' = 'true') */;

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

create catalog my_mysql with (
    'type' = 'jdbc',
    'default-database' = 'test',
    'username' = 'root',
    'password' = 'XXX',
    'base-url' = 'jdbc:mysql://ip:port'
);

create database if not exists sink_db
comment 'test_sink'
with (
'connector' = 'hudi',
'path' = 'hdfs://namenode:8020/user/hive/warehouse/$tableName_mor'
) as database `my_mysql`.`trade_log`
including all tables
/*+ `OPTIONS`('server-time-zone' = 'Asia/Shanghai') */; -- 声明解析timestamp字段的时区类型

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

create catalog my_mysql with (
    'type' = 'jdbc',
    'default-database' = 'test',
    'username' = 'root',
    'password' = 'XXX',
    'base-url' = 'jdbc:mysql://ip:port'
);

create database if not exists sink_db
comment 'test_sink'  with (
'connector' = 'clickhouse://172.11.11.11:8123',
 -- 如果ClickHouse集群未配置账号密码可以不指定
 --'username' = 'root',                            -- ClickHouse集群用户名
 --'password' = 'root',                            -- ClickHouse集群的密码
 'database-name' = 'testdb',                       -- 数据写入目的数据库
 'table-name' = 'test_table1',                     -- 数据写入目的数据表
 'sink.batch-size' = '1000',
 'table.collapsing.field' = 'Sign'
) as database my_mysql.test_db
including all tables
/*+ OPTIONS('server-time-zone' = 'Asia/Shanghai') */;

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

create catalog my_mysql with (
    'type' = 'jdbc',
    'default-database' = 'test',
    'username' = 'root',
    'password' = 'XXX',
    'base-url' = 'jdbc:mysql://ip:port'
);

create database if not exists sink_db
comment 'test_sink'  with (
'connector' = 'doris',
'table.identifier' = 'trade_log.$tableName',
'username' = 'admin',
'password' = 'xxx',
'sink.batch.size' = '500',
'sink.batch.interval' = '1s',
'fenodes' = 'ip:port'
) as database `my_mysql`.`trade_log`
including all tables
/*+ `OPTIONS`('server-time-zone' = 'Asia/Shanghai') */; -- 声明解析timestamp字段的时区类型

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

create catalog my_mysql with (
    'type' = 'jdbc',
    'default-database' = 'test',
    'username' = 'root',
    'password' = 'XXX',
    'base-url' = 'jdbc:mysql://ip:port'
);

create database if not exists sink_db
comment 'test_sink'  with (
'connector' = 'hive',
'hive-version' = '2.3.6',
'hive-database' = 'test_100',
'hive-table' = '$tableName',
'sink.partition-commit.policy.kind' = 'metastore'
) as database `my_mysql`.`trade_log`
including all tables
/*+ `OPTIONS`('append-mode' = 'true', 'server-time-zone' = 'Asia/Shanghai') */;  
-- 因为hive sink不支持变更数据，此处的hint会把原始cdc的变更数据转成成append流下发

SET  table.optimizer.mysql-cdc-source.merge.enabled = true;
SET table.optimizer.deterministic-operator-uid-for-cdas=true;

-- 注册mysql的catalog
create catalog `my_mysql` with (
  'type' = 'jdbc',
  'default-database' = 'testdb', -- 数据库名
  'username' = '${username}', -- mysql用户名
  'password' = '${password}', -- mysql密码
  'base-url' = 'jdbc:mysql://ip:3306'
);

-- 整库同步
create database if not exists `my_dlc_database`
    comment 'test db sync'
with (
    'connector' = 'iceberg-inlong', -- 固定值
    'catalog-database' = 'test', -- DLC内表所在的数据库名称
    'catalog-table' = 'my_$tableName', -- DLC内表名称，$tableName会自动替换为要同步的表名，DLC表需要提前创建
    'default-database' = 'test', -- DLC内表所在的数据库名称
    'catalog-name' = 'HYBRIS', -- 固定值
    'catalog-impl' = 'org.apache.inlong.sort.iceberg.catalog.hybris.DlcWrappedHybrisCatalog', -- 固定值
    'qcloud.dlc.managed.account.uid' = '100026378089', -- 固定值，DLC管理账号的uid
    'qcloud.dlc.secret-id' = '${secret_id}', -- DLC 用户的secretId从https://console.cloud.tencent.com/cam/capi中获取
    'qcloud.dlc.secret-key' = '${secret_key}', -- DLC 用户的secretKey，从https://console.cloud.tencent.com/cam/capi中获取
    'qcloud.dlc.region' = 'ap-guangzhou', -- DLC 所在地域，必须填ap-地域格式
    'qcloud.dlc.jdbc.url' = 'jdbc:dlc:dlc.internal.tencentcloudapi.com?task_type=SparkSQLTask&database_name=test&datasource_connection_name=DataLakeCatalog&region=ap-guangzhou&data_engine_name=${engine_name}', -- DLC jdbc接入url，格式见https://cloud.tencent.com/document/product/1342/61547
    'uri' = 'dlc.internal.tencentcloudapi.com', -- 固定值
    'user.appid' = '${appid}', -- DLC 用户的 appid
    'request.identity.token' = '100026378089' -- 固定值，DLC内表接入的token
)
    as database `my_mysql`.`test` including  table 'user.*'
    excluding table 'user_info_20230530|user_behavior_tmp'
/* +`OPTIONS`(
 'server-time-zone' = 'Asia/Shanghai',
 'scan.newly-added-table.enabled' = 'true',
 'scan.lastchunk.optimize.enable' = 'true'
*/
;