流计算 Oceanus StarRocks 连接器

StarRocks 连接器通过 Stream Load 把上游变更事件写入 StarRocks，支持自动建库建表、表结构变更同步以及数据实时同步，通过 YAML 作业定义即可完成一条数据同步链路。
﻿
前置准备
使用 StarRocks Sink 前，需要确保 StarRocks 集群满足下列条件：
﻿
FE 的 MySQL 协议端口（默认 9030）与 HTTP 端口（默认 8030）对作业运行环境均可访问。
数据库账号具备目标库的建库、建表、读写权限。目标库或目标表不存在时，连接器会自动创建。
集群已开启 Stream Load 导入能力。
源表必须包含主键。连接器只支持 StarRocks 主键表，源表没有主键时作业会失败。
﻿
版本说明
Flink 版本
说明
1.20
基于 Flink-cdc 社区 release-3.6 开发，JDK 版本为 11，支持 StarRocks 版本为 2.x、3.x
使用范围
可作为 Sink（type: starrocks）使用，不支持作为 Source。
﻿
快速示例
MySQL 同步到 StarRocks
source:
  type: mysql
  name: MySQL Source
  hostname: 127.0.0.1
  port: 3306
  username: admin
  password: pass
  tables: adb.\\.*
  server-id: 5401-5404
  server-time-zone: Asia/Shanghai
﻿
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
  table.create.properties.replication_num: 1
﻿
pipeline:
  name: MySQL to StarRocks Pipeline
  parallelism: 4
