Doris Routine Load 的 exactly-once 语义：label 机制到底怎么用？

一个生产事故

Kafka 里的订单事件正在被 Doris Routine Load 消费，跑了 3 个月没问题。一天运维重启 Kafka 集群，Routine Load 卡住了。

重启后查状态：

• SHOW ALL ROUTINE LOAD\G 显示 Job 状态是 RUNNING
• 但 data_quality 表里显示过去 10 分钟 0 条数据进入
• 业务方已经反馈”订单数对不上”

排查发现：Kafka 集群重启时，Routine Load 的 offset 提交丢了。重启后 Doris 重新消费了一批 Kafka 消息，但有些消息已经被前一次消费过——SELECT COUNT(*) FROM orders 比真实多 200 条。

这就是”至少一次（at-least-once）”的代价。Doris Routine Load 的官方文档写”支持 exactly-once”，但你必须正确使用 label 机制。

“exactly-once” 这个词在 Doris 里到底是什么意思？

Doris 官方文档里写：

Routine Load 支持 exactly-once 语义，依赖 Kafka offset 的提交机制和 Doris 的 label 去重机制。

实际语义是：

在 Doris 侧，数据不会重复。通过 label 唯一标识一次导入任务，即使 Routine Load 重启、重试、offset 回退，Doris 不会写入重复数据。

注意：**”数据不会重复”是有条件的**。如果你没用对 label，或者 Kafka 侧产生了真正的新消息（不是回放），Doris 还是会写入。

label 机制的原理

每次 CREATE ROUTINE LOAD 时你必须指定一个 label，或者让 Doris 自动生成：

CREATE ROUTINE LOAD load_orders_kafka ON orders
COLUMNS (order_id, user_id, amount, status, created_at)
PROPERTIES (
    "desired_concurrent_number" = "3",
    "max_error_number" = "1000"
)
FROM KAFKA (
    "kafka_broker_list" = "kafka-1:9092,kafka-2:9092",
    "kafka_topic" = "order_events",
    "property.kafka_default_offsets" = "OFFSET_BEGINNING"
);

label 就是 load_orders_kafka 这个 job name。Doris 内部维护了一个 __internal_schema.column_statistics 类似的小表，记录每个 label 已经处理过的 Kafka offset 范围。

当 Routine Job 重启或重试时：

1. Doris 从 Kafka 读消息
2. 在写入前，检查这个 label 的 offset 范围是否已经处理过
3. 如果 offset ≤ 已处理的最大 offset，直接跳过（去重）
4. 如果 offset > 已处理的最大 offset，正常写入并更新

这就是”exactly-once”的核心：Doris 自己维护”哪些 offset 已经写过”的状态，重启后能跳过。

label 的命名规则

很多人不知道，label 必须是全局唯一的、长度 ≤ 128 字符的字符串。

我自己在生产里见过的几个坑：

坑 1：label 写死导致 Job 重建失败

-- 第一次创建 Job
CREATE ROUTINE LOAD load_orders_kafka ON orders ...;

-- Job 出问题，删了重建
DROP ROUTINE LOAD load_orders_kafka;

-- 重新创建
CREATE ROUTINE LOAD load_orders_kafka ON orders ...;
-- ERROR 5073 (HY000): label load_orders_kafka already used

原因：Doris 在内部保留了 label 的去重记录，删除 Routine Load Job 不会清除这些记录。Doris 的设计哲学是：数据导入是不可变的，label 一旦用过就永久占用。

解法：

• 重建时用新的 label（带时间戳或版本号）
• 或用 ALTER ROUTINE LOAD 修原 Job，不要 DROP + CREATE

坑 2：把同一个 Kafka topic 用不同 label 重复消费

-- Job 1：写 orders 表
CREATE ROUTINE LOAD load_orders_v1 ON orders ...;

-- Job 2：写 orders_archive 表
CREATE ROUTINE LOAD load_orders_v2 ON orders_archive ...;
-- 期望：写两份数据
-- 实际：两个 Job 独立维护 label，互不冲突。数据会写两份。

这是正确的行为。但如果你误以为”label 决定数据去重”，可能会困惑。

坑 3：手动恢复时跳过 label 检查

-- 强制重置 offset 到某个位置
ALTER ROUTINE LOAD load_orders_kafka
FROM KAFKA ("kafka_offsets" = "OFFSET_BEGINNING");

这个命令会让 Routine Load 重新消费所有消息。如果此时 label 记录还存在，Doris 会根据 label 记录去重——已经处理过的 offset 段被跳过。

但如果你同时改了 label（比如 DROP + CREATE），新 label 没有任何历史记录，重放的数据会全部重新写入。

生产环境的 label 管理建议

1. label 命名带上时间戳和环境

