0
0

第3.4章：StarRocks数据导入--Flink Connector与CDC秒级数据同步

Flink作为当前流行的流式计算框架，在对接StarRocks时，若直接使用JDBC的方式“流式”写入数据，对StarRocks是不友好的，StarRocks作为一款MVCC的数据库，其导入的核心思想还是“攒微批+降频率”。为此，StarRocks单独开发了flink-connector-starrocks，其内部实现仍是通过对数据缓存攒批后执行Stream Load导入。

StarRocks Flink Connector目前也已开源，其github地址为：

GitHub - StarRocks/flink-connector-starrockshttps://github.com/StarRocks/flink-connector-starrocks

在整理这篇文章时，flink-connector-starrocks除了支持向StarRocks中写入数据（Sink），也支持了从StarRocks中读取数据（Source）的功能，后面咱们一并介绍。

StarRocks官网文档中对Flink Connector的介绍非常详细，给出了在代码中使用的示例以及详细的参数介绍：

设计背景 @ Flink-connector-starrocks @ StarRocks Docshttps://docs.starrocks.com/zh-cn/main/loading/Flink-connector-starrocks

这里我们就使用Flink-SQL简单演示，把需要注意的几个点引一下。首先还是搭建演示环境：

节点IP

部署服务

端口

版本

说明

192.168.110.101

[node01]

FE

9030

2.0.1

query_port

BE

用户名密码均为root

Broker

Broker名称：hdfs_broker

mysql-client

5.7.36

Flink

8081

flink-1.13.5-bin-scala_2.11

推荐使用1.13，最低支持1.11

flink-connector-starrocks-1.1.14-snapshot_flink-1.13_2.11.jar

官方公测的同时支持Source和Sink的Connector

flink-sql-connector-mysql-cdc-2.0.2.jar

MySQL CDC依赖

flink-connector-kafka_2.11-1.13.5.jar

读取Kafka依赖

flink-connector-jdbc_2.11-1.13.5.jar

JDBC MySQL依赖

192.168.110.101

[node01]

Zookeeper

2181

Zookeeper version: 3.4.13

Kafka

9092

kafka_2.13-2.8.1

无认证

192.168.110.102

[node02]

MySQL Community Server

3306

5.7.36

用户名密码均为root

上面的jar包都需要拷贝到flink的lib目录下。其中，CDC的版本要特别注意，其与Flink之间的版本对应关系如下表：

Flink CDC连接器版本

Flink版本

1.0.0

1.11.*

1.1.0

1.11.*

1.2.0

1.12.*

1.3.0

1.12.*

1.4.0

1.13.*

2.0.*

1.13.*

参考地址：

About Flink CDC — Flink CDC 2.0.0 documentationhttps://ververica.github.io/flink-cdc-connectors/release-2.0/content/about.html#supported-flink-versions

接下来，我们就以五个业务中常见的场景简单说明，其中StarRocks中的建表都以主键模型为例，因为主键模型可以更好的支持实时/频繁更新的场景。

场景一：Flink读取Kafka数据写入StarRocks

Routine Load是StarRocks自带的可以消费Kafka数据的导入方式，其特点是简单易用，不依赖外部组件，但若需要对Kafka中的数据进行较复杂的ETL，Routine Load可能就不能胜任了，这时我们就可以考虑使用Flink去消费Kafka中的数据，进行清洗转换后，再sink至StarRocks。

咱们举一个业务中非常常见的实时报表的例子，我们使用Flink对Kafka中追加写入的数据进行实时处理，然后将数据源源不断的同步入库StarRocks。

这个场景用到的jar包有两个：flink-connector-kafka_2.11-1.13.5.jar和flink-connector-starrocks-1.1.14-snapshot_flink-1.13_2.11.jar，在场景模拟前要确认jar包已放入flink/lib目录下。

1.1 数据准备

在Kafka中创建主题behavior和province：

kafka-topics.sh --zookeeper 192.168.110.101:2181 --create --replication-factor 1 --partitions 1 --topic behavior

