流计算 Oceanus 数据仓库 Doris

二、版本说明

三、使用范围

Flink Connector Doris 目前仅支持 Doris sink。支持的 Doris 版本为0.14.0及以上版本，并且要求开启配置 enable_http_server_v2 = true。

四、DDL 定义

注意：Flink 1.13 与其他 Flink 版本 的 DDL 参数不同，请选择对应的版本使用。

五、作为数据目的地 Sink（Flink 1.13）

CREATE TABLE doris_sink_table ( id INT, name VARCHAR) WITH ( 'connector' = 'doris', — 固定值 'doris' 'fenodes' = 'FE_IP:FE_HTTP_PORT', — Doris FE HTTP 地址 'table.identifier' = 'test.sales_order', — Doris 表名 格式：db.tbl 'username' = 'root', — 访问Doris的用户名，拥有库的写权限 'password' = 'password', — 访问Doris的密码 'sink.batch.size' = '500', — 单次写BE的最大行数 'sink.batch.interval' = '1s' — flush 间隔时间，超过该时间后异步线程将 缓存中数据写入BE。 默认值为1秒，支持时间单位ms、s、min、h和d。设置为0表示关闭定期写入。);

六、作为数据目的地 Sink

CREATE TABLE doris_sink_table ( id INT, name VARCHAR) WITH ( 'connector' = 'doris', — 固定值 'doris' 'fenodes' = 'FE_IP:FE_HTTP_PORT', — Doris FE HTTP 地址 'table.identifier' = 'test.sales_order', — Doris 表名 格式：db.tbl 'username' = 'root', — 访问Doris的用户名，拥有库的写权限 'password' = 'password' — 访问Doris的密码);

七、作为 Catalog

CREATE CATALOG doris_catalog WITH ( 'type' = 'doris', 'fenodes' = 'FE_IP:FE_HTTP_PORT', — Doris FE HTTP 地址 'username' = 'root', — 访问Doris的用户名，拥有库的写权限 'password' = 'password', — 访问Doris的密码 'default-database' = 'default');

八、作为维度表

社区本身支持 Doris connector 作为维表查询，但仍然推荐使用 JDBC connector，更加稳定和通用：

CREATE TABLE `doris` ( `user_id` bigint, `item_id` bigint, `behavior` STRING, PRIMARY KEY (`user_id`) NOT ENFORCED) WITH ( 'connector' = 'jdbc', 'url' = 'jdbc:mysql://172.28.28.98:9030/CDB?rewriteBatchedStatements=true&serverTimezone=Asia/Shanghai', — query_port, FE mysql server port 'table-name' = 'my-table', — 需要写入的数据表 'username' = 'admin', — 数据库访问的用户名（需要提供 INSERT 权限） 'password' = 'MyPa$$w0rd', — 数据库访问的密码 'lookup.cache.max-rows' = '100', — 读缓存大小 'lookup.cache.ttl' = '5000' — 读缓存的 TTL);

十、Sink

十一、Catalog

十三、Sink

更多参数说明请参见 Flink Doris Connector。

十四、类型映射

十六、Flink 1.13

—– Flink 1.13 —–CREATE TABLE datagen_source_table ( id INT, name STRING) WITH ( 'connector' = 'datagen', 'rows-per-second'='1' — 每秒产生的数据条数);CREATE TABLE doris_sink_table ( id INT, name STRING) WITH ( 'connector' = 'doris', — 固定值 'doris' 'fenodes' = 'FE_IP:FE_RESFUL_PORT', — Doris FE HTTP 地址 'table.identifier' = 'test.sales_order', — Doris 表名 格式：db.tbl 'username' = 'root', — 访问Doris的用户名，拥有库的写权限 'password' = 'password', — 访问Doris的密码 'sink.batch.size' = '500', — 单次写BE的最大行数 'sink.batch.interval' = '1s' — flush 间隔时间，超过该时间后异步线程将 缓存中数据写入BE。 默认值为1秒，支持时间单位ms、s、min、h和d。设置为0表示关闭定期写入。);INSERT INTO doris_sink_table select * from datagen_source_table;

十七、Flink 1.16+

CREATE TABLE datagen_source_table ( id INT, name STRING) WITH ( 'connector' = 'datagen', 'rows-per-second'='1' — 每秒产生的数据条数);CREATE TABLE doris_sink_table ( id INT, name STRING) WITH ( 'connector' = 'doris', 'fenodes' = 'FE_IP:HTTP_PORT', 'table.identifier' = 'db.table', 'username' = 'root', 'password' = 'password', 'sink.label-prefix' = 'doris_label', 'sink.label-prefix.uuid.enabled' = 'true');INSERT INTO doris_sink_table select * from datagen_source_table;

CREATE CATALOG doris_catalog WITH ( 'fenodes' = 'FE_IP:FE_RESFUL_PORT', — Doris FE HTTP 地址 'username' = 'root', — 访问Doris的用户名，拥有库的写权限 'password' = 'password', — 访问Doris的密码 'default-database' = 'default');CREATE TABLE datagen_source_table ( id INT, name STRING) WITH ( 'connector' = 'datagen', 'rows-per-second'='1' — 每秒产生的数据条数);INSERT INTO `doris_catalog`.`my_database`.`my_table` SELECT * FROM.datagen_source_table;

MySQL-CDC 对接 Doris 代码示例（Flink-1.16+)：

