Apache Druid 数据摄入（Ingestion）完全指南：流式与批量摄入方法详解

本篇指南围绕 Apache Druid 的数据摄入（Ingestion，也称 Indexing）展开：数据从源系统读入后如何被组织为 segment 文件、由哪些进程负责执行、最终如何落入深度存储（deep storage）并供查询使用。读完本文，你将掌握 Druid 的流式摄入（Kafka / Kinesis）与批量摄入（Native batch / SQL / Hadoop）全貌，能够依据方法对比表选择适合自身场景的摄入方案，并快速定位摄入规范（ingestion spec）、Schema 模型、Rollup 与分区等核心概念的详细文档。

数据摄入的基本概念：从源系统到 Segment

在 Druid 中，向系统加载数据被称为摄入（ingestion）或索引（indexing）。摄入过程中，Druid 从你的源系统读取数据，并将其存储在称为 segment 的数据文件中。一般而言，每个 segment 文件包含约几百万行数据。

除了 Hadoop 批摄入（在 YARN 上以 MapReduce 作业运行）外，绝大多数摄入方法都由以下两类进程实际加载源数据：

• MiddleManager 进程
• Indexer 进程

摄入完成后，Druid 会创建 segment 并存入 深度存储（deep storage），随后由 Historical 节点将 segment 加载进内存以响应查询。对于流式摄入，MiddleManager 和 Indexer 可以直接用到达的数据实时响应查询。更完整的存储链路可参考 Storage overview。

摄入前必读：四个通用核心概念

无论选择哪种摄入方法，以下四个概念适用于所有摄入方式，它们分别对应 docs/ingestion 目录下的四份主题文档：

概念	解决的问题	对应文档
Schema 模型	datasource、主时间戳、维度、指标如何定义	Druid schema model
Rollup	摄入期预聚合的原理与收益最大化建议	Data rollup
Partitioning	时间分块与二级分区机制	Partitioning
Ingestion spec	摄入规范中各配置项的完整参考	Ingestion spec reference

Schema 模型：datasource、主时间戳、维度与指标

Druid 将数据存储在 datasource 中，它类似于传统关系型数据库中的表。Druid 的数据模型同时具备关系型与时间序列数据模型的特征：

• 主时间戳（Primary timestamp）：Druid 的 Schema 必须包含主时间戳。它用于数据的分区与排序，帮助 Druid 在查询时快速定位时间范围内的数据，也用于基于时间的数据管理操作（如删除时间分块、覆盖时间分块、基于时间的保留规则）。主时间戳在摄入时由 timestampSpec 解析，无论源字段名是什么，最终总是存储在名为 __time 的列中。其余基于主时间戳的操作由 granularitySpec 控制；若存在多个时间列，其余可存为次要时间戳。
• 维度（Dimensions）：按原样存储的列，可在查询时用于分组、过滤或作为聚合器输入。禁用 rollup 时，维度集合的行为与不支持 rollup 的普通数据库一致。摄入时通过 dimensionsSpec 配置。
• 指标（Metrics）：以聚合形式存储的列，在启用 rollup 时最为有用。摄入时通过 metricsSpec 配置。rollup 会将具有相同时间戳与维度值的多行合并为一行、同时保留指标的汇总信息；此外，部分近似聚合器若在摄入期先做部分计算，查询时能更快出结果。

Ingestion spec：摄入规范三组件

除 SQL 摄入外，其余摄入方式都通过 ingestion spec（摄入规范）来配置。一个摄入任务（task）由以下三大组件构成：

• dataSchema：配置 datasource 名称、主时间戳、维度、指标，以及可选的转换与过滤；
• ioConfig：告诉 Druid 如何连接源系统以及如何解析数据；
• tuningConfig：控制各摄入方法特有的调优参数。

以 index_parallel（Native batch）任务为例的完整摄入规范：

