📚 课程目录第4章实时摄入配置

4. 实时摄入配置：Tranquility与Kafka Indexing Service的配置详解，从Spec文件到任务提交

实时摄入是Druid最吸引人的能力之一。说白了，就是数据一边进来，你一边就能查。但配置起来，坑也不少。我最早接触Druid时，就被Tranquility和Kafka Indexing Service这两个东西搞晕过——到底该用哪个？Spec文件怎么写？任务怎么提交？

今天咱们就把这两条路彻底走一遍。从Spec文件的结构，到任务提交的细节，我会把我在项目中踩过的坑、总结的经验，全都摊开来讲。

4.1 两条路：Tranquility vs. Kafka Indexing Service

先搞清楚概念。Druid实时摄入有两条主流路径：

对比项	Tranquility	Kafka Indexing Service
本质	独立的Java/Scala客户端库	Druid内置的Overlord任务
部署方式	需要单独启动Tranquility进程或嵌入应用	直接提交任务到Overlord
数据源	支持Kafka、Storm、Spark Streaming等	原生支持Kafka
容错机制	依赖Kafka offset管理，重启后可能重复消费	内置offset管理，任务重启自动恢复
适用场景	老项目、非Kafka数据源、需要灵活控制	新项目、纯Kafka场景、希望简化运维

我个人习惯：新项目一律用Kafka Indexing Service。为什么？因为它跟Druid集成得更深，任务状态、offset管理、自动伸缩都做得更好。Tranquility更像一个“外挂”，虽然也能用，但多一个组件就多一个故障点。

我的建议：如果你还在用Tranquility，可以考虑迁移到Kafka Indexing Service。Druid官方从0.10.0版本开始就主推后者了。迁移成本不高，收益却很实在。

4.2 Spec文件：实时摄入的“灵魂”

不管选哪条路，你都得写一个Spec文件。这个文件定义了数据怎么来、怎么处理、怎么存。我见过太多人因为Spec写错，导致数据丢了一整天。

先看一个完整的Kafka Indexing Service Spec示例：

{
  "type": "kafka",
  "dataSchema": {
    "dataSource": "my_events",
    "timestampSpec": {
      "column": "event_time",
      "format": "iso"
    },
    "dimensionsSpec": {
      "dimensions": ["user_id", "page_url", "action"]
    },
    "metricsSpec": [
      {"type": "count", "name": "count"},
      {"type": "longSum", "name": "total_duration", "fieldName": "duration"}
    ],
    "granularitySpec": {
      "segmentGranularity": "HOUR",
      "queryGranularity": "MINUTE",
      "rollup": true
    }
  },
  "ioConfig": {
    "topic": "raw_events",
    "consumerProperties": {
      "bootstrap.servers": "kafka-broker:9092",
      "group.id": "druid-ingest-group"
    },
    "taskCount": 1,
    "replicas": 1,
    "taskDuration": "PT1H"
  },
  "tuningConfig": {
    "type": "kafka",
    "maxRowsPerSegment": 5000000,
    "intermediatePersistPeriod": "PT10M"
  }
}

这个文件分三大部分：dataSchema、ioConfig、tuningConfig。咱们一个一个拆开讲。

4.2.1 dataSchema：数据怎么处理

这部分定义了数据源名称、时间戳、维度、指标，以及分桶策略。

dataSource：数据源名称，相当于数据库里的表名。注意，一旦创建就不能改，除非你删了重建。
timestampSpec：指定时间戳字段和格式。我遇到过有人把字段名写错，结果所有数据的时间都变成了1970-01-01。嗯，排查了半天。
dimensionsSpec：维度字段列表。这里有个坑：如果你不显式声明维度，Druid会自动推断。但自动推断有时会把数值型字段当成维度，导致聚合结果不对。我建议永远显式声明所有维度。
metricsSpec：指标定义。常用的有count、longSum、doubleSum、hyperUnique等。
granularitySpec：分桶粒度。segmentGranularity决定Segment的时间跨度，queryGranularity决定查询时的最小时间单位。rollup决定是否预聚合。

我曾经踩过的坑：有一次我把segmentGranularity设成了DAY，但数据量特别大，单个Segment超过了2GB。Druid查询时直接OOM。后来改成HOUR，问题解决。记住：Segment大小控制在300MB-700MB之间最理想。

4.2.2 ioConfig：数据从哪里来

这部分配置Kafka连接信息和消费策略。

