Apache Druid中先批量创建数据源再实时流摄入追加数据的方法

核心前提 #

Druid允许同一数据源（datasource）同时包含批量生成的历史数据段和流摄入的实时数据段，核心要求是批量与流摄入的Schema完全兼容（时间列、维度、指标、粒度等定义严格一致）。

1. 批量初始化数据源（固定Schema） #

执行批量摄入任务时，明确指定统一的datasource名称，并固化Schema定义，后续流摄入必须严格匹配：

• 示例批量摄入Spec片段（以Parallel Index任务为例）：

{
  "type": "index_parallel",
  "spec": {
    "dataSchema": {
      "dataSource": "user_behavior",
      "timestampSpec": {
        "column": "event_time",
        "format": "iso"
      },
      "dimensionsSpec": {
        "dimensions": ["user_id", "event_type", "device"]
      },
      "metricsSpec": [
        {"type": "count", "name": "event_count"},
        {"type": "longSum", "name": "duration", "fieldName": "duration"}
      ],
      "granularitySpec": {
        "type": "uniform",
        "segmentGranularity": "DAY",
        "queryGranularity": "HOUR"
      }
    },
    // 补充输入源、IO配置等批量任务必要参数
    "ioConfig": { /* ... */ },
    "tuningConfig": { /* ... */ }
  }
}

• 确保批量任务执行完成，数据源生成对应的历史数据段。

2. 配置实时流摄入任务（匹配Schema） #

创建流摄入任务（以Kafka源为例）时，datasource名称必须与批量任务完全一致，且Schema严格对齐：

• 示例Kafka流摄入Spec片段：

{
  "type": "kafka",
  "spec": {
    "dataSchema": {
      "dataSource": "user_behavior",
      "timestampSpec": {
        "column": "event_time",
        "format": "iso"
      },
      "dimensionsSpec": {
        "dimensions": ["user_id", "event_type", "device"]
      },
      "metricsSpec": [
        {"type": "count", "name": "event_count"},
        {"type": "longSum", "name": "duration", "fieldName": "duration"}
      ],
      "granularitySpec": {
        "type": "uniform",
        "segmentGranularity": "DAY",
        "queryGranularity": "HOUR"
      }
    },
    "ioConfig": {
      "type": "kafka",
      "consumerProperties": {
        "bootstrap.servers": "kafka-broker:9092"
      },
      "topic": "user_behavior_topic",
      "useEarliestOffset": true
    },
    "tuningConfig": {
      "type": "kafka",
      "maxRowsPerSegment": 5000000
    }
  }
}

• 提交流摄入任务后，Druid会自动将实时数据追加到同一数据源下。

3. 关键注意事项 #

• Schema一致性：禁止批量与流摄入的Schema存在差异（如维度增减、指标类型变更），否则会导致查询失败或数据丢失。如需修改Schema，需先通过批量任务更新历史数据的Schema，再同步修改流摄入配置。
• 时间范围管控：流摄入的实时数据时间应晚于批量历史数据的时间范围，避免数据重复。若存在重复数据，可通过配置主键（primaryKey）启用Druid的 deduplication 功能，或后续执行段合并任务去重。
• 段性能优化：流摄入会生成较多小数据段，可通过Druid的自动合并规则（在tuningConfig中配置maxRowsPerSegment、intermediatePersistPeriod等参数）或手动提交段合并任务，提升查询性能。
• 任务监控：通过Druid控制台查看批量任务的完成状态、流摄入任务的运行状态，确保数据持续正常追加。

内容的提问来源于stack exchange，提问作者k yogeshwar