kafka-topics.sh --zookeeper 192.168.110.101:2181 --create --replication-factor 1 --partitions 1 --topic province

向主题behavior生产数据：

kafka-console-producer.sh --broker-list 192.168.110.101:9092 --topic behavior

生产数据：

10001,zs,18,11,shopping

10002,ls,19, 11,add

10003,ww,19,61,star

向主题province生产数据：

kafka-console-producer.sh --broker-list 192.168.110.101:9092 --topic province

生产数据：

11,北京

61,陕西

1.2 StarRocks准备

StarRocks中创建主键模型表s_province：

mysql> create database starrocks;

mysql> use starrocks;

mysql> CREATE TABLE IF NOT EXISTS starrocks.s_province (

uid int(10) NOT NULL COMMENT "",

p_id int(2) NOT NULL COMMENT "",

p_name varchar(30) NULL COMMENT ""

)

PRIMARY KEY(uid)

DISTRIBUTED BY HASH(uid) BUCKETS 1

PROPERTIES (

"replication_num" = "1"

);

1.3 Flink准备

启动Flink：

[root@node01 bin]# ./start-cluster.sh

启动sql-client：

[root@node01 bin]# ./sql-client.sh embedded

执行Flink SQL，创建上下游的映射表：

Source部分，创建Flink向Kafka的映射表kafka_source_behavior：

Flink SQL> CREATE TABLE kafka_source_behavior (

uuid int,

name string,

age int,

province_id int,

behavior string

) WITH (

'connector' = 'kafka',

'topic' = 'behavior',

'properties.bootstrap.servers' = '192.168.110.101:9092',

'properties.group.id' = 'source_behavior',

'scan.startup.mode' = 'earliest-offset',

'format' = 'csv'

);

创建映射表kafka_source_province：

Flink SQL> CREATE TABLE kafka_source_province (

pid int,

p_name string

) WITH (

'connector' = 'kafka',

'topic' = 'province',

'properties.bootstrap.servers' = '192.168.110.101:9092',

'properties.group.id' = 'source_province',

'scan.startup.mode' = 'earliest-offset',

'format' = 'csv'

);

Sink部分，创建Flink向StarRocks的映射表sink_province：

Flink SQL> CREATE TABLE sink_province (

uid INT,

p_id INT,

p_name STRING,

PRIMARY KEY (uid) NOT ENFORCED

)WITH (

'connector' = 'starrocks',

'jdbc-url'='jdbc:mysql://192.168.110.101:9030',

'load-url'='192.168.110.101:8030',

'database-name' = 'starrocks',

'table-name' = 's_province',

'username' = 'root',

'password' = 'root',

'sink.buffer-flush.interval-ms' = '5000',

'sink.properties.column_separator' = '\x01',

'sink.properties.row_delimiter' = '\x02'

);

1.4 执行同步任务

执行Flink SQL，开始同步任务：

Flink SQL> insert into sink_province select b.uuid as uid, b.province_id as p_id, p.p_name from kafka_source_behavior b join kafka_source_province p on b.province_id = p.pid;

1.5 StarRocks查看数据

[root@node01 ~]# mysql -h192.168.110.101 -P9030 -uroot –proot

mysql> use starrocks;

mysql> select * from s_province;

+----------+----------+-------------+

| uid | p_id | p_name |

+----------+----------+-------------+

| 10003 | 61 | 陕西 |

| 10002 | 11 | 北京 |

| 10001 | 11 | 北京 |

+----------+----------+-------------+

场景二：Flink JDBC读取MySQL数据写入StarRocks

使用Flink JDBC方式读取MySQL数据的实时场景不多，因为JDBC下Flink只能获取执行命令时MySQL表的数据，所以更适合离线场景。假设有复杂的MySQL数据，我们就可以在Flink中跑定时任务，来获取清洗后的数据，完成后写入StarRocks。

这个场景用到的jar包有：flink-connector-jdbc_2.11-1.13.5.jar和flink-connector-starrocks-1.1.14-snapshot_flink-1.13_2.11.jar。