{
  "type": "index_parallel",
  "spec": {
    "dataSchema": {
      "dataSource": "wikipedia",
      "timestampSpec": {
        "column": "timestamp",
        "format": "auto"
      },
      "dimensionsSpec": {
        "dimensions": [
          "page",
          "language",
          { "type": "long", "name": "userId" }
        ]
      },
      "metricsSpec": [
        { "type": "count", "name": "count" },
        { "type": "doubleSum", "name": "bytes_added_sum", "fieldName": "bytes_added" },
        { "type": "doubleSum", "name": "bytes_deleted_sum", "fieldName": "bytes_deleted" }
      ],
      "granularitySpec": {
        "segmentGranularity": "day",
        "queryGranularity": "none",
        "intervals": ["2013-08-31/2013-09-01"]
      }
    },
    "ioConfig": {
      "type": "index_parallel",
      "inputSource": {
        "type": "local",
        "baseDir": "examples/indexing/",
        "filter": "wikipedia_data.json"
      },
      "inputFormat": {
        "type": "json",
        "flattenSpec": {
          "useFieldDiscovery": true,
          "fields": [
            { "type": "path", "name": "userId", "expr": "$.user.id" }
          ]
        }
      }
    },
    "tuningConfig": {
      "type": "index_parallel"
    }
  }
}