﻿
Sink 配置项
参数
是否必填
默认值
类型
说明
jdbc-url
required
(none)
String
访问 FE 上 MySQL 服务的地址，用于建表、表结构变更等元数据操作。
多个地址用英文逗号（,）分隔，格式为 jdbc:mysql://fe_host1:9030,fe_host2:9030。
load-url
required
(none)
String
访问 FE 上 HTTP 服务的地址，用于 Stream Load 写入。
多个地址用英文分号（;）分隔，格式为 fe_host1:8030;fe_host2:8030。
未写 http/https 前缀时默认使用 http，也可以显式写成 https://fe_nlb。
username
required
(none)
String
StarRocks 集群的用户名。
password
required
(none)
String
StarRocks 集群的密码。无密码时填空字符串 ""。
sink.label-prefix
optional
(none)
String
Stream Load 事务标签的前缀，用于区分不同作业的导入事务。
可用字符范围为 [-_A-Za-z0-9]。
sink.connect.timeout-ms
optional
30000
Integer
与 FE 建立 HTTP 连接的超时时间，单位为毫秒。
sink.socket.timeout-ms
optional
-1
Integer
HTTP 客户端等待数据的超时时间，单位为毫秒。默认值 -1 表示不超时。
sink.wait-for-continue.timeout-ms
optional
30000
Integer
等待 FE 返回 HTTP 100-continue 应答的超时时间，单位为毫秒。
sink.buffer-flush.max-bytes
optional
157286400(150MB)
Long
内存中缓冲的数据量大小。缓冲区由所有写入的表共享，达到阈值后会选择一个或多个表的数据写入 StarRocks。
sink.buffer-flush.max-rows
optional
500000
Long
单批次最多写入的行数。
sink.buffer-flush.interval-ms
optional
300000
Long
每个表缓冲数据的发送间隔，用于控制写入延迟，单位为毫秒。
sink.scan-frequency.ms
optional
50
Long
检查各表是否到达发送间隔的频率，单位为毫秒。
sink.io.thread-count
optional
2
Integer
执行 Stream Load 的线程数，不同表之间可以并发导入。
sink.at-least-once.use-transaction-stream-load
optional
true
Boolean
可用时是否使用事务 Stream Load 来保证 at-least-once 语义。
sink.metric.histogram-window-size
optional
100
Integer
直方图指标的窗口大小。
sink.datetime.fractional-seconds.enabled
optional
false
Boolean
序列化记录时是否保留 DATETIME 的小数秒部分。
开启后按源列精度输出小数位（例如 2024-01-15 10:30:00.123）， StarRocks 3.3.5 及以上版本会保留该精度，更早的版本会截断。
关闭时（默认）截断到秒，与历史行为一致。
sink.properties.*
optional
(none)
String
透传给 Stream Load 的导入参数，例如 sink.properties.timeout: 600、 sink.properties.partial_update: 'true'。
属性名去掉该前缀后作为 Stream Load 属性传入。
连接器已固定设置 format=json、strip_outer_array=true、 ignore_json_size=true，无需再自行配置。
名称与值均按原样传给 StarRocks，不做大小写转换。
table.create.num-buckets
optional
(none)
Integer
自动建表时使用的分桶数。未配置时由 StarRocks 自行决定。
使用 StarRocks 2.5 之前的版本时必须配置，否则无法自动建表。
table.create.properties.*
optional
(none)
String
自动建表时附加到表上的 StarRocks 属性，例如 table.create.properties.replication_num: 1、 table.create.properties.fast_schema_evolution: 'true'。
属性名去掉该前缀并转为小写后作为表属性传入。
table.schema-change.timeout
optional
1800s
Duration
StarRocks 侧执行表结构变更的超时时间，必须是秒的整数倍。
超时后 StarRocks 会取消该变更并导致作业失败。
table.schema-change.merge-existing-schema
optional
false
Boolean
执行新增列、修改列类型前，是否先通过 JDBC 读取 StarRocks 侧的实际表结构并与变更合并。
合并时会用现有结构覆盖可空性、默认值，以及列长度（仅当现有长度更大时）。
合并后若与目标状态完全一致，则跳过该次变更，可避免重复变更导致作业失败。
table.schema-change.flush-current-table-only
optional
false
Boolean
应用表结构变更前，是否只冲刷该表已缓冲的数据，而不是冲刷所有表的数据。
表结构变更较频繁时开启该参数，可以避免因频繁冲刷产生大量小规模的 Stream Load。
表结构变更同步
连接器支持下列表结构变更的自动同步：
变更类型
说明
CREATE TABLE
自动建库建表。源表必须有主键，否则建表失败。
ADD COLUMN
新增列。只能追加到最后一列，源端指定的列位置信息会被忽略。
重复新增同名列时，若目标表已存在该列则视为成功，不会导致作业失败。
DROP COLUMN
删除列。重复删除已不存在的列时视为成功，不会导致作业失败。
RENAME COLUMN
重命名列。
ALTER COLUMN TYPE
修改列类型。
TRUNCATE TABLE
清空表数据。
DROP TABLE
删除表。
修改表注释暂不支持，收到该事件时连接器会记录告警并跳过。
﻿
自动建表时按下列规则生成 DDL：
﻿
表模型固定为主键表（PRIMARY KEY）。
主键列会被排到表结构的最前面，其余列保持原有相对顺序。
分桶键与主键相同。
不设置分区键。
分桶数由 table.create.num-buckets 控制；未配置时由 StarRocks 自行决定。
﻿
数据类型映射
StarRocks 作为 Sink 时，按下表把 CDC 类型转换为目标表的 StarRocks 字段类型。
CDC type
StarRocks 字段类型
NOTE
BOOLEAN
BOOLEAN
﻿
TINYINT
TINYINT
﻿
SMALLINT
SMALLINT
﻿
INT
INT
﻿
BIGINT
BIGINT
﻿
FLOAT
FLOAT
﻿
DOUBLE
DOUBLE
﻿
DECIMAL(p, s)
DECIMAL(p, s) / VARCHAR
非主键列使用 DECIMAL(p, s)，注意 StarRocks 的 DECIMAL 精度上限为 38，超过后建表会失败。
StarRocks 不支持 DECIMAL 作为主键列，因此主键列降级为 VARCHAR， 长度为 min(p + 2, 1048576)（s 为 0 时取 min(p + 1, 1048576)）， 多出的 2 位用于容纳负号与小数点。
CHAR(n)
CHAR(n*3) / VARCHAR(n*3)
StarRocks 以 UTF-8 存储字符串，一个中文字符占 3 字节，因此长度统一乘 3。
非主键列且 n*3 不超过 255（即 n 不超过 85）时使用 CHAR(n*3)；
超过时使用 VARCHAR(min(n*3, 1048576))。
主键列一律使用 VARCHAR，因为 StarRocks 主键列不支持 CHAR。
VARCHAR(n)
VARCHAR(min(n*3, 1048576))
长度统一乘 3，上限为 1048576。
STRING
VARCHAR(1048576)
按 VARCHAR 的最大长度处理。
BINARY(n)
VARBINARY(n)
VARBINARY(min(n, 1048576))
长度上限为 1048576。
BYTES
VARBINARY(1048576)
按 VARBINARY 的最大长度处理。
DATE
DATE
写入格式为 yyyy-MM-dd。
TIME(p)
VARCHAR
StarRocks 不支持 TIME 类型，映射为 VARCHAR。
p 为 0 时长度为 8，写入格式为 HH:mm:ss；
p 大于 0 时长度为 8 + 1 + p，写入格式为 HH:mm:ss.<p 位小数>。
TIME 的小数秒始终按源列精度输出，不受 sink.datetime.fractional-seconds.enabled 控制。
TIMESTAMP(p)
DATETIME
写入格式为 yyyy-MM-dd HH:mm:ss。
开启 sink.datetime.fractional-seconds.enabled 后按 p 追加小数秒位。
TIMESTAMP_LTZ(p)
DATETIME
写入前先按 pipeline 的 local-time-zone 换算为本地时间； 未配置 local-time-zone 时使用作业运行环境的系统时区。
小数秒行为与 TIMESTAMP 相同。
未在上表中出现的 CDC 类型暂不支持，同步包含此类字段的表会导致作业失败。TIMESTAMP WITH TIME ZONE、ARRAY、MAP、ROW 均不支持。