2.1 MySQL准备

在node02中登陆MySQL：

[root@node02 ~]# mysql -uroot –proot

在MySQL中创建表s_user：

mysql> use ODS;

mysql> CREATE TABLE s_user (

id INT(11) NOT NULL,

name VARCHAR(32) DEFAULT NULL,

p_id INT(2) DEFAULT NULL,

PRIMARY KEY (id)

);

插入数据：

mysql> insert into s_user values(10086,'lm',61),(10010, 'ls',11), (10000,'ll',61);

2.2 StarRocks准备

转到node01，访问StarRocks:

[root@node01 ~]# mysql -h192.168.110.101 -P9030 -uroot -proot

在StarRocks创建表s_user：

mysql> use starrocks;

mysql> CREATE TABLE IF NOT EXISTS starrocks.s_user (

id int(10) NOT NULL COMMENT "",

name varchar(20) NOT NULL COMMENT "",

p_id INT(2) NULL COMMENT ""

)

PRIMARY KEY(id)

DISTRIBUTED BY HASH(id) BUCKETS 1

PROPERTIES (

"replication_num" = "1"

);

2.3 Flink创建映射表

启动Flink（若前面服务未停止，这里可以跳过）：

[root@node01 bin]# ./start-cluster.sh

启动sql-client：

[root@node01 bin]# ./sql-client.sh embedded

Source部分，创建映射至MySQL的映射表source_mysql_suser：

Flink SQL> CREATE TABLE source_mysql_suser (

id INT,

name STRING,

p_id INT,

PRIMARY KEY (id) NOT ENFORCED

)WITH (

'connector' = 'jdbc',

'url' = 'jdbc:mysql://192.168.110.102:3306/ODS',

'table-name' = 's_user',

'username' = 'root',

'password' = 'root'

);

Sink部分，创建至StarRocks的映射表sink_starrocks_suser：

Flink SQL> CREATE TABLE sink_starrocks_suser (

id INT,

name STRING,

p_id INT,

PRIMARY KEY (id) NOT ENFORCED

)WITH (

'connector' = 'starrocks',

'jdbc-url'='jdbc:mysql://192.168.110.101:9030',

'load-url'='192.168.110.101:8030',

'database-name' = 'starrocks',

'table-name' = 's_user',

'username' = 'root',

'password' = 'root',

'sink.buffer-flush.interval-ms' = '5000',

'sink.properties.column_separator' = '\x01',

'sink.properties.row_delimiter' = '\x02'

);

2.4 Flink清洗数据并写入StarRocks

这里只是简单做一个where筛选，实际业务可能是多表join的复杂场景：

Flink SQL> insert into sink_starrocks_suser select id,name,p_id from source_mysql_suser where p_id = 61;

数据写入StarRocks后，Flink任务完成并结束。此时若我们再对MySQL中s_user表的数据进行增删或修改操作，Flink亦不会感知。

2.5 StarRocks查看数据

[root@node01 ~]# mysql -h192.168.110.101 -P9030 -uroot –proot

mysql> use starrocks;

mysql> select * from s_user;

+----------+----------+--------+

| id | name | p_id |

+----------+----------+--------+

| 10000 | ll | 61 |

| 10086 | lm | 61 |

+----------+----------+--------+

场景三：Flink读取StarRocks数据写入MySQL

还使用场景二中的MySQL s_user表和StarRocks的s_user表，这次我们将业务流程反转一下，读取StarRocks中的数据写入其他业务库，例如MySQL。

这里用到的jar包还是：flink-connector-jdbc_2.11-1.13.5.jar和flink-connector-starrocks-1.1.14-snapshot_flink-1.13_2.11.jar（公测版，支持Source）。

Source部分目前可以参考论坛说明及git dev分支：

flink-connector-source 功能内测包 - 功能使用相关 - StarRocks数据库论坛https://forum.starrocks.com/t/topic/1179

