云服务器特惠 99/年起
  1. 腾讯云优惠网首页
  2. 腾讯云产品

流计算 Oceanus Doris 连接器

• 腾讯云产品

Doris 连接器通过 Stream Load 把上游变更事件写入 Apache Doris，支持自动建库建表、表结构变更同步以及自动分区表，通过 YAML 作业定义即可完成一条数据同步链路。

相关产品
流计算 Oceanus 产品页，含版本规格、价格与购买入口。

一、前置准备

使用 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/Shanghaisink: 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: 50000pipeline: 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 类型暂不支持，同步包含此类字段的表会导致作业失败。

相关内容：

  1. 流计算 Oceanus 数据仓库 Doris Flink Connector Doris 目前支持通过 Flink 将数据写入 Doris，基于 开源版本 实现。...
  2. 流计算 Oceanus StarRocks 连接器 StarRocks 连接器通过 Stream Load 把上游变更事件写入 StarRocks，支持自动建库建表、表结构变更同步以及数据实时同步，通过 YAML 作业定义即可完成一条数据同步链路。...
  3. 流计算 Oceanus 数据仓库 StarRocks flink-connector-starrocks 基于开源的 starrocks-connector-for-apache-flink 实现，支持通过 Flink 读写 StarRocks。...
  4. 流计算 Oceanus Kafka 连接器 Kafka 连接器既可以作为 Source（type: kafka）从 Kafka 主题读取变更事件，也可以作为 Sink（type: kafka）把变更事件写入 Kafka 主题。...
  5. 流计算 Oceanus DLC 连接器 DLC 连接器可以作为 Sink（type: iceberg-dlc），把上游的变更事件写入腾讯云数据湖计算（Data Lake Compute，DLC）托管的 Iceberg 表，提供自动建表、表结构变更同步与数据实时同步能力。数据以 Apache Iceberg 格式写入，由 DLC 托管的 catalog 管理元数据。...
  6. 流计算 Oceanus Iceberg 连接器 Iceberg 连接器可以作为 Sink（type: iceberg），把上游的变更事件写入 Apache Iceberg 湖存储，提供自动建表、表结构变更同步与数据实时同步能力。...
  7. 流计算 Oceanus Paimon 连接器 Paimon 连接器可以作为 Sink，把上游的变更事件写入 Apache Paimon 湖存储，提供自动建表、表结构变更同步与数据实时同步能力。...
  8. 流计算 Oceanus 消息队列 Upsert Kafka Upsert Kafka 连接器支持以 upsert 方式从 Kafka topic 中读取数据并将数据写入 Kafka topic。作为 Source，Upsert Kafka 连接器生产 changelog 流，其中每条数据记录代表一个更新或删除事件。更准确地说，数据记录中的 value 被解释为同一 key 的最后一个 value 的 UPDATE，如果有这个 key（如果不存在相应的 key，则该更新被视为 。...
赞 (0)
上一篇 4天前
下一篇 4天前