Flink 版本	说明
1.20	基于 Flink-cdc 社区 release-3.6 开发，JDK 版本为 11，支持 StarRocks 版本为 2.x、3.x

参数	是否必填	默认值	类型	说明
jdbc-url	required	(none)	String	访问 FE 上 MySQL 服务的地址，用于建表、表结构变更等元数据操作。多个地址用英文逗号（,）分隔，格式为 jdbc:mysql://fe_host1:9030,fe_host2:9030。
load-url	required	(none)	String	访问 FE 上 HTTP 服务的地址，用于 Stream Load 写入。多个地址用英文分号（;）分隔，格式为 fe_host1:8030;fe_host2:8030。未写 http/https 前缀时默认使用 http，也可以显式写成 https://fe_nlb。
username	required	(none)	String	StarRocks 集群的用户名。
password	required	(none)	String	StarRocks 集群的密码。无密码时填空字符串 ""。
sink.label-prefix	optional	(none)	String	Stream Load 事务标签的前缀，用于区分不同作业的导入事务。可用字符范围为 [-_A-Za-z0-9]。
sink.connect.timeout-ms	optional	30000	Integer	与 FE 建立 HTTP 连接的超时时间，单位为毫秒。
sink.socket.timeout-ms	optional	-1	Integer	HTTP 客户端等待数据的超时时间，单位为毫秒。默认值 -1 表示不超时。
sink.wait-for-continue.timeout-ms	optional	30000	Integer	等待 FE 返回 HTTP 100-continue 应答的超时时间，单位为毫秒。
sink.buffer-flush.max-bytes	optional	157286400(150MB)	Long	内存中缓冲的数据量大小。缓冲区由所有写入的表共享，达到阈值后会选择一个或多个表的数据写入 StarRocks。
sink.buffer-flush.max-rows	optional	500000	Long	单批次最多写入的行数。
sink.buffer-flush.interval-ms	optional	300000	Long	每个表缓冲数据的发送间隔，用于控制写入延迟，单位为毫秒。
sink.scan-frequency.ms	optional	50	Long	检查各表是否到达发送间隔的频率，单位为毫秒。
sink.io.thread-count	optional	2	Integer	执行 Stream Load 的线程数，不同表之间可以并发导入。
sink.at-least-once.use-transaction-stream-load	optional	true	Boolean	可用时是否使用事务 Stream Load 来保证 at-least-once 语义。
sink.metric.histogram-window-size	optional	100	Integer	直方图指标的窗口大小。
sink.datetime.fractional-seconds.enabled	optional	false	Boolean	序列化记录时是否保留 DATETIME 的小数秒部分。开启后按源列精度输出小数位（例如 2024-01-15 10:30:00.123）， StarRocks 3.3.5 及以上版本会保留该精度，更早的版本会截断。关闭时（默认）截断到秒，与历史行为一致。
sink.properties.*	optional	(none)	String	透传给 Stream Load 的导入参数，例如 sink.properties.timeout: 600、 sink.properties.partial_update: 'true'。属性名去掉该前缀后作为 Stream Load 属性传入。连接器已固定设置 format=json、strip_outer_array=true、 ignore_json_size=true，无需再自行配置。名称与值均按原样传给 StarRocks，不做大小写转换。
table.create.num-buckets	optional	(none)	Integer	自动建表时使用的分桶数。未配置时由 StarRocks 自行决定。使用 StarRocks 2.5 之前的版本时必须配置，否则无法自动建表。
table.create.properties.*	optional	(none)	String	自动建表时附加到表上的 StarRocks 属性，例如 table.create.properties.replication_num: 1、 table.create.properties.fast_schema_evolution: 'true'。属性名去掉该前缀并转为小写后作为表属性传入。
table.schema-change.timeout	optional	1800s	Duration	StarRocks 侧执行表结构变更的超时时间，必须是秒的整数倍。超时后 StarRocks 会取消该变更并导致作业失败。
table.schema-change.merge-existing-schema	optional	false	Boolean	执行新增列、修改列类型前，是否先通过 JDBC 读取 StarRocks 侧的实际表结构并与变更合并。合并时会用现有结构覆盖可空性、默认值，以及列长度（仅当现有长度更大时）。合并后若与目标状态完全一致，则跳过该次变更，可避免重复变更导致作业失败。
table.schema-change.flush-current-table-only	optional	false	Boolean	应用表结构变更前，是否只冲刷该表已缓冲的数据，而不是冲刷所有表的数据。表结构变更较频繁时开启该参数，可以避免因频繁冲刷产生大量小规模的 Stream Load。

