实时计算Flink版：StarRocks

特色功能

EMR的StarRocks支持通过Flink CDC数据摄入作业实现单表的结构和数据同步，实现整库同步或者同一库中的多表结构和数据同步，详情请参见基于实时计算Flink使用CTAS&CDAS功能同步MySQL数据至StarRocks。

语法结构

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

WITH参数

类型映射

代码示例

CREATE TEMPORARY TABLE IF NOT EXISTS `runoob_tbl_source` (
  `runoob_id` BIGINT NOT NULL,
  `runoob_title` STRING NOT NULL,
  `runoob_author` STRING NOT NULL,
  `submission_date` DATE NULL
) WITH (
  'connector' = 'starrocks',
  'jdbc-url' = 'jdbc:mysql://ip:9030',
  'scan-url' = 'ip:18030',
  'database-name' = 'db_name',
  'table-name' = 'table_name',
  'password' = 'xxxxxxx',
  'username' = 'xxxxx'
);
CREATE TEMPORARY TABLE IF NOT EXISTS `runoob_tbl_sink` (
  `runoob_id` BIGINT NOT NULL,
  `runoob_title` STRING NOT NULL,
  `runoob_author` STRING NOT NULL,
  `submission_date` DATE NULL
  PRIMARY KEY(`runoob_id`)
  NOT ENFORCED
) WITH (
  'jdbc-url' = 'jdbc:mysql://ip:9030',
  'connector' = 'starrocks',
  'load-url' = 'ip:18030',
  'database-name' = 'db_name',
  'table-name' = 'table_name',
  'password' = 'xxxxxxx',
  'username' = 'xxxx',
  'sink.buffer-flush.interval-ms' = '5000'
);

INSERT INTO runoob_tbl_sink SELECT * FROM runoob_tbl_source;

特色功能

• 自动建库建表。

  如果来自上游的数据库及数据表不存在于下游StarRocks实例中，则对应的数据库及数据表会被自动创建。您可以通过table.create.properties.*参数设定自动创建表时的选项。

• 表结构变更同步。

  目前，StarRocks连接器支持自动将建表事件（CreateTableEvent）、增加列事件（AddColumnEvent）和删除列（DropColumnEvent）事件自动应用到下游数据库中。

• 实时计算引擎VVR 11.1及以上版本支持兼容的列类型变更，详情请参见ALTER TABLE | StarRocks。

语法结构

source:
  ...

sink:
  type: starrocks
  name: StarRocks Sink
  jdbc-url: jdbc:mysql://127.0.0.1:9030
  load-url: 127.0.0.1:8030
  username: root
  password: pass
  sink.buffer-flush.interval-ms: 5000   # 设定 Flush 数据的间隔

配置项

复用已有 Catalog

自VVR 11.5版本起，您可以在Flink CDC数据摄入作业中直接引用“数据管理”页面中创建的内置StarRocks Catalog，减少手写连接属性工作量。

sink:
  type: starrocks
  using.built-in-catalog: starrocks_catalog

目前，数据摄入作业支持自动复用以下 StarRocks Catalog 参数：

• jdbc-url

• http-url

• username

• password

• table.num-buckets

如果希望覆盖以上自动复用的参数，可显式写出相应的 YAML 参数，其具备更高的优先级。

类型映射

表结构变更

目前，StarRocks作为数据摄入目标端支持以下表结构变更事件：

• CREATE TABLE EVENT

  说明

  在下游 StarRocks 表已经存在时，不会尝试重复建表。您需要保证下游表结构与上游兼容。

• ADD COLUMN EVENT

  说明

  StarRocks 要求主键列总是位于最前。新插入的列也需要保持此限制。

• ALTER COLUMN TYPE EVENT

  说明

  支持的表结构变更路径，请参考 StarRocks 官方文档。

• DROP COLUMN EVENT

• TRUNCATE TABLE EVENT

• DROP TABLE EVENT

单表同步

将 MySQL 中的一张表同步到 StarRocks，下游库表不存在时会自动创建为主键表。

pipeline:
    name: MySQL to StarRocks Pipeline
source:
  type: mysql
  name: MySQL Source
  hostname: <yourHostname>
  port: 3306
  username: <yourUsername>
  password: ${secret_values.mysql_password}
  tables: test_db.test_source_table
  server-id: 5401-5499
  # （可选）增量阶段实时同步新创建表的数据，无需重启作业
  scan.binlog.newly-added-table.enabled: true
  # （可选）向下游同步表注释和字段注释
  include-comments.enabled: true
  # （可选）只解析被捕获表的 binlog，加速读取
  scan.only.deserialize.captured.tables.changelog.enabled: true

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

pipeline:
  name: MySQL to StarRocks Pipeline

整库同步

将 MySQL 一个库中的所有表一次性同步到 StarRocks，下游自动建库、建主键表，无需提前逐表建表。

source:
  type: mysql
  name: MySQL Source
  hostname: <yourHostname>
  port: 3306
  username: <yourUsername>
  password: ${secret_values.mysql_password}
  # 正则匹配整个库的表；多个库可用英文逗号分隔写多个模式
  tables: test_db.\.*
  server-id: 5401-5499
  # （可选）增量阶段实时同步新创建表的数据，无需重启作业
  scan.binlog.newly-added-table.enabled: true
  # （可选）向下游同步表注释和字段注释
  include-comments.enabled: true