–mysql cdc 源表CREATE TABLE `mysql_cdc_source_table` ( `id` INT NOT NULL, `name` VARCHAR, PRIMARY KEY (`id`) NOT ENFORCED) WITH ( 'connector' = 'mysql-cdc', — 固定值 'mysql-cdc' 'hostname' = 'YourHostName', — 数据库的 IP 'port' = '3306', — 数据库的访问端口 'username' = 'YourUserName', — 数据库访问的用户名（需要提供 SHOW DATABASES、REPLICATION SLAVE、REPLICATION CLIENT、SELECT 和 RELOAD 权限） 'password' = 'YourPassword', — 数据库访问的密码 'database-name' = 'YourDatabase', — 需要同步的数据库 'table-name' = 'YourTable' — 需要同步的数据表名 );–写入doris表CREATE TABLE `print_table` ( `id` INT, `name` STRING, PRIMARY KEY (`id`) NOT ENFORCED) WITH ( 'connector' = 'doris', — 固定值 'doris' 'fenodes' = 'FE_IP:FE_RESFUL_PORT', — Doris FE HTTP 地址 'table.identifier' = 'dbName.tableName', — Doris 表名 格式：db.tbl 'username' = 'YourUserName', — 访问Doris的用户名，拥有库的写权限 'password' = 'YourPassword', — 访问Doris的密码 'sink.label-prefix' = 'doris_label', 'sink.label-prefix.uuid.enabled' = 'true');insert into print_tableselect id,name from mysql_cdc_source_table;

十八、实践教程（Flink 写入 Doris）

1. 推荐使用 Flink 1.16+ 的 Doris Connector。

新版本对两阶段提交、Checkpoint 协议适配更完善，稳定性与新特性支持都更好，建议新接入项目直接使用 Flink 1.16 及以上版本。

2. sink.enable-2pc 参数说明。

sink.enable-2pc 控制 Stream Load 是否走两阶段提交，默认开启。开启后，Flink Checkpoint 完成时才提交事务，配合 Checkpoint 可实现 端到端 Exactly-Once。在一些业务场景下可以显式关闭2PC，有效规避 Transaction not found、Too many transactions、事务超时等常见问题。

建议关闭（'sink.enable-2pc' = 'false'）的场景：

·写入 Unique Key 模型表：主键天然幂等，无需 2PC，可避免事务挂起。

·b. 业务可接受 At-Least-Once 语义（如日志、指标采集）。

·c. 未开启 Flink Checkpoint 的作业（必须关闭，否则事务永远不会提交，导致堆积超时）。

3. 建议同时开启 sink.label-prefix.uuid.enabled = true。

Stream Load 使用 label 做幂等控制，作业重启或多个作业共用 prefix 时容易出现 Label has already been used 错误。

开启该参数后，连接器会在 label 后自动拼接随机 UUID，从根本上避免 label 冲突。如果开启2PC，强烈建议也一起配置该参数。

4. 数据格式：CSV（默认） vs JSON

·默认 CSV：性能高、传输体积小，但字段中若包含分隔符 \\x01 / \\x02 等特殊字符，会导致解析失败。

·可选 JSON：稳定性更好，几乎无解析冲突风险，推荐用于字段内容不可控、含特殊字符的业务场景。

除非追求极致吞吐，否则推荐增加下面配置换取稳定性：

'sink.properties.format' = 'json','sink.properties.read_json_by_line' = 'true'

20、Upsert

若需要 Upsert ，则要求 Doris 表必须是 Unique 模型或者 Aggregate 模型。建表示例如下：

— Uniqe 模型建表语句CREATE TABLE `doris_sink_table` ( `id` int(11), `name` varchar(1024)) UNIQUE KEY(`id`) DISTRIBUTED BY HASH(`id`) BUCKETS 10PROPERTIES("replication_num" = "3");– Aggregate 模型建表语句CREATE TABLE `doris_sink_table` ( `id` int(11), `name` varchar(1024) REPLACE DEFAULT '0') AGGREGATE KEY(`id`)DISTRIBUTED BY HASH(`id`) BUCKETS 10PROPERTIES("replication_num" = "3"); — 注意若 BE 节点不够，会报 `Failed to find enough host in all backends` 错误，可适当减少该值。

21、用户权限

用户必须拥有对应的库的写权限。

CREATE USER 'test' IDENTIFIED BY 'test_passwd';GRANT ALL ON test TO test;

22、作业无法停止

作业停止时默认勾选触发快照。如果作业持续有写入报错，特别是在两阶段提交开启的情况下，作业触发快照失败会导致无法停止。此时可以在停止作业时取消勾选触发快照。后续作业选择最新的快照启动即可。

23、写入失败

报错：“too many filtered rows xxx, "ErrorURL":"或 Insert has filtered data in strict mode, tracking url=xxxx.”？

原因：表的 schema、分区等与导入的数据不匹配。

可在 Doris 客户端执行命令查看具体原因：show load warnings on `<url>`。其中<url> 即为报错信息中返回的 error url。

24、事务提交失败

报错："msg": "TStatus: errCode = 2, detailMessage = transaction [XXX] not found"

原因：Doris 事务太久没提交（可能是作业停止过久、或者作业由于写入失败导致 failover 时间过长），事务在 Doris 这边被删除了，无法从快照恢复。

解决措施：需要丢状态启动或者指定时间位点启动。Doris 侧可以增加事务保存个数。

25、CSV 写入格式

当写入格式为 CSV 时，默认行列分隔符如下：

26、更多常见问题

请参见 Doris Flink Connector 页面文档。