使用Serveless Flink实现MySQL到StarRocks数据集成

1 支持的版本

支持采集 EMR-3.1.1 及以上 StarRocks 集群和 OLAP 服务中 1.1.0 及以上版本全托管 StarRocks 引擎中的数据。

2 使用前提

2.1 准备MySQL的数据库信息

• 准备云数据库MySQL，建立在同一个VPC网络内，您可前往控制台创建MySQL实例

2.2 获取StarRocks的数据库信息

1. StarRocks 数据源配置时，EMR 集群对应的集群信息、数据库用户名密码需填写正确：

2. 填写的数据库用户名信息，必须拥有相应数据库表的读写权限，来保障任务数据能够被正常读取或写入 StarRocks 中。

   1. 必须有账户密码，其中 root 账户无密码，不符合安全规范，数据源配置时无法使用。

   2. 用户名密码获取方式，您可在 EMR 集群详情 ＞ 服务列表 ＞ StarRocks 服务名称 ＞ 服务参数 ＞ starrocks-env 参数文件下，看到 StarRocks 已经预置了一个账户和密码，推荐使用该账户/密码来配置 StarRocks 数据源。 其余用户创建方式详见StarRocks官网 CREATE USER。

   

3. EMR StarRocks 集群和独享集成资源组中的 VPC 必须一致。

   1. Mysql数据源和Serverless Flink资源池 建议在同一个私有网络

   2. 其 VPC 下的子网和安全组也尽可能保持一致。

   3. 若 VPC 不一致时，则需要在 StarRocks 集群的安全组上，在入方向规则处，添加独享集成资源组子网的 IP 网段：

      • 在 EMR StarRocks 集群详情界面，进入集群所在的安全组，并添加入方向规则。

      

      • 在弹窗中，填写独享集成资源组子网的 IP 网段：

      

2.3 创建Serveless Flink集群

• Flink 资源池是项目中用来管理计算资源的，资源池中的计算资源相互隔离，相互独立。任务运行和调试需要消耗计算资源，在开发任务前，需要先创建 Flink 资源池。详见：flink资源池创建。

• 创建Flink任务，参考：开发 Flink SQL 任务。

3 MySQL 导入到StarRocks链路

3.1 准备MySQL数据源

CREATE DATABASE test;

CREATE TABLE test.score_board(
    id int(11) NOT NULL COMMENT "",
    name varchar(255) NULL DEFAULT "" COMMENT "",
    score int(11) NOT NULL DEFAULT "0" COMMENT "",
PRIMARY KEY(id)
);


#准备若干条数据
INSERT INTO `score_board` VALUES (201, 'mysqlinsert',123), (202, 'mysql-uui',345);

3.2 准备StarRocks 表

mysql -uxxx -pxxx -h xxx -P 9030

CREATE DATABASE test;

CREATE TABLE test.score_board(
    id int(11) NOT NULL COMMENT "",
    name varchar(65533) NULL DEFAULT "" COMMENT "",
    score int(11) NOT NULL DEFAULT "0" COMMENT ""
)
ENGINE=OLAP
PRIMARY KEY(id)
DISTRIBUTED BY HASH(id);

StarRocks 连接器目前仅支持在 Flink 1.16-volcano 引擎版本中使用 。

3.3 编写 Flink任务运行导入

starrocks的数据源connector定义如下

CREATE TABLE starrocks_table(
 name VARCHAR,
 score BIGINT
 ) WITH ('connector' = 'starrocks','jdbc-url'='jdbc:mysql://fe1_ip:query_port,fe2_ip:query_port,fe3_ip:query_port','load-url'='fe1_ip:http_port;fe2_ip:http_port;fe3_ip:http_port','database-name' = 'xxx','table-name' = 'xxx','username' = 'xxx','password' = 'xxx'
 );
SQL

编写数据源写入

CREATE TABLE
  mysql_jdbc (`id` INT, `name` STRING, `score` INT)
WITH
  (
    'connector' = 'jdbc',
    'url' = 'jdbc:mysql://xxx.rds.ivolces.com:3306/xxx',
    'table-name' = 'xxx',
    'username' = 'xxx',
    'password' = 'xxx'
  );

CREATE TABLE
  `score_board_sr` (
    `id` INT,
    `name` STRING,
    `score` INT,
    PRIMARY KEY (id) NOT ENFORCED
  )
WITH
  (
    'connector' = 'starrocks',
    'jdbc-url' = 'jdbc:mysql://xxx:9030',
    'load-url' = 'xxx:8030',
    'database-name' = 'xxx',
    'table-name' = 'xxx',
    'username' = 'xxx',
    'password' = 'xxx'
  );

insert into
  score_board_sr
select
  *
from
  mysql_jdbc;

启动任务运行

验证StarRocks端可以看到数据通过flink写入成功

4 详细参数介绍

通用参数

源表参数

结果表参数

数据类型映射

• 源表

• 结果表

其它参考示例

• 结果表示例

CREATE TABLE datagen_source(
  id INTEGER,
  name STRING,
  score INTEGER
) WITH ('connector' = 'datagen','rows-per-second' = '1','fields.name.length' = '9','fields.id.min' = '1','fields.id.max' = '1000','fields.score.min' = '1','fields.score.max' = '1000'
);

CREATE TABLE starrocks_sink(
  id INTEGER,
  name STRING,
  score INTEGER,PRIMARY KEY (id) NOT ENFORCED
) WITH ('connector' = 'starrocks','jdbc-url' = 'jdbc:mysql://172.*.*.*:9030', 'load-url' = '172.*.*.*:8030', 'database-name' = 'xxx','table-name' = 'xxx','username' = 'xxx','password' = 'xxx','sink.parallelism' = '1'
);

INSERT INTO starrocks_sink
SELECT * FROM datagen_source;