https://github.com/StarRocks/flink-connector-starrocks/tree/devhttps://github.com/StarRocks/flink-connector-starrocks/tree/dev

3.1 Flink创建映射表

启动Flink（若前面服务未停止，这里可以跳过）：

[root@node01 bin]# ./start-cluster.sh

启动sql-client：

[root@node01 bin]# ./sql-client.sh embedded

Source部分，创建StarRocks映射表source_starrocks_suser：

Flink SQL> CREATE TABLE source_starrocks_suser (

id INT,

name STRING,

p_id INT

)WITH (

'connector' = 'starrocks',

'scan-url'='192.168.110.101:8030',

'jdbc-url'='jdbc:mysql://192.168.110.101:9030',

'database-name' = 'starrocks',

'table-name' = 's_user',

'username' = 'root',

'password' = 'root'

);

Sink部分，创建向MySQL的映射表sink_mysql_suser：

Flink SQL> CREATE TABLE sink_mysql_suser (

id INT,

name STRING,

p_id INT,

PRIMARY KEY (id) NOT ENFORCED

)WITH (

'connector' = 'jdbc',

'url' = 'jdbc:mysql://192.168.110.102:3306/ODS',

'table-name' = 's_user',

'username' = 'root',

'password' = 'root'

);

3.2 MySQL准备

在node02中登陆MySQL：

[root@node02 ~]# mysql -uroot –proot

清空MySQL s_user表数据，为一会儿导入新数据做准备：

mysql> use ODS;

mysql> truncate table s_user;

3.3 Flink执行导入任务

这里还是简单梳理操作，实际业务可能会对StarRocks中多个表的数据进行分组或者join等处理然后再导入。在node01的sql-client中执行导入任务：

Flink SQL> insert into sink_mysql_suser select id,name,p_id from source_starrocks_suser;

3.4 查看MySQL数据

mysql> select * from s_user;

+----------+----------+--------+

| id | name | p_id |

+----------+----------+--------+

| 10000 | ll | 61 |

| 10086 | lm | 61 |

+----------+----------+--------+

场景四：Flink CDC同步MySQL数据至StarRocks

在场景二使用Flink JDBC来读取MySQL数据时，我们已经解释了JDBC的方式是“一次性”的导入，若我们希望让Flink感知MySQL数据源的数据变化，并近实时的实现数据同步，就需要使用Flink CDC。

CDC是变更数据捕获（Change Data Capture）技术的缩写，它可以将源数据库（Source）的数据变动记录，同步到一个或多个数据目的地中（Sink）。直观的说就是当数据源的数据变化时，通过CDC可以让目标库中的数据同步发生变化（仅限于DML操作）。

这里我们还使用前面MySQL的s_user表以及StarRocks的s_user表来演示。

场景四这里需要用到的jar包有：flink-sql-connector-mysql-cdc-2.0.2.jar和flink-connector-starrocks-1.1.14-snapshot_flink-1.13_2.11.jar。

4.1 MySQL准备

首先，在node02中为MySQL开启binlog（格式为ROW模式）：

[root@node02 ~]# vi /etc/my.cnf

在配置文件末尾添加：

log-bin=mysql-bin # 开启binlog

binlog-format=ROW # 选择ROW模式

server_id=1 # 配置MySQL replaction

保存退出后，重启MySQL服务：

[root@node02 ~]# systemctl restart mysqld

4.2 StarRocks准备

在StarRocks中清空s_user表中的数据，不影响后面的同步任务：

[root@node01 ~]# mysql -h192.168.110.101 -P9030 -uroot –proot

mysql> use starrocks;

mysql> truncate table s_user;

4.3 Flink准备

启动Flink（若前面服务未停止，这里可以跳过）：

[root@node01 bin]# ./start-cluster.sh

启动sql-client：

[root@node01 bin]# ./sql-client.sh embedded

Source部分，创建MySQL映射表cdc_mysql_suser：

