基于Flink平台同步MySQL数据至StarRocks-开源大数据平台 E-MapReduce(EMR)-阿里云帮助中心

Flink CDC数据摄入可以实现单表的结构和数据同步、整库同步或者同一库中的多表结构和数据同步。本文为您介绍如何使用实时计算Flink平台和E-MapReduce StarRocks通过Flink CDC数据摄入实现实时数仓中TP（Transaction Processing）和AP（Analytical Processing）数据同步的场景。

背景信息

通过Flink CDC数据摄入可以在StarRocks中自动创建和MySQL中表结构一致的表，并进行数据同步，还能实时同步上游表结构（Schema）的变更到下游表，提高您在目标存储中创建表和维护源表结构变更的效率。

当运行Flink CDC数据摄入任务时，Flink会按照以下流程执行：

编辑数据摄入作业并部署启动。
作业运行后，首先检查目标存储中是否存在该目标表。
- 如果不存在，则自动在目标存储中创建相应的目标表，该目标表具有和数据源相同的Schema。
- 如果存在，则跳过建表。
同步数据源的数据以及Schema的变更到目标表中。

Flink CDC数据摄入作业在实时同步数据的同时，还能同步源表Schema的变更到目标表中，可以通过表结构变更同步配置来选择不同的同步策略，支持的详细变更类型可以在StarRocks Sink文档中查看细节。

说明

如果遇到不支持的Schema变更，可以在sink配置中配置跳过的 schema 变更类型来进行兼容处理，详情可见exclude.schema.changes。
如果需要修改表的主键定义、添加分区定义、数据过滤或者添加元数据字段，可以通过transform模块进行添加，详情可见Transform模块。
如果需要重命名表或者多表合并，可以通过route模块进行修改，详情可见Route模块。

前提条件

已开通阿里云实时计算Flink全托管并创建了Flink集群，详情请参见开通Flink全托管和Flink CDC数据摄入作业。
已创建StarRocks集群，详情请参见创建StarRocks集群。
已创建RDS MySQL，详情请参见创建RDS MySQL实例。

说明

本文以5.7版本的MySQL、EMR-3.39.1版本的StarRocks集群和11.9版本的Flink为例介绍。

使用限制

创建的Flink集群、StarRocks集群以及RDS MySQL实例需要在同一个VPC下。
RDS MySQL须为5.7及以上版本。
StarRocks须开启公网访问。
Flink集群中的Flink须为11.9及以上版本。

步骤一：准备测试数据

创建测试的数据库和账号，详情请参见创建数据库和账号。
创建完数据库和账号后，需要授权测试账号的读写权限。
说明
本文创建的数据库名称为test_cdc，账号为test。
使用创建的测试账号连接MySQL实例，详情请参见通过DMS登录RDS MySQL。

在MySQL中执行以下命令，创建数据表。

use test_cdc;

CREATE TABLE IF NOT EXISTS `runoob_tbl`(
   `runoob_id` INT UNSIGNED AUTO_INCREMENT,
   `runoob_title` VARCHAR(100) NOT NULL,
   `runoob_author` VARCHAR(40) NOT NULL,
   `submission_date` DATE,
   `add_col` int DEFAULT NULL,
   PRIMARY KEY ( `runoob_id` )
)ENGINE=InnoDB DEFAULT CHARSET=utf8;


INSERT INTO test_cdc.`runoob_tbl` (`runoob_id`,`runoob_title`,`runoob_author`,`submission_date`,`add_col`) values (18,'first','tom','2022-06-22 17:13:44',3)

使用SSH方式登录StarRocks集群，详情请参见登录集群。
执行以下，连接StarRocks集群。
```
mysql -h127.0.0.1 -P 9030 -uroot
```

执行以下命令，创建用户和授权。

CREATE DATABASE test_cdc;
CREATE USER 'test' IDENTIFIED by '123456';
GRANT CREATE TABLE ON DATABASE test_cdc TO test;

步骤二：在实时计算Flink版编写Flink CDC数据摄入作业

在阿里云实时计算Flink控制台的数据开发->数据摄入页面中，创建从MySQL到StarRocks的数据摄入作业。详情请参见Flink CDC数据摄入作业。

