流计算 Oceanus Doris 连接器

Doris 连接器通过 Stream Load 把上游变更事件写入 Apache Doris，支持自动建库建表、表结构变更同步以及自动分区表，通过 YAML 作业定义即可完成一条数据同步链路。
﻿
前置准备
使用 Doris Sink 前，需要确保 Doris 集群满足下列条件：
﻿
FE 的 HTTP 端口（默认 8030）对作业运行环境可访问。
数据库账号具备目标库的建库、建表、读写权限。目标库或目标表不存在时，连接器会自动创建。
集群已开启 Stream Load 导入能力。
需要使用自动分区表时，Doris 版本需大于 2.1.6。
﻿
版本说明
Flink 版本
说明
1.20
基于 Flink-cdc 社区 release-3.6 开发，JDK 版本为 11，支持 Doris 版本为 1.x、2.x、3.x
使用范围
可作为 Sink（type: doris）使用，不支持作为 Source。
﻿
快速示例
MySQL 同步到 Doris
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: doris
  name: Doris Sink
  fenodes: 127.0.0.1:8030
  username: root
  password: pass
  table.create.properties.replication_num: 1
  sink.buffer-flush.interval: 10s
  sink.buffer-flush.max-rows: 50000
﻿
pipeline:
  name: MySQL to Doris Pipeline
  parallelism: 4
﻿
Sink 配置项
参数
是否必填
默认值
类型
说明
fenodes
required
(none)
String
Doris 集群 FE 的 HTTP 地址，例如 127.0.0.1:8030。多个地址用英文逗号分隔。
username
required
(none)
String
Doris 集群的用户名。
password
optional
(none)
String
Doris 集群的密码。无密码时填空字符串 ""。
benodes
optional
(none)
String
Doris 集群 BE 的 HTTP 地址，例如 127.0.0.1:8040。开启 auto-redirect 后可不配置。
jdbc-url
optional
(none)
String
Doris 集群的 JDBC 地址，例如 jdbc:mysql://127.0.0.1:9030/db。 当前版本的写入与建表均通过 FE 的 HTTP 接口完成，该参数通常无需配置。
auto-redirect
optional
false
Boolean
是否通过 FE 重定向直连 BE 写入。开启后无需显式配置 BE 地址列表。
charset-encoding
optional
UTF-8
String
Doris HTTP 客户端的字符集编码。
sink.enable.batch-mode
optional
true
Boolean
是否采用攒批方式写入。默认开启，通过缓存攒批后批量提交，适合大吞吐场景； 关闭后改为逐条流式写入。
sink.buffer-flush.max-rows
optional
50000
Integer
单个批次最多攒积的记录数，达到后触发写入。
sink.buffer-flush.max-bytes
optional
10485760(10MB)
Integer
单个批次最多攒积的字节数，达到后触发写入。
sink.buffer-flush.interval
optional
10s
Duration
攒批的最长等待时间，超过该时长后由异步线程提交已缓存的数据。
table.create.properties.*
optional
(none)
String
自动建表时附加到表上的 Doris 属性。
例如 table.create.properties.replication_num: 1、 table.create.properties.light_schema_change: true。
属性名去掉该前缀后作为 Doris 表属性传入。
sink.enable-2pc
optional
false
Boolean
是否开启两阶段提交，用于保证写入的精确一次语义。开启后会带来额外的提交开销。
sink.enable-delete
optional
true
Boolean
是否同步删除事件。关闭后上游的删除操作不会传递到 Doris。
该能力依赖目标表的 Unique 模型，且要求上游表包含主键。
sink.ignore.update-before
optional
true
Boolean
是否忽略 update-before 数据。上游主键与下游主键不一致时，需要把 update-before 作为删除数据传给下游才能正确删除旧数据；默认忽略，即按 upsert 语义处理。
sink.max-retries
optional
3
Integer
写入失败后的最大重试次数。
sink.check-interval
optional
10000
Integer
写入过程中检查异常状态的间隔，单位为毫秒。
sink.buffer-size
optional
1048576(1MB)
Integer
Stream Load 单条写入请求的缓存大小，单位为字节。
sink.buffer-count
optional
3
Integer
Stream Load 的缓存个数。
sink.flush.queue-size
optional
2
Integer
异步提交队列的长度。
sink.label-prefix
optional
(空)
String
Stream Load 事务标签的前缀，用于区分不同作业的导入事务。
sink.use-cache
optional
false
Boolean
是否使用缓存以支持断点续传。
sink.properties.*
optional
(none)
String
透传给 Stream Load 的导入参数。
例如 sink.properties.strict_mode: true、 sink.properties.format: json、 sink.properties.read_json_by_line: true。
属性名去掉该前缀后作为 Stream Load 属性传入。
﻿
自动分区表
通过 `table.create.auto-partition.properties.*` 前缀的参数，可以让连接器在建表时创建 Doris 的 AUTO RANGE PARTITION 分区表。分区键所在列必须是 DATE、TIMESTAMP、TIMESTAMP_LTZ 或 TIMESTAMP WITH TIME ZONE 类型，分区函数为 `date_trunc`，且 Doris 版本必须大于 2.1.6。
参数
是否必填
说明
table.create.auto-partition.properties.default-partition-key
optional
默认分区键（列名）。所有未单独指定分区键的表都使用它。
table.create.auto-partition.properties.default-partition-unit
optional
默认分区单位，配合默认分区键使用，例如 year、month、day。
table.create.auto-partition.properties.DB.TABLE.partition-key
optional
特定表的分区键，例如 table.create.auto-partition.properties.mydb.mytable.partition-key。 未配置时取默认分区键。
table.create.auto-partition.properties.DB.TABLE.partition-unit
optional
特定表的分区单位。未配置时取默认分区单位。
table.create.auto-partition.properties.include
optional
需要创建分区表的表集合，格式为 库名.表名，支持正则表达式，多条用英文逗号分隔。 未配置时表示包含所有表。
table.create.auto-partition.properties.exclude
optional
需要排除的表集合，在 include 结果之上做二次排除，用法与 include 相同。
配置示例：
sink:
  type: doris
  name: Doris Sink
  fenodes: 127.0.0.1:8030
  username: root
  password: xxx
  table.create.auto-partition.properties.include: mydb\\.order_\\.*
  table.create.auto-partition.properties.exclude: mydb\\.order_tmp_\\.*
  table.create.auto-partition.properties.default-partition-key: create_time
  table.create.auto-partition.properties.default-partition-unit: month
  table.create.auto-partition.properties.mydb.order_detail.partition-key: order_date
  table.create.auto-partition.properties.mydb.order_detail.partition-unit: day