Flink SQL> CREATE TABLE cdc_mysql_suser (

id INT,

name STRING,

p_id INT

) WITH (

'connector' = 'mysql-cdc',

'hostname' = '192.168.110.102',

'port' = '3306',

'username' = 'root',

'password' = 'root',

'database-name' = 'ODS',

'scan.incremental.snapshot.enabled'='false',

'table-name' = 's_user'

);

Sink部分，创建向StarRocks的cdc_starrocks_suser：

Flink SQL> CREATE TABLE cdc_starrocks_suser (

id INT,

name STRING,

p_id INT,

PRIMARY KEY (id) NOT ENFORCED

)WITH (

'connector' = 'starrocks',

'jdbc-url'='jdbc:mysql://192.168.110.101:9030',

'load-url'='192.168.110.101:8030',

'database-name' = 'starrocks',

'table-name' = 's_user',

'username' = 'root',

'password' = 'root',

'sink.buffer-flush.interval-ms' = '5000',

'sink.properties.column_separator' = '\x01',

'sink.properties.row_delimiter' = '\x02'

);

4.4 执行同步任务

Flink SQL> insert into cdc_starrocks_suser select id,name,p_id from cdc_mysql_suser;

不同于场景二中的JDBC，在CDC场景下，Flink SQL执行后同步任务将会持续进行，当MySQL中数据出现变化，Flink会快速感知，并将变化同步至StarRocks中。

4.5 数据观察

在MySQL库中观察数据：

[root@node02 ~]# mysql -uroot –proot

mysql> use ODS;

mysql> select * from s_user;

+----------+----------+--------+

| id | name | p_id |

+----------+----------+--------+

| 10000 | ll | 61 |

| 10086 | lm | 61 |

+----------+----------+--------+

StarRocks库中观察数据

[root@node01 ~]# mysql -h192.168.110.101 -P9030 -uroot –proot

mysql> use starrocks;

mysql> select * from s_user;

+----------+----------+--------+

| id | name | p_id |

+----------+----------+--------+

| 10000 | ll | 61 |

| 10086 | lm | 61 |

+----------+----------+--------+

在MySQL中，对数据进行增删改操作：

mysql> INSERT INTO s_user VALUES(12345,'SR',61);

mysql> DELETE FROM s_user WHERE id = 10010;

mysql> UPDATE s_user SET name='No.1' WHERE id = 10086;

完成后，直接查看StarRocks中表的数据：

mysql> select * from s_user;

+----------+----------+--------+

| id | name | p_id |

+----------+----------+--------+

| 12345 | SR | 61 |

| 10086 | No.1 | 61 |

+----------+----------+--------+

可以确认对MySQL源表数据的增加、修改和删除操作引起的数据变化，都能同步至StarRocks目标表中。

场景五：通过CDC+SMT实现MySQL多表数据的秒级同步

场景四是针对单表的数据同步，那种方式只能同步数据，并不能同步表结构，我们需要先在目标库中创建对应的表，然后再执行同步任务同步数据。但若需要同步的数据表比较多或者需要整库同步，在StarRocks中逐个建表就会比较麻烦，在Flink中逐个写任务也会相对繁琐。

为了友好的解决多表同步时的问题，StarRocks发布了StarRocks-migrate-tools（简称smt）工具，来快捷生成StarRocks表结构和Flink-SQL映射表及同步语句。Smt目前可用于MySQL、PostgreSQL、Oracle和hive，后面三个数据库的同步还在公测中，我们就先以MySQL来进行演示，后续Release版发布后再逐个补充。

StarRocks-migrate-tools下载地址：

https://cdn-thirdparty.starrocks.com/smt.tar.gzhttps://cdn-thirdparty.starrocks.com/smt.tar.gz

官网操作介绍文档：

设计背景 @ Flink-connector-starrocks @ StarRocks Docshttps://docs.starrocks.com/zh-cn/main/loading/Flink-connector-starrocks#%E4%BD%BF%E7%94%A8-flink-connector-%E5%86%99%E5%85%A5%E5%AE%9E%E7%8E%B0-mysql-%E6%95%B0%E6%8D%AE%E5%90%8C%E6%AD%A5