说明

参数仅供参考格式，具体内容请根据实际情况配置。
YAML连接器当前仅支持至少一次语义。即使显式设置sink.semantic: exactly-once，也会被覆盖为at-least-once，不会报错。

整库同步代码示例，其他场景使用示例可见Starrocks数据摄入连接器代码示例

source:
  type: mysql
  hostname: rm-2zepd6e20u3od****.mysql.rds.aliyuncs.com
  port: 3306
  username: test
  password: 123456
  server-id: 8601-8604
  # 正则匹配表名，"." 需要转义
  tables: test_cdc.\.*
  #（可选）同步增量阶段新创建的表的数据
  scan.binlog.newly-added-table.enabled: true
  #（可选）同步表注释和字段注释
  include-comments.enabled: true
  #（可选）优先分发无界的分片以避免可能出现的TaskManager OutOfMemory问题
  scan.incremental.snapshot.unbounded-chunk-first.enabled: true
  #（可选）开启解析过滤，加速读取
  scan.only.deserialize.captured.tables.changelog.enabled: true

sink:
  type: starrocks
  jdbc-url: 172.16.**.**:9030   
  load-url: 172.16.**.**:8030              
  username: test
  password: 123456
  # （可选）数据量不大的作业建议调低 flush 间隔，避免数据长时间不落盘（默认 300000，即 5 分钟）
  sink.buffer-flush.interval-ms: 5000
  # （可选）自动建表的分桶数；StarRocks 2.5.7 以下必须显式配置，更高版本可自动推断
  table.create.num-buckets: 8
  # （可选）上游为 utf8mb4 字符集时建议设为 4，避免文本截断（默认 3）
  unicode-char.max-bytes: 4
  # （可选）自动建表的副本数，按集群情况配置
  table.create.properties.replication_num: 3
  # （可选）StarRocks 3.2 及以上建议开启，加速表结构变更
  table.create.properties.fast_schema_evolution: true
  # 注意：通过 transform 变更主键时，必须同时设置 sink.ignore.update-before: false，
  # 否则旧主键对应的行会残留在下游

参数配置

Source配置

参数	说明
type	类型，固定值为mysql。
hostname	RDS的内网地址。您可以在RDS的数据库连接页面，单击内网地址进行复制。例如，rm-2zepd6e20u3od****.mysql.rds.aliyuncs.com。
port	MySQL数据库服务的端口号，默认值为3306。
username	MySQL数据库服务的用户名。填写步骤一：准备测试数据中账号的用户名。本示例为test。
password	MySQL数据库服务的密码。填写步骤一：准备测试数据中账号的密码。
server-id	数据库客户端的用于同步的数字ID或范围，该ID必须是MySQL集群中全局唯一的。
tables	需要同步的MySQL数据表，表名支持正则表达式以读取多个表的数据。
scan.binlog.newly-added-table.enabled	在增量阶段，是否发送匹配到的新增表的数据。
include-comments.enabled	是否同步表注释和字段注释。
scan.incremental.snapshot.unbounded-chunk-first.enabled	快照读取阶段是否先分发无界的分片，避免可能出现的TaskManager OutOfMemory问题。
scan.only.deserialize.captured.tables.changelog.enabled	在增量阶段，是否仅对指定表的变更事件进行反序列化。

Sink配置

参数	说明
type	类型，固定值为starrocks。
jdbc-url	JDBC连接的URL。
load-url	连接到FE节点的HTTP服务URL。
username	StarRocks的用户名。填写步骤一：准备测试数据中账号的用户名。本示例为test。
password	StarRocks数据库服务的密码。填写步骤一：准备测试数据中账号的密码。
sink.buffer-flush.interval-ms	每张表连续两次Flush之间的间隔时间。
table.create.num-buckets	自动建表时的Bucket数量。
unicode-char.max-bytes	为每个Unicode字符分配多少个字节。
table.create.properties.*	在自动建表时需要传递的额外参数。如示例中，table.create.properties.replication_num添加自动建表的副本数，table.create.properties.fast_schema_evolution在StarRocks 3.2 及以上建议开启，加速表结构变更。