﻿
使用自动分区表时需要注意：
﻿
分区类型不满足要求，或分区键不存在于表结构中时，该表不会创建分区，也不会导致作业失败。
Doris 的 AUTO RANGE PARTITION 不允许分区列为 NULL。如果分区键的值为空，连接器会自动填充默认值：DATE 类型填充 1970-01-01，TIMESTAMP、TIMESTAMP_LTZ、TIMESTAMP WITH TIME ZONE 类型填充1970-01-01 00:00:00。请选择合适的分区键，避免大量数据落入同一个默认分区。
时间戳类型字段的默认值 0000-00-00 00:00:00 会被改写为 1970-01-01 00:00:00 后再建表。
include 与 exclude 匹配的是经过 route 之后的目标表名。
﻿
表结构变更同步
连接器支持下列表结构变更的自动同步：
变更类型
说明
CREATE TABLE
目标表不存在时自动创建，库不存在时自动创建库。
ADD COLUMN
新增列。列已存在时跳过，不报错。
DROP COLUMN
删除列。列不存在时跳过，不报错。
RENAME COLUMN
重命名列。
ALTER COLUMN TYPE
修改列类型。列不存在时跳过，不报错。
TRUNCATE TABLE
清空表数据。
DROP TABLE
删除表。
ALTER TABLE COMMENT
修改表注释。
自动建表时按下列规则生成 DDL：
﻿
表模型：上游表有主键时使用 Unique 模型，无主键时使用 Duplicate 模型。
键列与分桶列：取上游表的主键；无主键时取表的第一列。
分桶方式：DISTRIBUTED BY HASH(键列) BUCKETS AUTO。
键列类型为 STRING 时会自动降级为 VARCHAR(65533)，因为 Doris 不支持 STRING 作为键列。
配置了自动分区时，追加 AUTO PARTITION BY RANGE (date_trunc(分区键, 分区单位))。
﻿
数据类型映射
Doris 作为 Sink 时，按下表把 CDC 类型转换为目标表的 Doris 字段类型。
CDC type
Doris 字段类型
NOTE
BOOLEAN
BOOLEAN
﻿
TINYINT
TINYINT
﻿
SMALLINT
SMALLINT
﻿
INT
INT
﻿
BIGINT
BIGINT
﻿
FLOAT
FLOAT
﻿
DOUBLE
DOUBLE
﻿
DECIMAL(p, s)
DECIMALV3(p, s) / STRING
p 不超过 38 时使用 DECIMALV3(p, s)，超过 38 时降级为 STRING。
CHAR(n)
CHAR(n*3) / VARCHAR(n*3) / STRING
Doris 以 UTF-8 存储字符串，一个中文字符占 3 字节，因此长度统一乘 3。
n*3 不超过 255 时使用 CHAR(n*3)；超过 255 时按 VARCHAR(n*3) 处理； n*3 达到 65533 时降级为 STRING。
VARCHAR(n)
VARCHAR(n*3) / STRING
同上，长度统一乘 3。n*3 达到 65533 时降级为 STRING。
STRING
STRING
﻿
BINARY(n)
VARBINARY(n)
STRING
Doris 不支持二进制类型，统一映射为 STRING。
DATE
DATEV2
写入格式为 yyyy-MM-dd。
TIME(p)
STRING
Doris 不支持 TIME 类型，统一映射为 STRING。
p 为 0 时写入格式为 HH:mm:ss，p 大于 0 时为 HH:mm:ss.SSS。
TIMESTAMP(p)
DATETIMEV2(p)
精度按 [0, 6] 区间截断，超过 6 时按 6 处理。
写入格式为 yyyy-MM-dd HH:mm:ss.SSSSSS。
TIMESTAMP_LTZ(p)
DATETIMEV2(p)
精度按 [0, 6] 区间截断。
写入前先按 pipeline 的 local-time-zone 换算为本地时间，因此落库结果不受作业运行环境时区影响。
TIMESTAMP WITH TIME ZONE(p)
DATETIMEV2(p)
精度按 [0, 6] 区间截断。
ARRAY
STRING
Doris 不支持 ARRAY 类型，建表时映射为 STRING，写入时以 JSON 数组形式写出。
MAP
STRING
Doris 不支持 MAP 类型，建表时映射为 STRING，写入时以 JSON 对象形式的字符串写出。
ROW
STRING
Doris 不支持 ROW 类型，建表时映射为 STRING，写入时以 JSON 对象形式的字符串写出。
未在上表中出现的 CDC 类型暂不支持，同步包含此类字段的表会导致作业失败。

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

