Doris Connector

本文介绍了 Pipeline Doris Connector 的使用。

示例

source:
   type: values
   name: ValuesSource

sink:
   type: doris
   name: Doris Sink
   fenodes: 127.0.0.1:8030
   username: root
   password: ""
   table.create.properties.replication_num: 1

pipeline:
   parallelism: 1

连接器配置项

Option	Required	Default	Type	Description
type	required	(none)	String	指定要使用的 Sink, 这里是 'doris'。
name	optional	(none)	String	PipeLine 的名称
fenodes	required	(none)	String	Doris 集群 FE 的 Http 地址，比如 127.0.0.1:8030
benodes	optional	(none)	String	Doris 集群 BE 的 Http 地址，比如 127.0.0.1:8040
jdbc-url	optional	(none)	String	Doris 集群的 JDBC 地址，比如：jdbc:mysql://127.0.0.1:9030/db
username	required	(none)	String	Doris 集群的用户名
password	optional	(none)	String	Doris 集群的密码
auto-redirect	optional	false	Boolean	是否通过 FE 重定向写入，直连 BE 写入
charset-encoding	optional	UTF-8	String	Doris Http 客户端字符集编码，默认 UTF-8
sink.enable.batch-mode	optional	true	Boolean	是否使用攒批方式写入 Doris
sink.enable-delete	optional	true	Boolean	是否启用删除
sink.max-retries	optional	3	Integer	写入失败后最大重试次数
sink.flush.queue-size	optional	2	Integer	攒批写入的队列大小
sink.buffer-flush.max-rows	optional	50000	Integer	单个批次最大 Flush 的记录数
sink.buffer-flush.max-bytes	optional	10485760(10MB)	Integer	单个批次最大 Flush 的字节数
sink.buffer-flush.interval	optional	10s	String	Flush 的间隔时长，超过这个时间，将异步 Flush 数据
sink.ignore.update-before	optional	true	Boolean	在 CDC 场景中，当上游的主键与下游的主键不一致时，需要将 update-before 数据作为已删除数据传递给下游，否则数据将无法被删除。默认设置为忽略，即执行 upsert 语义
sink.enable-2pc	optional	false	Boolean	是否在导入数据时启用两阶段提交（2PC）
sink.check-interval	optional	10000	Integer	导入过程中检查异常的间隔（毫秒）
sink.buffer-size	optional	1048576(1MB)	Integer	Stream Load 缓存数据的缓冲区大小（字节）
sink.buffer-count	optional	3	Integer	Stream Load 缓存数据的缓冲区数量
sink.label-prefix	optional	(none)	String	Stream Load 的唯一 label 前缀
sink.use-cache	optional	false	Boolean	是否为断点续传使用缓冲区缓存
sink.properties.	optional	(none)	String	StreamLoad 的参数。 For example: sink.properties.strict_mode: true. 查看更多关于 StreamLoad 的属性
table.create.properties.*	optional	(none)	String	创建表的 Properties 配置。 For example: table.create.properties.replication_num: 1. 查看更多关于 Doris Table 的属性
table.create.auto-partition.properties.*	optional	(none)	String	创建自动分区表的配置。 当前仅支持 DATE/DATETIME 类型列的 AUTO RANGE PARTITION，分区函数为 date_trunc，且 Doris 版本必须大于 2.1.6，查看更多关于 Doris 自动分区 支持的属性有： table.create.auto-partition.properties.include 包含的经过 route 后的表集合，用逗号分隔，支持正则表达式； table.create.auto-partition.properties.exclude 排除的经过 route 后的表集合，用逗号分隔，支持正则表达式； table.create.auto-partition.properties.default-partition-key 默认分区键； table.create.auto-partition.properties.default-partition-unit 默认分区单位； table.create.auto-partition.properties.DB.TABLE.partition-key 特定表的分区键，如未配置取默认分区键； table.create.auto-partition.properties.DB.TABLE.partition-unit 特定表的分区单位，如未配置取默认分区单位。 注意： 1: 如果分区键不为 DATE/DATETIME 类型，则不会创建分区表。 2: Doris AUTO RANGE PARTITION 不支持 NULLABLE 列作为分区列，如果您配置的分区键的值为空或者表创建完成后新增了 NULLABLE 分区列，系统将自动填充默认值（DATE 类型为 1970-01-01，DATETIME 类型为 1970-01-01 00:00:00），请选择合适的分区键。

数据类型映射

CDC type	Doris type	NOTE
TINYINT	TINYINT
SMALLINT	SMALLINT
INT	INT
BIGINT	BIGINT
DECIMAL	DECIMAL
FLOAT	FLOAT
DOUBLE	DOUBLE
BOOLEAN	BOOLEAN
DATE	DATE
TIMESTAMP [(p)]	DATETIME [(p)]
TIMESTAMP_LTZ [(p)]	DATETIME [(p)]
CHAR(n)	CHAR(n*3)	在 Doris 中，字符串是以 UTF-8 编码存储的，所以英文字符占 1 个字节，中文字符占 3 个字节。这里的长度统一乘 3，CHAR 最大的长度是 255，超过后会自动转为 VARCHAR 类型
VARCHAR(n)	VARCHAR(n*3)	同上，这里的长度统一乘 3，VARCHAR 最大的长度是 65533，超过后会自动转为 STRING 类型
BINARY(n)	STRING
VARBINARY(N)	STRING
TIME	STRING	Doris 不支持 TIME 数据类型，需转换为 STRING 类型以兼容
STRING	STRING