5.1 MySQL准备

我们在node02已开启binlog的MySQL中创建数据库CDC，并在其中创建表departments和jobs，创建完成后再导入少量数据。

在node02上登陆MySQL：

[root@node02 ~]# mysql -uroot –proot

创建表departments：

mysql> CREATE DATABASE CDC;

mysql> USE CDC;

mysql> CREATE TABLE departments (

department_id int(4) NOT NULL AUTO_INCREMENT,

department_name varchar(3) DEFAULT NULL,

manager_id int(6) DEFAULT NULL,

location_id int(4) DEFAULT NULL,

PRIMARY KEY (department_id)

);

为表departments插入数据：

mysql> insert into departments(department_id,department_name,manager_id,location_id) values (10,'Adm',200,1700),(20,'Mar',201,1800),(30,'Pur',114,1700),(40,'Hum',203,2400),(50,'Shi',121,1500),(60,'IT',103,1400),(70,'Pub',204,2700),(80,'Sal',145,2500),(90,'Exe',100,1700),(100,'Fin',108,1700),(110,'Acc',205,1700),(120,'Tre',NULL,1700),(130,'Cor',NULL,1700),(140,'Con',NULL,1700),(150,'Sha',NULL,1700),(160,'Ben',NULL,1700),(170,'Man',NULL,1700),(180,'Con',NULL,1700),(190,'Con',NULL,1700),(200,'Ope',NULL,1700),(210,'IT ',NULL,1700),(220,'NOC',NULL,1700),(230,'IT ',NULL,1700),(240,'Gov',NULL,1700),(250,'Ret',NULL,1700),(260,'Rec',NULL,1700),(270,'Pay',NULL,1700);

创建表jobs：

mysql> CREATE TABLE jobs (

job_id varchar(10) NOT NULL,

job_title varchar(35) DEFAULT NULL,

min_salary int(6) DEFAULT NULL,

max_salary int(6) DEFAULT NULL,

PRIMARY KEY (job_id)

);

为表jobs插入数据：

mysql> insert into jobs(job_id,job_title,min_salary,max_salary) values ('AC_ACCOUNT','Public Accountant',4200,9000),('AC_MGR','Accounting Manager',8200,16000),('AD_ASST','Administration Assistant',3000,6000),('AD_PRES','President',20000,40000),('AD_VP','Administration Vice President',15000,30000),('FI_ACCOUNT','Accountant',4200,9000),('FI_MGR','Finance Manager',8200,16000),('HR_REP','Human Resources Representative',4000,9000),('IT_PROG','Programmer',4000,10000),('MK_MAN','Marketing Manager',9000,15000),('MK_REP','Marketing Representative',4000,9000),('PR_REP','Public Relations Representative',4500,10500),('PU_CLERK','Purchasing Clerk',2500,5500),('PU_MAN','Purchasing Manager',8000,15000),('SA_MAN','Sales Manager',10000,20000),('SA_REP','Sales Representative',6000,12000),('SH_CLERK','Shipping Clerk',2500,5500),('ST_CLERK','Stock Clerk',2000,5000),('ST_MAN','Stock Manager',5500,8500);

5.2 配置SMT工具

下载smt工具，解压后修改配置文件：

[root@node01 smt]# vi conf/config_prod.conf

首先配置MySQL部分：

[db]

host = 192.168.110.102 #MySQL所在服务器IP

port = 3306 #MySQL服务端口

user = root #用户名

password = root #密码

currently available types: mysql, pgsql, oracle, hive

type = mysql #类型选择MySQL，目前PostgreSQL、Oracle和Hive正在公测中

# 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 #配置StarRocks BE的节点数，以便生成更合理bucket数量的建表语句

decimal_v3 is supported since StarRocks-1.18.1

use_decimal_v3 = true #使用更高精度的Decimal类型，1.18后的版本都支持

file to save the converted DDL SQL

output_dir = ./result #后续生成sql文件的保存目录

