StarRocks数据摄入连接器

配置项

复用已有 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 参数，其具备更高的优先级。

支持的事件

• CREATE TABLE EVENT

  说明

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

• ADD COLUMN EVENT

  说明

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

• ALTER COLUMN TYPE EVENT

  说明

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

• RENAME COLUMN EVENT

  说明

  需要实时计算引擎版本高于 VVR 11.9、且 StarRocks 集群版本高于 3.3.2。

• DROP COLUMN EVENT

• TRUNCATE TABLE EVENT

• DROP TABLE EVENT

LENIENT（默认）

LENIENT模式下支持的Schema变更策略详情如下：

• 添加可空列：会自动在结果表Schema末尾添加对应的列，并自动同步新增列的数据。

• 添加非空列：会自动在结果表Schema末尾添加对应的列，新增的列会设置为可空列，对于添加列发生之前的数据自动设置为NULL值。

• 删除列：不会直接在结果表中删除该列。旧列原为非空时，将其变更为可空列；已可空时不产生变更。

• 重命名列：被看作为添加新列并保留旧列。在结果表末尾添加改名后的可空列；旧列原为非空时，将其变更为可空列。

• 列顺序调整：忽略，不同步到下游。

• 列类型变更：不因LENIENT模式自动忽略，仍交给StarRocks执行，仅支持兼容的变更路径。允许的路径请参见ALTER TABLE，VVR 11.1及以上版本的支持范围沿用原帮助文档。

• 删除表、清空表：默认被排除，不同步到下游。

EVOLVE

EVOLVE模式下支持的Schema变更策略详情如下：

• 添加列：支持。StarRocks执行器忽略上游新增位置，追加到已有列尾部。

• 删除列：支持。会在结果表中实际删除该列。

• 重命名列：支持（需要 StarRocks 3.3.2 及更高版本）。会修改下游列的名字。

• 列类型变更：支持。仅支持兼容的变更路径。

• 列顺序调整：支持。要求主键列仍在最前且保持原主键顺序。

• 删除表、清空表：未被排除时实际执行。

单表同步

将 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