本文為您介紹如何使用Flink CDC將MySQL數(shù)據(jù)同步至EMR Serverless StarRocks中。
前提條件
- 已在新版控制臺創(chuàng)建DataFlow集群,詳情請參見創(chuàng)建集群。
- 已創(chuàng)建EMR Serverless StarRocks實例,詳情請參見創(chuàng)建實例。
- 已創(chuàng)建RDS MySQL,詳情請參見創(chuàng)建RDS MySQL實例。
說明 本文以5.7版本的MySQL、EMR-3.39.1版本的DataFlow集群為例介紹。
使用限制
- DataFlow集群、EMR Serverless StarRocks實例和RDS MySQL實例需要在同一個VPC下。
- DataFlow集群和EMR Serverless StarRocks實例均須開啟公網(wǎng)訪問。
- RDS MySQL須為5.7及以上版本。
操作流程
步驟一:準備測試數(shù)據(jù)
- 創(chuàng)建測試的數(shù)據(jù)庫和賬號,詳情請參見創(chuàng)建數(shù)據(jù)庫和賬號。創(chuàng)建完數(shù)據(jù)庫和賬號后,需要授權(quán)測試賬號的讀寫權(quán)限。說明 本文創(chuàng)建的數(shù)據(jù)庫名稱為test_cdc。
- 使用創(chuàng)建的測試賬號連接MySQL實例,詳情請參見通過DMS登錄RDS MySQL。
- 執(zhí)行以下命令,創(chuàng)建數(shù)據(jù)表。
CREATE TABLE test_cdc.`t_user` ( `id` bigint(20) NOT NULL AUTO_INCREMENT, `name` varchar(255) DEFAULT NULL, `age` tinyint(4) DEFAULT NULL, `create_time` datetime DEFAULT NULL, `update_time` datetime DEFAULT NULL, PRIMARY KEY (`id`) ) ENGINE=InnoDB DEFAULT CHARSET=utf8;
步驟二:配置同步工具和啟動Flink任務(wù)
- 使用SSH方式登錄DataFlow集群,詳情請參見登錄集群。
- 下載Flink CDC connector和Flink StarRocks Connector。說明 下載過程需要一定時間,請耐心等待。
- 執(zhí)行以下命令,將下載的Flink CDC Connector和Flink StarRocks Connector文件復(fù)制到DataFlow集群的/opt/apps/FLINK/flink-current/lib目錄下。
cp flink-* /opt/apps/FLINK/flink-current/lib/
- 執(zhí)行以下命令,啟動集群。重要 本文示例僅供測試,如果是生產(chǎn)級別的Flink作業(yè)請使用YARN或Kubernetes方式提交,詳情請參見Apache Hadoop YARN和Native Kubernetes。
/opt/apps/FLINK/flink-current/bin/start-cluster.sh
- 下載并修改配置文件。
- 執(zhí)行以下命令,將所有建表語句都生成在result目錄下。
./starrocks-migrate-tool
您可以通過
ls result
命令,查看result下的目錄。返回信息如下所示。flink-create.1.sql flink-create.all.sql starrocks-create.1.sql starrocks-create.all.sql starrocks-external-create.1.sql starrocks-external-create.all.sql
說明- 本文示例的配置文件僅定義了table-rule.1等規(guī)則,.1.sql格式文件對應(yīng)的就是table-rule.1的建表語句。如果您的配置文件有table-rule.2等規(guī)則,則.all.sql的文件為所有規(guī)則的集合。
external-create
后綴的文件,為對應(yīng)數(shù)據(jù)源的外表。如果對于部分場景小的維表您不想同步,則可以直接通過外表查詢,使用該文件可以生成對應(yīng)的外表。本文示例未使用。
- 執(zhí)行以下命令,創(chuàng)建StarRocks表。
mysql -h<EMR Serverless StarRocks實例FE節(jié)點的內(nèi)網(wǎng)地址> -P9030 -uroot -p < result/starrocks-create.1.sql
說明 如果修改config_prod.conf文件時,沒有設(shè)置StarRocks連接密碼,則直接按回車鍵。 - 執(zhí)行以下命令,啟動Flink任務(wù)。
/opt/apps/FLINK/flink-current/bin/sql-client.sh -f result/flink-create.1.sql
步驟三:驗證數(shù)據(jù)同步結(jié)果
查詢數(shù)據(jù)
- 登錄并連接EMR Serverless StarRocks實例,詳情請參見連接StarRocks實例(客戶端方式)。
- 執(zhí)行以下命令,查看數(shù)據(jù)庫信息。
show databases;
返回信息如下所示。+--------------------+ | Database | +--------------------+ | _statistics_ | | information_schema | | test_cdc | +--------------------+ 3 rows in set (0.00 sec)
- 在StarRocks連接窗口執(zhí)行以下命令,查看表數(shù)據(jù)。
use test_cdc; select * from t_user;
由于本文示例中t_user表中還沒有數(shù)據(jù),因此預(yù)期返回數(shù)據(jù)為空。
查詢插入后的數(shù)據(jù)
- 在RDS數(shù)據(jù)庫窗口執(zhí)行以下命令,插入數(shù)據(jù)。
INSERT INTO test_cdc.t_user(`name`,age,create_time,update_time) VALUES("aliyun.com.0",30,NOW(),NOW()); INSERT INTO test_cdc.t_user(`name`,age,create_time,update_time) VALUES("aliyun.com.1",31,NOW(),NOW()); INSERT INTO test_cdc.t_user(`name`,age,create_time,update_time) VALUES("aliyun.com.2",32,NOW(),NOW());
- 在StarRocks連接窗口執(zhí)行以下命令,查看表數(shù)據(jù)。
select * from t_user;
返回信息如下,表示數(shù)據(jù)已成功插入。+------+--------------+------+---------------------+---------------------+ | id | name | age | create_time | update_time | +------+--------------+------+---------------------+---------------------+ | 4 | aliyun.com.0 | 30 | 2022-03-10 13:22:41 | 2022-03-10 13:22:41 | | 5 | aliyun.com.1 | 31 | 2022-03-10 13:22:41 | 2022-03-10 13:22:41 | | 6 | aliyun.com.2 | 32 | 2022-03-10 13:22:42 | 2022-03-10 13:22:42 | +------+--------------+------+---------------------+---------------------+ 3 rows in set (0.00 sec)
同步數(shù)據(jù)更新
- 在RDS數(shù)據(jù)庫窗口執(zhí)行以下命令,更新指定數(shù)據(jù)。
UPDATE test_cdc.t_user SET age=35 where name="aliyun.com.0";
- 在StarRocks連接窗口執(zhí)行以下命令,查看表數(shù)據(jù)。
select * from t_user where name = "aliyun.com.0";
返回信息如下,表示數(shù)據(jù)已同步更新。+------+--------------+------+---------------------+---------------------+ | id | name | age | create_time | update_time | +------+--------------+------+---------------------+---------------------+ | 4 | aliyun.com.0 | 35 | 2022-03-10 13:22:41 | 2022-03-10 13:22:41 | +------+--------------+------+---------------------+---------------------+ 1 row in set (0.01 sec)
同步數(shù)據(jù)刪除
- 在RDS數(shù)據(jù)庫窗口執(zhí)行以下命令,刪除指定數(shù)據(jù)。
DELETE FROM test_cdc.t_user where 1=1;
- 在StarRocks連接窗口執(zhí)行以下命令,查看表數(shù)據(jù)。
select * from t_user;
返回信息如下,表示數(shù)據(jù)已同步刪除。Empty set (0.01 sec)
更多信息
- 本文檔僅供測試使用,生產(chǎn)級別的Flink作業(yè)請使用阿里云VVP產(chǎn)品進行配置,或者使用YARN或者Kubernetes提交作業(yè)。
- 如果RDS的表有修改(
ALTER TABLE
),則MySQL中alter table
之后的Schema變更需要在StarRocks中手動同步。如果RDS的表有新建,則MySQL新建的表需要重新運行StarRocks Migrate Tool以進行數(shù)據(jù)同步。 - StarRocks Migrate Tool的配置文件示例。
[db] host = rm-bp1nu0c46fn9k****.mysql.rds.aliyuncs.com port = 3306 user = *** password = *** # currently available types: `mysql`, `pgsql`, `oracle`, `hive`, `clickhouse` type = mysql # # only takes effect on `type == hive`. # # Available values: kerberos, none, nosasl, kerberos_http, none_http, zk, ldap # authentication = kerberos [other] # number of backends in StarRocks be_num = 1 # `decimal_v3` is supported since StarRocks-1.8.1 use_decimal_v3 = false # directory to save the converted DDL SQL output_dir = ./result # !!!`database` `table` `schema` are case sensitive in `oracle`!!! [table-rule.1] # pattern to match databases for setting properties # !!! database should be a `whole instance(or pdb) name` but not a regex when it comes with an `oracle db` !!! database = ^test.*$ # pattern to match tables for setting properties table = ^.*$ # `schema` only takes effect on `postgresql` and `oracle` and `sqlserver` schema = ^.*$ ############################################ ### starrocks table configurations ############################################ # # set a column as the partition_key # partition_key = p_key # # override the auto-generated partitions # partitions = START ("2021-01-02") END ("2021-01-04") EVERY (INTERVAL 1 day) # # only take effect on tables without primary keys or unique indexes # duplicate_keys=k1,k2 # # override the auto-generated distributed keys # distributed_by=k1,k2 # # override the auto-generated distributed buckets # bucket_num=32 # # properties.xxxxx: properties used to create tables # properties.in_memory = false ############################################ ### flink sink configurations ### DO NOT set `connector`, `table-name`, `database-name`, they are auto-generated ############################################ flink.starrocks.jdbc-url=jdbc:mysql://fe-c-9b354c83e891****-internal.starrocks.aliyuncs.com:9030 flink.starrocks.load-url=fe-c-9b354c83e891****-internal.starrocks.aliyuncs.com:8030 flink.starrocks.username=admin flink.starrocks.password=1qaz!QAZ flink.starrocks.sink.max-retries=10 flink.starrocks.sink.buffer-flush.interval-ms=15000 flink.starrocks.sink.properties.format=json flink.starrocks.sink.properties.strip_outer_array=true # # used to set the server-id for mysql-cdc jobs instead of using a random server-id # flink.cdc.server-id = 5000 ############################################ ### flink-cdc plugin configuration for `postgresql` ############################################ # # for `9.*` decoderbufs, wal2json, wal2json_rds, wal2json_streaming, wal2json_rds_streaming # # refer to https://ververica.github.io/flink-cdc-connectors/master/content/connectors/postgres-cdc.html # # and https://debezium.io/documentation/reference/postgres-plugins.html # flink.cdc.decoding.plugin.name = decoderbufs