参数	是否必填	默认值	类型	说明
fenodes	required	(none)	String	Doris 集群 FE 的 HTTP 地址，例如 127.0.0.1:8030。多个地址用英文逗号分隔。
username	required	(none)	String	Doris 集群的用户名。
password	optional	(none)	String	Doris 集群的密码。无密码时填空字符串 ""。
benodes	optional	(none)	String	Doris 集群 BE 的 HTTP 地址，例如 127.0.0.1:8040。开启 auto-redirect 后可不配置。
jdbc-url	optional	(none)	String	Doris 集群的 JDBC 地址，例如 jdbc:mysql://127.0.0.1:9030/db。当前版本的写入与建表均通过 FE 的 HTTP 接口完成，该参数通常无需配置。
auto-redirect	optional	false	Boolean	是否通过 FE 重定向直连 BE 写入。开启后无需显式配置 BE 地址列表。
charset-encoding	optional	UTF-8	String	Doris HTTP 客户端的字符集编码。
sink.enable.batch-mode	optional	true	Boolean	是否采用攒批方式写入。默认开启，通过缓存攒批后批量提交，适合大吞吐场景；关闭后改为逐条流式写入。
sink.buffer-flush.max-rows	optional	50000	Integer	单个批次最多攒积的记录数，达到后触发写入。
sink.buffer-flush.max-bytes	optional	10485760(10MB)	Integer	单个批次最多攒积的字节数，达到后触发写入。
sink.buffer-flush.interval	optional	10s	Duration	攒批的最长等待时间，超过该时长后由异步线程提交已缓存的数据。
table.create.properties.*	optional	(none)	String	自动建表时附加到表上的 Doris 属性。例如 table.create.properties.replication_num: 1、 table.create.properties.light_schema_change: true。属性名去掉该前缀后作为 Doris 表属性传入。
sink.enable-2pc	optional	false	Boolean	是否开启两阶段提交，用于保证写入的精确一次语义。开启后会带来额外的提交开销。
sink.enable-delete	optional	true	Boolean	是否同步删除事件。关闭后上游的删除操作不会传递到 Doris。该能力依赖目标表的 Unique 模型，且要求上游表包含主键。
sink.ignore.update-before	optional	true	Boolean	是否忽略 update-before 数据。上游主键与下游主键不一致时，需要把 update-before 作为删除数据传给下游才能正确删除旧数据；默认忽略，即按 upsert 语义处理。
sink.max-retries	optional	3	Integer	写入失败后的最大重试次数。
sink.check-interval	optional	10000	Integer	写入过程中检查异常状态的间隔，单位为毫秒。
sink.buffer-size	optional	1048576(1MB)	Integer	Stream Load 单条写入请求的缓存大小，单位为字节。
sink.buffer-count	optional	3	Integer	Stream Load 的缓存个数。
sink.flush.queue-size	optional	2	Integer	异步提交队列的长度。
sink.label-prefix	optional	(空)	String	Stream Load 事务标签的前缀，用于区分不同作业的导入事务。
sink.use-cache	optional	false	Boolean	是否使用缓存以支持断点续传。
sink.properties.*	optional	(none)	String	透传给 Stream Load 的导入参数。例如 sink.properties.strict_mode: true、 sink.properties.format: json、 sink.properties.read_json_by_line: true。属性名去掉该前缀后作为 Stream Load 属性传入。