sink:
  type: starrocks
  name: StarRocks Sink
  jdbc-url: jdbc:mysql://<yourFeHostname>:9030
  load-url: <yourFeHostname>:8030
  username: <yourUsername>
  password: ${secret_values.starrocks_password}
  # （可选）数据量不大的作业建议调低 flush 间隔，避免数据长时间不落盘（默认 300000，即 5 分钟）
  sink.buffer-flush.interval-ms: 5000
  # （可选）上游为 utf8mb4 字符集时建议设为 4，避免文本截断（默认 3）
  unicode-char.max-bytes: 4
  # （可选）自动建表的分桶数；StarRocks 2.5.7 以下必须显式配置，更高版本可自动推断
  table.create.num-buckets: 8
  # （可选）自动建表的副本数，按集群情况配置
  table.create.properties.replication_num: 3
  # （可选）StarRocks 3.2 及以上建议开启，加速表结构变更
  table.create.properties.fast_schema_evolution: true

pipeline:
  name: MySQL to StarRocks Pipeline

整库同步时排除部分表

整库同步时用正则跳过不需要同步到下游的表，例如临时表、敏感表。

source:
  type: mysql
  name: MySQL Source
  hostname: <yourHostname>
  port: 3306
  username: <yourUsername>
  password: ${secret_values.mysql_password}
  tables: test_db.\.*
  # 此正则命中的表都不会被同步
  tables.exclude: test_db.tmp_.\*
  server-id: 5401-5499

sink:
  type: starrocks
  name: StarRocks Sink
  jdbc-url: jdbc:mysql://<yourFeHostname>:9030
  load-url: <yourFeHostname>:8030
  username: <yourUsername>
  password: ${secret_values.starrocks_password}
  # （可选）导入接口版本：V2 需 StarRocks 2.4 及以上；EMR Serverless 如遇串流可改为 V1
  sink.version: V2
  # （可选）数据量不大的作业建议调低，避免数据长时间不落盘（默认 300000，即 5 分钟）
  sink.buffer-flush.interval-ms: 5000
  # （可选）自动建表的分桶数；StarRocks 2.5.7 以下必须显式配置
  table.create.num-buckets: 8
  # （可选）StarRocks 3.2 及以上建议开启，加速表结构变更
  table.create.properties.fast_schema_evolution: true

pipeline:
  name: MySQL to StarRocks Pipeline

同步到指定库表

下游 StarRocks 的库名、表名需要与上游不一致时（例如写入 ODS 层库），用 route 统一重命名。

source:
  type: mysql
  name: MySQL Source
  hostname: <yourHostname>
  port: 3306
  username: <yourUsername>
  password: ${secret_values.mysql_password}
  tables: test_db.\.*
  server-id: 5401-5499

sink:
  type: starrocks
  name: StarRocks Sink
  jdbc-url: jdbc:mysql://<yourFeHostname>:9030
  load-url: <yourFeHostname>:8030
  username: <yourUsername>
  password: ${secret_values.starrocks_password}
  # （可选）数据量不大的作业建议调低，避免数据长时间不落盘（默认 300000，即 5 分钟）
  sink.buffer-flush.interval-ms: 5000
  # （可选）自动建表的分桶数；StarRocks 2.5.7 以下必须显式配置
  table.create.num-buckets: 8

route:
  # 将 MySQL test_db 中的所有表同步到 StarRocks test_db2 库，表名保持不变；
  # <> 为占位符，会被匹配到的源表名替换
  - source-table: test_db.\.*
    sink-table: test_db2.<>
    replace-symbol: <>

pipeline:
  name: MySQL to StarRocks Pipeline

分库分表合并

将多张结构相同的分表合并写入一张 StarRocks 表，统一查询分析；要求各分表 schema 一致。

source:
  type: mysql
  name: MySQL Source
  hostname: <yourHostname>
  port: 3306
  username: <yourUsername>
  password: ${secret_values.mysql_password}
  # 匹配所有分表，如 user_0、user_1……
  tables: test_db.user\.*
  server-id: 5401-5499

sink:
  type: starrocks
  name: StarRocks Sink
  jdbc-url: jdbc:mysql://<yourFeHostname>:9030
  load-url: <yourFeHostname>:8030
  username: <yourUsername>
  password: ${secret_values.starrocks_password}
  # （可选）数据量不大的作业建议调低，避免数据长时间不落盘（默认 300000，即 5 分钟）
  sink.buffer-flush.interval-ms: 5000
  # （可选）合并表建议显式指定分桶数，按合并后的数据量评估
  table.create.num-buckets: 8

route:
  # 所有分表合并到一张 StarRocks test_db.user 表
  - source-table: test_db.user\.*
    sink-table: test_db.user

pipeline:
  name: MySQL to StarRocks Pipeline

开启 EVOLVE 模式

默认（LENIENT 模式）下，删列、删表、清空表等结构变更不会同步到下游；如确需严格同步表结构，可开启 EVOLVE 模式。该模式限制明显，使用前务必确认以下注意事项。

限制与注意事项

• 不支持列改名：上游出现列改名事件时作业会直接失败。

• 删列、删表、清空表会在下游真实执行：上游的误操作会直接影响下游表；默认 LENIENT 模式下删表、清空表事件不会同步，更安全。

• 在未删除结果表的情况下无状态重启，可能出现上游与结果表结构不一致导致作业失败，需要手动调整下游表结构。

source:
  type: mysql
  name: MySQL Source
  hostname: <yourHostname>
  port: 3306
  username: <yourUsername>
  password: ${secret_values.mysql_password}
  tables: test_db.test_source_table
  server-id: 5401-5499

sink:
  type: starrocks
  name: StarRocks Sink
  jdbc-url: jdbc:mysql://<yourFeHostname>:9030
  load-url: <yourFeHostname>:8030
  username: <yourUsername>
  password: ${secret_values.starrocks_password}

pipeline:
  name: MySQL to StarRocks Pipeline
  # 开启 EVOLVE：严格同步表结构变更，遇到不支持的变更（如列改名）作业失败
  schema.change.behavior: evolve