步骤三：部署作业

在作业开发页面的高级配置中，选择vvr-11.9及以上的版本。
单击部署。
在作业运维页面，单击目标作业操作列的启动。

步骤四：场景演示

查询数据

使用SSH方式登录StarRocks集群，详情请参见登录集群。
执行以下，连接StarRocks集群。
```
mysql -h127.0.0.1 -P 9030 -uroot
```

在StarRocks连接窗口执行以下命令，查看表数据。

use test_cdc;
select * from runoob_tbl1;

返回信息如下，表示MySQL上的数据已同步至StarRocks。

+-----------+--------------+---------------+-----------------+---------+
| runoob_id | runoob_title | runoob_author | submission_date | add_col |
+-----------+--------------+---------------+-----------------+---------+
|        18 | first        | tom           | 2022-06-22      |       3 |
+-----------+--------------+---------------+-----------------+---------+

查询插入后的数据

在RDS数据库窗口执行以下命令，插入数据。

INSERT INTO runoob_tbl(`runoob_id`,`runoob_title`,`runoob_author`,`submission_date`,`add_col`)  values(1,'second','tom2','2022-06-23',1)

在StarRocks连接窗口执行以下命令，查看表数据。

select * from runoob_tbl1;

返回信息如下，表示数据已成功插入。

+-----------+--------------+---------------+-----------------+---------+
| runoob_id | runoob_title | runoob_author | submission_date | add_col |
+-----------+--------------+---------------+-----------------+---------+
|         1 | second       | tom2          | 2022-06-23      |       1 |
|        18 | first        | tom           | 2022-06-22      |       3 |
+-----------+--------------+---------------+-----------------+---------+

同步数据更新

在RDS数据库窗口执行以下命令，更新指定数据。
```
update runoob_tbl set runoob_title= 'new' where runoob_id = 18
```

在StarRocks连接窗口执行以下命令，查看表数据。

select * from runoob_tbl1;

返回信息如下，表示数据已同步更新。

+-----------+--------------+---------------+-----------------+---------+
| runoob_id | runoob_title | runoob_author | submission_date | add_col |
+-----------+--------------+---------------+-----------------+---------+
|         1 | second       | tom2          | 2022-06-23      |       1 |
|        18 | new          | tom           | 2022-06-22      |       3 |
+-----------+--------------+---------------+-----------------+---------+

同步数据删除

在RDS数据库窗口执行以下命令，删除指定数据。
```
DELETE FROM runoob_tbl WHERE runoob_id = 1
```

在StarRocks连接窗口执行以下命令，查看表数据。

select * from runoob_tbl1;

返回信息如下，表示数据已同步删除。

+-----------+--------------+---------------+-----------------+---------+ 
| runoob_id | runoob_title | runoob_author | submission_date | add_col | 
+-----------+--------------+---------------+-----------------+---------+
|        18 | new          | tom           | 2022-06-22      |       3 | 
+-----------+--------------+---------------+-----------------+---------+

增加可空列

在RDS数据库窗口执行以下命令，增加可空列。
```
alter table `runoob_tbl` add COLUMN `add_col2` INT;
```

执行以下命令，插入数据。

INSERT INTO runoob_tbl(`runoob_id`,`runoob_title`,`runoob_author`,`submission_date`,`add_col`,`add_col2`)  values(1,'second','tom2','2022-06-23',1,2)

在StarRocks连接窗口执行以下命令，查看表数据。

select * from runoob_tbl1;

返回信息如下，表示Schema已经成功变更。

+-----------+--------------+---------------+-----------------+---------+----------+
| runoob_id | runoob_title | runoob_author | submission_date | add_col | add_col2 | 
+-----------+--------------+---------------+-----------------+---------+----------+ 
|        18 | new          | tom           | 2022-06-22      |       3 |     NULL |
+-----------+--------------+---------------+-----------------+---------+----------+ 
|         1 | second       | tom2          | 2022-06-23      |       1 |      2   |
|        18 | first        | tom           | 2022-06-22      |       3 |     NULL |
+-----------+--------------+---------------+-----------------+---------+----------+