变更类型	说明
CREATE TABLE	目标表不存在时自动创建，库不存在时自动创建库。
ADD COLUMN	新增列。列已存在时跳过，不报错。
DROP COLUMN	删除列。列不存在时跳过，不报错。
RENAME COLUMN	重命名列。
ALTER COLUMN TYPE	修改列类型。列不存在时跳过，不报错。
TRUNCATE TABLE	清空表数据。
DROP TABLE	删除表。
ALTER TABLE COMMENT	修改表注释。

CDC type	Doris 字段类型	NOTE
BOOLEAN	BOOLEAN
TINYINT	TINYINT
SMALLINT	SMALLINT
INT	INT
BIGINT	BIGINT
FLOAT	FLOAT
DOUBLE	DOUBLE
DECIMAL(p, s)	DECIMALV3(p, s) / STRING	p 不超过 38 时使用 DECIMALV3(p, s)，超过 38 时降级为 STRING。
CHAR(n)	CHAR(n3) / VARCHAR(n3) / STRING	Doris 以 UTF-8 存储字符串，一个中文字符占 3 字节，因此长度统一乘 3。 n3 不超过 255 时使用 CHAR(n3)；超过 255 时按 VARCHAR(n3) 处理； n3 达到 65533 时降级为 STRING。
VARCHAR(n)	VARCHAR(n*3) / STRING	同上，长度统一乘 3。n*3 达到 65533 时降级为 STRING。
STRING	STRING
BINARY(n) VARBINARY(n)	STRING	Doris 不支持二进制类型，统一映射为 STRING。
DATE	DATEV2	写入格式为 yyyy-MM-dd。
TIME(p)	STRING	Doris 不支持 TIME 类型，统一映射为 STRING。 p 为 0 时写入格式为 HH:mm:ss，p 大于 0 时为 HH:mm:ss.SSS。
TIMESTAMP(p)	DATETIMEV2(p)	精度按 [0, 6] 区间截断，超过 6 时按 6 处理。写入格式为 yyyy-MM-dd HH:mm:ss.SSSSSS。
TIMESTAMP_LTZ(p)	DATETIMEV2(p)	精度按 [0, 6] 区间截断。写入前先按 pipeline 的 local-time-zone 换算为本地时间，因此落库结果不受作业运行环境时区影响。
TIMESTAMP WITH TIME ZONE(p)	DATETIMEV2(p)	精度按 [0, 6] 区间截断。
ARRAY	STRING	Doris 不支持 ARRAY 类型，建表时映射为 STRING，写入时以 JSON 数组形式写出。
MAP	STRING	Doris 不支持 MAP 类型，建表时映射为 STRING，写入时以 JSON 对象形式的字符串写出。
ROW	STRING	Doris 不支持 ROW 类型，建表时映射为 STRING，写入时以 JSON 对象形式的字符串写出。

Doris 连接器

本页目录：

前置准备

版本说明

使用范围

快速示例

MySQL 同步到 Doris

Sink 配置项

自动分区表

表结构变更同步

数据类型映射