!!!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 = CDC #配置需要同步的数据库，需使用正则表达式的写法

pattern to match tables for setting properties

table = departments|jobs #配置需要同步的表，需使用正则表达式的写法

schema only takes effect on postgresql and oracle

schema = ^public$ #同步MySQL时不需要管这个

配置StarRocks集群信息：

############################################

flink sink configurations #这部分与Flink Sink部分写法相似

DO NOT set connector, table-name, database-name, they are auto-generated

############################################

flink.starrocks.jdbc-url=jdbc:mysql://192.168.110.101:9030

flink.starrocks.load-url=192.168.110.101:8030

flink.starrocks.username=root

flink.starrocks.password=root

flink.starrocks.sink.properties.format=json #以json格式攒批

flink.starrocks.sink.properties.strip_outer_array=true #展开为数组

flink.starrocks.sink.buffer-flush.interval-ms=10000 #攒批10秒导入一次

# used to set the server-id for mysql-cdc jobs instead of using a random server-id

flink.cdc.server-id = 5000

5.3 SMT工具使用

前面的配置完成后，执行smt工具：

[root@node01 smt]# ./starrocks-migrate-tool

会在配置的./result路径下生成sql语句文件：

[root@node01 result]# ll

total 24

-rw-r--r-- 1 root root 2229 Jan 19 21:48 flink-create.1.sql

-rw-r--r-- 1 root root 2229 Jan 19 21:48 flink-create.all.sql

-rw-r--r-- 1 root root 732 Jan 19 21:48 starrocks-create.1.sql

-rw-r--r-- 1 root root 732 Jan 19 21:48 starrocks-create.all.sql

-rw-r--r-- 1 root root 838 Jan 19 21:48 starrocks-external-create.1.sql

-rw-r--r-- 1 root root 838 Jan 19 21:48 starrocks-external-create.all.sql

5.4 生成Flink table并开始同步

[root@node01 ~]# mysql -h192.168.110.101 -P9030 -uroot -proot < /opt/module/smt/result/starrocks-create.all.sql

[root@node01 bin]# ./sql-client.sh -f /opt/module/smt/result/flink-create.all.sql

5.5 观察任务状况

[root@node01 bin]# ./flink list

Waiting for response...

------------------ Running/Restarting Jobs -------------------

19.01.2022 21:55:30 : 80c4e81de2d0d7e34c8f1aac1c22a8c4 : insert-into_default_catalog.CDC.departments_sink (RUNNING)

19.01.2022 21:55:34 : b2b76afe7d33196a09a274142d9128cf : insert-into_default_catalog.CDC.jobs_sink (RUNNING)

5.6 数据观察

这里就不再演示改变数据了，与场景四中的情况相同，当数据源中的数据变化时，StarRocks中的数据也会同步变化，实现数据的近实时同步。

这个场景特别适合维度表的数据同步，因为当前StarRocks还不支持update语法，我们就可以将数据需要频繁更新的维度表放在MySQL中，使用Flink CDC+SMT实时的在StarRocks中同步数据，实现灵活的多表关联查询。

标签： flink 数据库 mysql
本文转载自: https://blog.csdn.net/ult_me/article/details/122885594
版权归原作者 流木随风 所有， 如有侵权，请联系我们删除。
登录后发布评论

“第3.4章：StarRocks数据导入--Flink Connector与CDC秒级数据同步”的评论:

还没有评论
关于作者
...
overfit同步小助手
文章同步
相关阅读

基于微信小程序的科学健身助手的设计与实现

自动驾驶车载SoC设计功能安全

量化交易系统开发-实时行情自动化交易-3.4.3.4.期货衍生数据

XCVU13P板卡设计原理图：509-基于XCVU13P的4路QSFP28光纤PCIeX16收发卡

近期两篇NeRF/3DGS-based SLAM方案赏析：TS-SLAM and MBA-SLAM

uni-app Vue3语法实现微信小程序样式穿透uview-plus框架

21.3D surface

文章导航