CREATE DATABASE AS（CDAS）语句

数据同步

表结构变更同步

在实时同步整库数据的同时，还支持将每张源表的表结构变更（加列等）实时同步到结果表中。CDAS的表结构变更同步策略与CTAS一致，详情请参见表结构变更同步。

语法限制

• 不支持调试功能。

• 不支持MiniBatch配置。

  重要

  • 创建SQL作业前：请确保配置管理页面的作业默认配置页签中的其他配置处删除了MiniBatch配置。

  • 创建SQL作业后：具体解决方案可参见报错：Currently does not support merge StreamExecMiniBatchAssigner type ExecNode in CTAS/CDAS syntax。

上下游存储兼容性

CDAS支持的上下游存储列表如下，您可以从下表的源表和结果表中各选一个进行组合。

整库同步

同步场景：将MySQL中的tpcds库下的所有表同步到Hologres。

前提条件：已在工作空间中注册以下Catalog。

• Hologres Catalog：名称为holo。

• MySQL Catalog：名称为mysql。

代码示例：

USE CATALOG holo;

CREATE DATABASE IF NOT EXISTS holo_tpcds  -- 在hologres中创建holo_tpcds库。
WITH ('sink.parallelism' = '4') -- 可选，指定目标库的参数，每个holo sink默认使用4并发。
AS DATABASE mysql.tpcds INCLUDING ALL TABLES  -- 同步mysql中tpcds库下所有表。
/*+ OPTIONS('server-id'='8001-8004') */ ; -- 可选，指定mysql-cdc源表的额外参数。

分库合并同步

同步场景：MySQL实例中有order_db01~order_db99多个分库，每个分库下都有order、order_detail等多张表。使用CDAS将MySQL的多个分库下的所有表全部同步到Hologres中，包括未来的数据变更和表结构变更。

同步方案：利用正则表达式的库名（`order_db[0-9]+`）来匹配所要同步的多个分库（order_db01~order_db99）。其中库名和表名会作为额外的两个字段写入到每张结果表中。为保证主键唯一性，库名、表名和原主键一起作为对应Hologres表的新联合主键。

代码示例及合并效果：

使用CDAS可以将上游多个分库下相同表名的数据合并同步到Hologres目标库对应表名的同一张表中，无需提前在Hologres中创建表。

USE CATALOG holo;

CREATE DATABASE IF NOT EXISTS holo_order--在Hologres中创建holo_order库，包括mysql中order分库的所有表。
WITH('sink.parallelism'='4')        --可选，指定目标库的参数，每个HologresSink默认并发为4。
AS DATABASE mysql.`order_db[0-9]+` INCLUDING ALL TABLES --同步mysql中order_db分库下所有表。
/*+OPTIONS('server-id'='8001-8004')*/;  --可选，指定mysql-cdc源表的额外参数。

源库新增表同步

同步场景：CDAS作业启动后，需要对源库新增表进行数据同步。

同步方案：在SQL作业中开启新增表读取功能，从作业快照重启，从而捕获到新的表，对新增表进行数据同步。

使用限制：VVR 8.0.6及以上版本支持源库新增表同步功能，启用该功能需确保源表启动模式为initial。

操作步骤：

1. 当出现新增的表需要同步时，在作业运维页面停止作业并勾选停止前创建一次快照。

2. 在SQL作业开发中开启新增表读取功能，然后重新部署作业。

   1. 在SQL作业中增加以下语句，开启CDAS新增表读取功能。

      SET 'table.cdas.scan.newly-added-table.enabled' = 'true';

   2. 单击部署。

3. 从快照恢复作业。

   1. 在作业运维页面单击目标作业名称，状态集管理页签，单击历史。

   2. 在作业快照列表中，找到停止作业时创建的快照。

   3. 单击目标快照操作列，选择更多 > 从该快照恢复作业完成作业启动。详情请参见作业启动。

多CDAS&CTAS语句

同步场景：通过一个作业将MySQL实例中tpcds、tpch、user_db01~user_db99（分库分表）多个库同步到Hologres。

同步方案：使用STATEMENT SET语法组合多条CDAS和CTAS语句作为一个作业提交。该方案可以复用一个Source节点读取所需表的数据，在MySQL CDC数据源场景可以减少server-id的使用及数据库的连接数与读取压力。

代码示例：

USE CATALOG holo;

BEGIN STATEMENT SET;

-- 同步user分库分表。
CREATE TABLE IF NOT EXISTS user
AS TABLE mysql.`user_db[0-9]+`.`user[0-9]+`
/*+ OPTIONS('server-id'='8001-8004') */;

-- 同步TPCDS库。
CREATE DATABASE IF NOT EXISTS holo_tpcds
AS DATABASE mysql.tpcds INCLUDING ALL TABLES
/*+ OPTIONS('server-id'='8001-8004') */ ;
-- 同步TPCH库。
CREATE DATABASE IF NOT EXISTS holo_tpch
AS DATABASE mysql.tpch INCLUDING ALL TABLES
/*+ OPTIONS('server-id'='8001-8004') */ ;

END;

多CDAS语句整库同步到Kafka

同步场景：将MySQL实例中tpcds、tpch多个库和表都同步到Kafka。

同步方案：在使用多个CDAS语句整库同步到Kafka时，由于不同的数据库中可能存在相同的表，为了防止topic冲突，需要使用cdas.topic.pattern配置。

cdas.topic.pattern定义了创建topic的名称的格式，其中可通过{table-name}占位符来替换为表名。例如：当设置'cdas.topic.pattern'='dbname-{table-name}'，对于上游表名为table1的表，在Kafka中对应的topic名称为dbname-table1。

代码示例：

USE CATALOG kafkaCatalog;

BEGIN STATEMENT SET;

-- 同步TPCDS库。
CREATE DATABASE IF NOT EXISTS kafka
WITH ('cdas.topic.pattern' = 'tpcds-{table-name}')
AS DATABASE mysql.tpcds INCLUDING ALL TABLES
/*+ OPTIONS('server-id'='8001-8004') */ ;

-- 同步TPCH库。
CREATE DATABASE IF NOT EXISTS kafka
WITH ('cdas.topic.pattern' = 'tpch-{table-name}')
AS DATABASE mysql.tpch INCLUDING ALL TABLES
/*+ OPTIONS('server-id'='8001-8004') */ ;

END;

实时计算Flink版提供MySQL整库同步到Kafka的能力，通过引入Kafka作为中间层，并使用CDAS整库同步或CTAS整表同步到Kafka来解决，具体操作请参见MySQL整库同步Kafka。

常见问题

作业运行异常

• 作业启动不起来，应该如何排查？

• 作业运行后出现重启，应该如何排查呢？

• 请求超时问题：报错：akka.pattern.AskTimeoutException

作业性能问题

• 如何排查作业反压问题？

• 全表读取阶段效率低且存在反压，如何解决？

• 如何排查作业延迟过高的问题？

• 消费上游数据速度不稳定，该如何排查？

数据同步问题

• 为什么主键为bigint unsigned的MySQL表注册Flink Catalog，主键会变为decimal？但是使用CTAS同步到Hologres后，主键又变为text？

• 使用MiniBatch配置报错：报错：Currently does not support merge StreamExecMiniBatchAssigner type ExecNode in CTAS/CDAS syntax

• 为什么作业没有输出？

• 如何定位Flink无法读取源数据的问题？

• 如何定位Flink无法将数据写入到结果表的问题？

• 作业数据不准确，该如何排查？