注意摄入规范各组件支持的具体选项取决于所选的摄入方法](#摄入方法全景对比)。此外，也可以完全不手写 ingestion spec，而是使用 Druid [Web 控制台 中的 “Load data” 可视化功能加载数据，它支持 Kafka、Kinesis 与 native batch 三种模式。

关于各组件更细粒度的字段（如 timestampSpec 的 format 支持 iso/posix/millis/micro/nano/auto 及任意 Joda DateTimeFormat 字符串；dimensionsSpec 的 schema 自动发现、维度对象与包含/排除规则；granularitySpec 的 segmentGranularity、queryGranularity、rollup、intervals；tuningConfig 的 maxRowsInMemory、maxBytesInMemory、indexSpec 及 front coding 字典编码等），可直接查阅 Ingestion spec reference。

摄入方法全景对比

Druid 提供两种流式摄入和三种批量摄入。流式摄入由持续运行的 supervisor 控制；批量摄入作业则与一个贯穿作业全程的 controller task 相关联。

流式摄入（Streaming）：Kafka 与 Kinesis

流式摄入的两个选项均由持续运行的 supervisor 驱动，用于管理随时间不断执行的任务集合。两者都支持摄入迟到数据（late data），并都提供精确一次（exactly-once）语义保证：

方法	Kafka	Kinesis
Supervisor 类型	kafka	kinesis
工作原理	Druid 直接从 Apache Kafka 读取数据。	Druid 直接从 Amazon Kinesis 读取数据。
能否摄入迟到数据？	是。	是。
精确一次保证？	是。	是。

从源码看，Kafka 摄入的核心实现是 KafkaIndexTask，它继承自 SeekableStreamIndexTask，负责对 Kafka 的分区（topic-partition）进行定位与消费；其对应的 supervisor 持续创建并监管这类任务。Kinesis 摄入机制与 Kafka 类似，只是数据源换成了 Kinesis 流（stream 的分片 shard 充当分区角色）。

值得说明的是，流式摄入通常采用尽力而为的 rollup（best-effort rollup），原因在于：流式摄入以增量发布方式推进——在一个时间分块的全部数据到达之前就可能完成并发布 segment，且并行化过程缺少 perfect rollup 所需的 shuffle 步骤，因此本可合并的记录可能落入不同 segment。

批量摄入（Batch）：Native batch、SQL 与 Hadoop

批量摄入有三个选项，各有其适用场景：

方法	Native batch	SQL（MSQ）	Hadoop-based
Controller 任务类型	index_parallel	query_controller	index_hadoop
提交方式	将 index_parallel spec 发送到 Tasks API。	向 SQL 任务 API 发送 INSERT 或 REPLACE 语句。	将 index_hadoop spec 发送到 Tasks API。
并行度	当 maxNumConcurrentSubTasks 大于 1 时，通过子任务并行。	使用 query_worker 子任务。	使用 YARN。
容错性	工作节点（worker）失败后自动重启；controller 任务失败则作业失败。	controller 或 worker 任务失败都导致作业失败。	YARN 容器失败后自动重启；controller 任务失败则作业失败。
能否追加？	是。	是（INSERT）。	否。
能否覆盖？	是。	是（REPLACE）。	是。
外部依赖	无。	无。	Hadoop 集群。
输入源	任意 inputSource。	任意 inputSource（通过 EXTERN）或 Druid datasource（通过 FROM）。	任意 Hadoop 文件系统或 Druid datasource。
输入格式	任意 inputFormat。	任意 inputFormat。	任意 Hadoop InputFormat。
二级分区选项	动态、哈希、范围分区均可用，详见 partitionsSpec。	范围分区（CLUSTERED BY）。	通过 partitionsSpec 使用哈希或范围分区。
Rollup 模式	当 tuningConfig 中 forceGuaranteedRollup = true 时为完美（perfect），否则为尽力而为（best-effort）。	始终完美。	始终完美。

Native batch（index_parallel）深入

index_parallel 是面向生产环境的多线程批量摄入任务类型，只依赖 Druid 自身资源，无需 Hadoop 等外部系统。其核心实现是 ParallelIndexSupervisorTask（继承自 AbstractBatchIndexTask），运行机制如下：

1. Supervisor 任务作为编排者，将输入数据切分成多个分片（split），为每个分片创建 worker 任务；
2. worker 任务提交给 Overlord，由 Overlord 调度到 MiddleManager 或 Indexer 上运行；
3. worker 成功处理完分片后，将生成的 segment 列表上报给 Supervisor；
4. Supervisor 周期性检查 worker 状态，失败的任务会重试直至达到配置上限；全部成功后在一次发布中完成 segment 发布并结束摄入。

并行摄入需要两个前提：ioConfig 中使用可拆分（splittable）的 inputSource（如 S3、本地文件、Druid datasource 等），以及 tuningConfig 中 maxNumConcurrentSubTasks 大于 1（否则任务退化为顺序模式，由 Supervisor 逐个读取文件并自行建 segment）。此外，Native batch 还支持通过 ioConfig 的 appendToExisting 追加数据，以及实验性的 dropExisting（用 tombstone 完全替换现有 segment）；输入文件支持 bz2、gz、xz、zip、sz（Snappy）、zst（ZSTD）等压缩格式。

index_parallel 的 partitionsSpec 支持四种分区策略，可从摄入速度、分区方式、适用 rollup 模式与查询期分区裁剪能力四个维度对比：

PartitionsSpec	摄入速度	分区方法	支持的 rollup 模式	查询时二级分区裁剪
dynamic	最快	基于 segment 行数的动态分区	尽力而为 rollup	不适用
hashed	中等	多维度哈希分区	完美 rollup	可基于全部 partitionDimensions 做 segment 裁剪
single_dim	较慢	单维度范围分区	完美 rollup	可基于 partitionDimension 做 segment 裁剪
range	最慢	多维度范围分区	完美 rollup	可基于第一个 partitionDimensions 做 segment 裁剪

范围分区（single_dim / range）的两个核心收益是：将相似数据合并进同一 segment 从而降低存储占用、提升压缩率；以及 Broker 层 segment 裁剪——当查询的 WHERE 条件命中分区维度时，可直接排除不可能包含匹配行的 segment。裁剪生效要求查询按分区维度从左到右逐级参与过滤，且仅支持字符串等值（=、IN）与字符串比较（<、>、<=、>=）等可裁剪的过滤条件，LIKE 类过滤不支持裁剪。

SQL-based ingestion（MSQ）

基于 druid-multi-stage-query 扩展的 SQL 摄入以 query_controller 为 controller 任务、query_worker 为子任务，通过 SQL 语句提交：

• INSERT：向现有或新建 datasource 追加数据；
• REPLACE：覆盖指定时间范围内的数据，配合 PARTITIONED BY 与 CLUSTERED BY 控制时间分区与范围分区。

SQL 摄入始终提供完美 rollup（perfect rollup），且不需要外部集群依赖。

Hadoop-based ingestion

index_hadoop 任务在 YARN 上以 MapReduce 作业运行，是唯一依赖外部 Hadoop 集群的摄入方式，适合超大规模数据集。它支持任何 Hadoop InputFormat 输入、可从 Hadoop 文件系统或 Druid datasource 读取，不能追加、只能覆盖。二级分区通过 partitionsSpec 使用哈希或范围分区，rollup 始终完美。由于历史原因，Hadoop 摄入仍使用 legacy dataSchema 中的 parser 而非 inputFormat。

如何选择摄入方法

选择摄入方法时，可围绕以下决策维度评估：

1. 数据到达方式：持续产生的事件流（如埋点、日志）优先选择 Kafka / Kinesis 流式摄入；存量文件或周期性全量/增量数据选择批量摄入。
2. 延迟与实时性需求：流式摄入中 MiddleManager / Indexer 可直接用到达数据实时响应查询；批量摄入则在任务完成后 segment 才对查询可见。
3. 数据一致性语义：Kafka 与 Kinesis 提供精确一次保证；批量摄入中 SQL 与 Hadoop 的 rollup 始终完美，而 Native batch 需在 tuningConfig 中设置 forceGuaranteedRollup: true 才能获得完美 rollup。
4. 追加与覆盖能力：需要追加选 Native batch（appendToExisting）或 SQL INSERT；需要覆盖选 Native batch 默认行为、SQL REPLACE 或 Hadoop。
5. 外部依赖：Hadoop 摄入需要可用的 Hadoop 集群（YARN）；其余方法仅依赖 Druid 自身资源。
6. 输入源与格式：需要读取 S3 / GCS / Azure / HDFS / 本地 / HTTP / Druid datasource / RDBMS（SQL）等，可查阅 Input sources 与 Source input formats。
7. 分区与查询裁剪需求：对查询性能敏感时，优先选择支持查询期分区裁剪的 hashed / single_dim / range 分区，并结合分区维度选择建议 将高频过滤维度前置到 dimensions 列表首位。

需要说明的还有 rollup 的选择逻辑：需要最优性能或存储空间受限、且不需要高基数维度的原始值时，建议开启 rollup；需要单行明细查询或需要对任意列执行 GROUP BY / WHERE 时，应关闭 rollup。若两种需求并存，可为不同用途建立多张 rollup 配置不同的表。想要量化 rollup 收益，可运行 SELECT SUM("num_rows") / (COUNT(*) * 1.0) FROM datasource 比较 Druid 内行数与摄入事件数。

快速上手：三个入门入口

1. 可视化加载（推荐新手）：在 Web 控制台 中使用 “Load data” 向导，免写 spec，支持 Kafka、Kinesis、Native batch。
2. JSON 批量任务：参照上文 index_parallel 示例，POST 到 Overlord 的 Tasks API 端点 /druid/indexer/v1/task，或使用 Druid 自带的 bin/post-index-task 脚本提交。批量摄入的入门可跟随 Tutorial: Loading a file。
3. SQL 任务：向 SQL 任务 API 提交 INSERT / REPLACE 语句，通过 PARTITIONED BY 与 CLUSTERED BY 控制分区，具体可参考 SQL-based ingestion concepts。

无论走哪条路径，摄入任务执行时都会产生任务（task） 记录；持续型 supervisor（Kafka / Kinesis）的运行状态可通过 Supervisor API 查看，任务状态与错误排查可参考 Tasks API。摄入完成后，若需要进一步优化已入库数据的分区或压缩，可通过重新索引（reindex） 或压缩（compaction） 在后台调整。