参数	说明	我的建议值
topic	Kafka主题名	与生产端保持一致
consumerProperties	Kafka消费者配置	bootstrap.servers必填，group.id建议自定义
taskCount	并行任务数	1-3，取决于分区数
replicas	副本数	1-2，生产环境建议2
taskDuration	单个任务运行时长	PT1H（1小时）

这里有个关键点：taskCount不要超过Kafka分区数。否则多出来的任务会闲置，浪费资源。我见过有人设了10个任务，但Kafka只有3个分区，结果7个任务一直在空转。

4.2.3 tuningConfig：性能怎么调

这部分控制Segment生成和持久化的行为。

maxRowsPerSegment：单个Segment的最大行数。默认500万，如果数据量小可以调低，加快Segment生成。
intermediatePersistPeriod：中间持久化间隔。默认10分钟。如果数据很重要，可以调短到5分钟，减少数据丢失风险。
maxPendingPersists：最大等待持久化的数量。默认0（不限制）。如果内存紧张，可以设一个上限。

一个小技巧：如果数据延迟敏感，可以把intermediatePersistPeriod调短，但不要低于1分钟。太频繁的持久化会增加磁盘IO，反而拖慢整体速度。

4.3 任务提交：两种方式

Spec写好了，怎么提交？有两种方式：

4.3.1 通过Overlord API提交

这是最直接的方式。用curl或者任何HTTP客户端，把Spec POST到Overlord的/druid/indexer/v1/task接口。

curl -X POST \
  -H 'Content-Type: application/json' \
  -d @my_spec.json \
  http://overlord-host:8090/druid/indexer/v1/task

返回结果会包含一个taskId。你可以用这个ID查询任务状态：

curl http://overlord-host:8090/druid/indexer/v1/task/{taskId}/status

我习惯写一个简单的脚本，把提交和状态查询封装起来。这样每次上线新数据源时，跑一遍脚本就行。

4.3.2 通过控制台提交

Druid Web UI也支持提交任务。在Ingestion页面，选择Kafka，然后粘贴Spec内容即可。这种方式适合测试环境快速验证。

但生产环境我建议用API。为什么？因为可以自动化。你想想看，每次上线都要手动点页面，万一漏了某个参数怎么办？

4.4 核心流程：一张图看懂

说了这么多，咱们用一张SVG图把整个流程串起来。

4.5 Tranquility配置要点

虽然我推荐Kafka Indexing Service，但如果你还在用Tranquility，这里也提几个要点。

Tranquility的配置通常写在tranquility-server.yaml里。核心结构类似：

dataSources:
  my_events:
    spec:
      dataSchema:
        dataSource: my_events
        parser:
          type: string
          parseSpec:
            format: json
            timestampSpec:
              column: event_time
              format: iso
            dimensionsSpec:
              dimensions: [user_id, page_url, action]
        metricsSpec:
          - type: count
            name: count
        granularitySpec:
          segmentGranularity: HOUR
          queryGranularity: MINUTE
    properties:
      task.partitions: 1
      task.replicants: 1

注意，Tranquility的Spec格式跟Kafka Indexing Service略有不同。比如parser和parseSpec是Tranquility特有的。如果你从Tranquility迁移到Kafka Indexing Service，需要调整这部分。

我曾经踩过的坑：Tranquility的task.partitions参数控制并发度。有一次我设了3，但Kafka只有1个分区，结果数据被重复消费了3次。后来查文档才发现，这个参数必须小于等于分区数。

4.6 任务监控与调试

任务提交后，怎么知道它跑得好不好？

Overlord控制台：查看任务状态（RUNNING、SUCCESS、FAILED）。如果任务失败，点进去看日志。
MiddleManager日志：每个任务会在MiddleManager上生成日志文件。路径一般是/var/log/druid/indexing/{taskId}.log。
Kafka消费延迟：用kafka-consumer-groups命令查看消费组的lag。如果lag持续增长，说明摄入速度跟不上生产速度。

我习惯写一个简单的监控脚本，每隔5分钟检查一次任务状态和Kafka lag。一旦发现异常，立刻报警。毕竟实时数据丢了，想补都补不回来。

4.7 小结

实时摄入的配置，说白了就是三件事：写对Spec、选对方式、盯紧监控。Spec文件是核心，ioConfig决定数据来源，tuningConfig决定性能表现。任务提交推荐用API，方便自动化。监控不能省，尤其是Kafka lag和任务状态。