Data ingestion system

The source notes for this track are currently maintained in Chinese.

预计阅读时间: 50 分钟 | 前置阅读: doc-02 (写入主线), doc-07 (Compaction) 下一次阅读: doc-12 (BE副本), doc-11 (FE HA)

1. 导入方式全景对比

源码导航

2. 统一事务模型

TxnManager——所有导入方式共享

所有导入方式
    └── GlobalTransactionMgr  (fe/fe-core/.../transaction/)
        ├── beginTransaction(dbId, tableId, label)
        │   → 创建 TxnState {label, status=PREPARE, ...}
        │   → Label 去重检查 (再次提交相同 label → 返回已有 txnId)
        │
        ├── commitTransaction(txnId, tabletCommitInfos)
        │   → 检查所有 BE 写入完成
        │   → 标记 Txn status=COMMITTED
        │   → 触发 PublishVersionDaemon
        │
        └── abortTransaction(txnId)
            → 清理已写数据, 释放资源

Label 去重——Exactly-Once 的基石

Label = "my_load_20260708_001"

第一次提交:  beginTxn("my_load_20260708_001")
  → txnId=1001, status=PREPARE → 写入数据 → commit → status=VISIBLE

第二次提交 (重试): beginTxn("my_load_20260708_001")
  → Label 已存在, 检查上一次的状态:
    - status=VISIBLE → 返回 "Duplicate label, already committed"
    - status=PREPARE → 返回 txnId=1001 (继续使用同一个事务)
    - status=ABORTED → 创建新 txnId

3. Group Commit——高频小批量写入

原理

多个客户端的 INSERT 请求在 BE 的 MemTable 层面合并, 然后一起 Flush, 大幅减少 Compaction 压力:

Client A: INSERT INTO t VALUES (1, 'a')      → MemTable
Client B: INSERT INTO t VALUES (2, 'b')      → MemTable
Client C: INSERT INTO t VALUES (3, 'c')      → MemTable
                    │
            ┌───────▼───────┐
            │  Group Commit  │  ← 等待 10ms 或达到 N 行
            │  一次性 Flush   │
            └───────┬───────┘
                    ▼
           1 个 Rowset (含3行)

参数

4. Broker Load——大批量外部文件导入

流程

FE: SUBMIT LOAD LABEL ... DATA INFILE "hdfs://path/*" INTO TABLE t
  → BrokerLoadJob
    → 1. 列出 HDFS 文件列表 (NameNode Client)
    → 2. 为每个文件范围划分 Task
    → 3. 分配 Task 到 BE
    → 4. BE 从 HDFS 拉取数据 → 正常写入 (DeltaWriter)
    → 5. 所有 Task 完成 → commit + Publish

关键: 数据从 HDFS → BE 是直接流式读取, 不经过 FE 中转。FE 只协调 Task 元数据。

5. Routine Load + Kafka——Exactly-Once 实现

Kafka Consumer 模型

Kafka Topic [MyTopic]
├── Partition 0 → RoutineLoadTask[0]  (1个 BE 上的 Consumer)
├── Partition 1 → RoutineLoadTask[1]
└── Partition 2 → RoutineLoadTask[2]

每个 Task:
  while (true):
    1. KafkaConsumer.poll() → 一批消息
    2. Label = "jobName_partition_offset" (唯一)
    3. FE: beginTxn(label) → txnId
    4. BE: DeltaWriter.write(messages) → MemTable
    5. FE: commitTxn(txnId)
    6. KafkaConsumer.commitSync(offset)  ← 现在才提交 offset

故障恢复: Task 崩溃 → Kafka offset 未提交 → 重启后从上次 committed offset 开始 → Label 相同 → Doris 去重保证幂等 → Exactly-Once

6. 性能调优参数

7. 常见问题 / 面试题

Q1: Exactly-Once 的边界在哪? Label 有效期多久? A: Label 有效期由 label_keep_max_num + label_keep_max_second 控制(默认 1000 个/3天)。超出范围的旧 Label 被清理后, 重试就会产生重复数据 → 退化到 At-Least-Once。

Q2: Group Commit vs Stream Load 选哪个? A: Group Commit 适合高频小批量 (如每秒数百个 INSERT), 自动合并, 无需客户端改造。Stream Load 适合周期性大文件导入 (如每小时一次 CSV import)。Group Commit 本质上是 Stream Load 的"微批优化"——内部也转换为 Stream Load。

Q3: Routine Load 的 Partition 数与 Doris Tablet 数如何匹配? A: 建议 Kafka Partition 数 = Doris Bucket 数 (或倍数)。Kafka Partition 太少 → 写入并行度不足; Kafka Partition 太多 → 单 Tablet 写入频繁, MemTable 过于碎片化, 导致 Compaction 压力大。

下一步

• FE HA: doc-11-fe-ha-replication.md——导入的事务信息如何在 FE 间同步