腾讯云数据仓库 TCHouse-C Oceanus SQL 同步

一、前提条件

1. 已开通 Oceanus 服务。

2. Oceanus 集群和云数据 TCHouse-C 集群须在同一个 VPC 下。

3. 流计算作业 SQL 作业需运行于流计算独享集群，若还没有集群，请参见 创建独享集群。

二、操作步骤

1. 登录云数据仓库 TCHouse-C ，创建目标表：

·若你的任务有 update 和 delete 操作，可以通过 CollapsingMergeTree 来实现。

CREATE TABLE test.test ON CLUSTER default_cluster( `id` Int32, `Sign` Int8)ENGINE = CollapsingMergeTree(Sign)ORDER BY id

·若你到任务中不需要 update，可以通过 MergeTree 来实现。

CREATE TABLE test.test ON CLUSTER default_cluster( `id` Int32)ENGINE = MergeTree()ORDER BY id

2. 在 Oceanus 控制台发布 SQL 作业，详细操作请参见 SQL作业开发。

2.1 登录 流计算 Oceanus 控制台，单击左侧导航工作空间，点击工作空间，进入作业管理页面。

2.2 单击新建作业，作业类型选中 SQL 作业，输入作业名称，并选择一个运行中的集群。

2.3 获取流计算服务委托授权。

2.4 编写 SQL 语句，并完成字段映射。
授权完成后，可在开发调试的代码编辑框中输入 SQL 语句，可无需另外准备数据快速创建作业。示例语句具体执行的内容如下：

Source 端Sink 端

MySQL-CDC Source（学生信息作为 cdc 源表）：CREATE TABLE `student` ( `id` INT NOT NULL, PRIMARY KEY (`ID`) NOT ENFORCED) WITH ( 'connector' = 'mysql-cdc', — 必须为 'mysql-cdc' 'hostname' = 'YoursIp', — 数据库的 IP 'port' = '3306', — 数据库的访问端口 'username' = '用户名', — 数据库访问的用户名（需要提供 SHOW DATABASES, REPLICATION SLAVE, REPLICATION CLIENT, SELECT, RELOAD 权限） 'password' = 'YoursPassword,– 数据库访问的密码'database-name' = 'mysqltestdb',– 需要同步的数据库'table-name' = 'student' — 需要同步的数据表名);

·当任务中无 update 时：

CREATE TABLE clickhouse_sink (`id` INT) WITH (– 指定数据库连接参数'connector' = 'clickhouse', — 指定使用clickhouse连接器'url' = 'clickhouse://xxx:8123', — 指定集群地址，可以通过ClickHouse集群界面查看– 如果ClickHouse集群未配置账号密码可以不指定–'username' = 'root', — ClickHouse集群用户名–'password' = 'root', — ClickHouse集群的密码'database-name' = 'test', — 数据写入目的数据库'table-name' = 'test', — 数据写入目的数据表'sink.batch-size' = '1000' — 触发批量写的条数);

·当任务中包含 update 和 delete 操作：

CREATE TABLE clickhouse_upsert_sink_table (`id` INT PRIMARY KEY (`id`) NOT ENFORCED — 如果要同步的数据库表定义了主键, 则这里也需要定义) WITH (– 指定数据库连接参数'connector' = 'clickhouse', — 指定使用clickhouse连接器'url' = 'clickhouse://xxx:8123', — 指定集群地址，可以通过ClickHouse集群界面查看– 如果ClickHouse集群未配置账号密码可以不指定–'username' = 'root', — ClickHouse集群用户名–'password' = 'root', — ClickHouse集群的密码'database-name' = 'test', — 数据写入目的数据库'table-name' = 'test', — 数据写入目的数据表'table.collapsing.field' = 'Sign', — CollapsingMergeTree 类型列字段的名称'sink.batch-size' = '1000' — 触发批量写的条数);

OCeanus ClickHouse_Sink 参数：

注意：定义 WITH 参数时，通常只需要填写必填参数即可。当你启用非必填参数时，请你一定要明确参数含义以及可能对数据写入产生的影响。

2.5 进行逻辑运算。
此示例中，只进行了简单的 Join 没有进行复杂的运算。详细运算逻辑可参见 Oceanus 运算符和内置函数 或者 Flink 页面 Flink SQL 开发。

INSERT INTO clickhouse_sinkSELECT idFROM student

2.6 发布运行 SQL 作业。

3. 登录 云数据仓库 TCHouse-C ，并查看数据。

select * from test.test ;