变更类型	说明
CREATE TABLE	自动建库建表。源表必须有主键，否则建表失败。
ADD COLUMN	新增列。只能追加到最后一列，源端指定的列位置信息会被忽略。重复新增同名列时，若目标表已存在该列则视为成功，不会导致作业失败。
DROP COLUMN	删除列。重复删除已不存在的列时视为成功，不会导致作业失败。
RENAME COLUMN	重命名列。
ALTER COLUMN TYPE	修改列类型。
TRUNCATE TABLE	清空表数据。
DROP TABLE	删除表。

CDC type	StarRocks 字段类型	NOTE
BOOLEAN	BOOLEAN
TINYINT	TINYINT
SMALLINT	SMALLINT
INT	INT
BIGINT	BIGINT
FLOAT	FLOAT
DOUBLE	DOUBLE
DECIMAL(p, s)	DECIMAL(p, s) / VARCHAR	非主键列使用 DECIMAL(p, s)，注意 StarRocks 的 DECIMAL 精度上限为 38，超过后建表会失败。 StarRocks 不支持 DECIMAL 作为主键列，因此主键列降级为 VARCHAR，长度为 min(p + 2, 1048576)（s 为 0 时取 min(p + 1, 1048576)），多出的 2 位用于容纳负号与小数点。
CHAR(n)	CHAR(n3) / VARCHAR(n3)	StarRocks 以 UTF-8 存储字符串，一个中文字符占 3 字节，因此长度统一乘 3。非主键列且 n3 不超过 255（即 n 不超过 85）时使用 CHAR(n3)；超过时使用 VARCHAR(min(n*3, 1048576))。主键列一律使用 VARCHAR，因为 StarRocks 主键列不支持 CHAR。
VARCHAR(n)	VARCHAR(min(n*3, 1048576))	长度统一乘 3，上限为 1048576。
STRING	VARCHAR(1048576)	按 VARCHAR 的最大长度处理。
BINARY(n) VARBINARY(n)	VARBINARY(min(n, 1048576))	长度上限为 1048576。
BYTES	VARBINARY(1048576)	按 VARBINARY 的最大长度处理。
DATE	DATE	写入格式为 yyyy-MM-dd。
TIME(p)	VARCHAR	StarRocks 不支持 TIME 类型，映射为 VARCHAR。 p 为 0 时长度为 8，写入格式为 HH:mm:ss； p 大于 0 时长度为 8 + 1 + p，写入格式为 HH:mm:ss.<p 位小数>。 TIME 的小数秒始终按源列精度输出，不受 sink.datetime.fractional-seconds.enabled 控制。
TIMESTAMP(p)	DATETIME	写入格式为 yyyy-MM-dd HH:mm:ss。开启 sink.datetime.fractional-seconds.enabled 后按 p 追加小数秒位。
TIMESTAMP_LTZ(p)	DATETIME	写入前先按 pipeline 的 local-time-zone 换算为本地时间；未配置 local-time-zone 时使用作业运行环境的系统时区。小数秒行为与 TIMESTAMP 相同。

StarRocks 连接器

本页目录：

前置准备

版本说明

使用范围

快速示例

MySQL 同步到 StarRocks

Sink 配置项

表结构变更同步

数据类型映射