OLAP之Druid之实时数据摄入

实时数据摄入

我们采用Kafka Indexing Service作为实时摄入数据的方案。

准备工作

将数据实时灌入某个Kafka topic中
与批量导入数据类似：考虑清楚数据中哪一列可以作为时间列、哪些列可以作为维度列、哪些列可以作为指标列（尤其是指标的聚合函数，包括count、sum、max、min等，如果涉及UV、留存的计算，则需要使用HyperUnique或者Theta sketch）
考虑最小时间粒度（即queryGranularity）和数据分片的时间粒度（即segmentGranularity），在我们的使用经验中最小时间粒度应该根据业务需求确定，而数据分片的时间粒度设为HOUR即可

提交摄取任务

官方提供的数据摄取JSON如下，可以以此为模版修改（用#开头的是我们添加的注释，正式提交的时候请将注释删除，否则不是合法JSON文件）:

{
  "type": "kafka", #注意这里的作业类型，与批量导入时不一样的
  "dataSchema": {
    "dataSource": "metrics-kafka", #导入druid之后的datasource名字，最好是可以识别团队的前缀
    "parser": {
      "type": "string",
      "parseSpec": {
        "format": "json", #原始数据的格式，可以是JSON、CSV、TSV
        "timestampSpec": { #指定导入数据的时间列以及时间格式
          "column": "timestamp",
          "format": "auto"
        },
        "dimensionsSpec": { #在此指定导入数据的维度
          "dimensions": [ #在此指定导入数据的维度
              "channel",
              "cityName",
              "comment",
              "countryIsoCode",
              "countryName",
              "isAnonymous",
              "isMinor",
              "isNew"
            ]
        }
      }
    },
    "metricsSpec": [ #指定导入数据的指标列，以及各指标列的聚合函数
      {
        "name": "count",
        "type": "count"
      },
      {
        "name": "value_sum",
        "fieldName": "value", #fieldName是原始数据中的列名，name是在druid中的列名，可以不同
        "type": "doubleSum"
      },
      {
        "name": "value_min",
        "fieldName": "value",
        "type": "doubleMin"
      },
      {
        "name": "value_max",
        "fieldName": "value",
        "type": "doubleMax"
      }
    ],
    "granularitySpec": {
      "type": "uniform",
      "segmentGranularity": "HOUR", #数据分片的时间粒度
      "queryGranularity": "NONE" #最小的查询时间粒度, None则为毫秒级
    }
  },
  "tuningConfig": {
    "type": "kafka",
    "maxRowsPerSegment": 5000000
  },
  "ioConfig": {
    "topic": "metrics", #指定Kafka中的topic名
    "consumerProperties": {
      "bootstrap.servers": "localhost:9092", #指定Kafka的broker列表
      "group.id": "druidxx" # 可以指定一个消费kafka的身份，需要注意的是不能有两个druid作业以同一个身份消费同一个topic
    },
    "taskCount": 1, #task作业并发数
    "replicas": 1, #task作业的副本数
    "taskDuration": "PT1H" #单个task进程运行的时间
  }
}

提交命令

curl -X 'POST' -H 'Content-Type:application/json' -d @my-index-task.json OVERLORD_IP:8090/druid/indexer/v1/supervisor

停止消费Kafka

curl -X 'POST' -H 'Content-Type:application/json' OVERLORD_IP:8090/druid/indexer/v1/supervisor/datasource名/shutdown

数据查询

建议在首次查询之前，向Broker提交TimeBoundary查询，方便掌握Druid中数据的时间分布

{
    "queryType" : "timeBoundary",
    "dataSource": "sample_datasource" #datasource名字
}