CREATE ROUTINE LOAD prod_orders_2026q3_v1 ON orders ...

好处：

• 区分 dev / staging / prod
• 同名 Job 重建时不会冲突
• 排查问题时一眼能看出 Job 是什么时候建的

2. 监控 data_quality 表的 progress

SELECT * FROM information_schema.loads 
WHERE label = 'prod_orders_2026q3_v1' 
ORDER BY create_time DESC LIMIT 10;

重点关注：

• progress：Kafka offset 进度，停了就是有积压
• num_filtered_rows：过滤掉的数量
• error_rows：错误行数
• status：PENDING / LOADING / FINISHED / FAILED

3. 设置 max_error_number 而不是默认的 0

PROPERTIES (
    "max_error_number" = "1000",
    "max_batch_interval" = "20",
    "max_batch_rows" = "500000",
    "max_batch_size" = "209715200"
)

max_error_number=0 意味着任何一条脏数据都会让整个 Job 卡住。生产环境必须留出容错空间（1000-10000 视业务而定）。

4. 配套的 Kafka 侧配置

Doris 端的 exactly-once 依赖 Kafka 侧不丢消息：

• acks=all：生产者等所有副本确认
• min.insync.replicas=2：最少 2 个副本确认
• 副本数 ≥ 3：单点故障不丢
• 关闭 auto.create.topics.enable

如果 Kafka 侧配置错了，Doris 端 label 做得再好也救不了。

exactly-once 的真实边界

必须承认：Doris 的 exactly-once 也不是绝对 exactly-once。

几种会破坏 exactly-once 的情况：

情况 1：Kafka 侧消息本身有重复

如果上游生产者用 at-least-once 模式发送（比如 try-catch 重试），同一条业务事件可能在 Kafka 里有两条。Doris 会照单全收写入。

对策：业务侧用唯一 ID 去重（在 Doris 写入前去重或用 UNIQUE KEY 模型）。

情况 2：Doris 表引擎是 Duplicate Key 而非 Unique Key

CREATE TABLE orders (
    order_id BIGINT,
    user_id BIGINT,
    amount DECIMAL(10, 2),
    created_at DATETIME
) DUPLICATE KEY(order_id)  -- 重复行保留
DISTRIBUTED BY HASH(order_id) BUCKETS 10;

Duplicate 模型下，同一个 order_id 会出现多行。要实现去重必须用 UNIQUE KEY 模型。

CREATE TABLE orders (
    order_id BIGINT,
    user_id BIGINT,
    ...
) UNIQUE KEY(order_id)
DISTRIBUTED BY HASH(order_id) BUCKETS 10;

UNIQUE KEY 模型的语义是”同 order_id 的写入按主键替换”，天然支持去重。

情况 3：手动删除数据后重新导入

DELETE FROM orders WHERE order_id = 12345;
-- 然后重新消费 Kafka 中 order_id=12345 的消息
-- UNIQUE KEY 模型下会重新写入
-- 看起来"去重"被破坏了

这种情况需要业务上接受”删除 + 重导”的语义。

监控和告警

生产环境必须监控：

-- 1. 看所有 Routine Load 状态
SHOW ALL ROUTINE LOAD\G

-- 2. 找 RUNNING 但实际卡住的 Job
SELECT NAME, 
       TIMESTAMPDIFF(SECOND, CREATE_TIME, NOW()) AS alive_seconds,
       -- progress 是 JSON 格式
       -- 解析 progress.kafka_offsets 看是否变化
       ...
FROM information_schema.routine_load_jobs
WHERE STATE = 'RUNNING';

告警策略：

• 5 分钟内 progress 不变 → 告警
• error_rows / total_rows 超过 5% → 告警
• alive_seconds > 3600 但 total_rows 极少 → 告警

意味着什么？

Doris Routine Load 的 exactly-once 是工程上的精确度问题，不是”配置 ON/OFF”那么简单：

• label 设计要可追溯（带环境、时间戳、版本号）
• 配合 UNIQUE KEY 模型才能真正去重
• Kafka 侧配置要同步（acks、min.insync.replicas）
• 监控和告警要齐（progress、error rate、alive time）

真正稳的 exactly-once 链路 = Doris 端 label + UNIQUE KEY + Kafka 端 acks=all + 上游业务唯一 ID。少了任何一环都会出问题。

数据密集型 AI 后端的工程师，对 exactly-once 的理解深度直接决定了你能否独立负责一条数据管道。这是基本功。

关联阅读

• Doris 的数据从哪来：Kafka/RocketMQ 导入的几种姿势与取舍
• 从 Canal 到 Doris：整条数据同步链路的故障恢复
• MySQL 与 Doris 的真实边界：OLTP/OLAP 的分工逻辑