前提條件

已創建E-MapReduce的Druid集群和Kafka集群，詳情請參見創建集群。

背景信息

Kafka Indexing Service是Apache Druid推出的使用Apache Druid的Indexing Service服務實時消費Kafka數據的插件。該插件會在Overlord中啟動一個Supervisor，Supervisor啟動后會在Middlemanager中啟動indexing task，這些task會連接到Kafka集群消費topic數據，并完成索引創建。您只需要準備一個數據消費格式文件，通過REST API手動啟動Supervisor。

配置Druid集群與Kafka集群交互

使用Kafka Indexing Service實時消費Kafka數據

1. 在Kafka集群（或Gateway）上執行以下命令創建一個名稱為metrics的topic。

   -- 如果開啟了Kafka高安全。
   export KAFKA_OPTS="-Djava.security.auth.login.config=/etc/ecm/kafka-conf/kafka_client_jaas.conf"
   
   kafka-topics.sh --create --zookeeper emr-header-1:2181,emr-header-2:2181,emr-header-3:2181 --partitions 1 --replication-factor 1 --topic metrics

   實際創建topic時，您需要根據您的環境配置來替換上述命令中的各個參數。其中，--zookeeper參數中路徑的獲取方式是：登錄阿里云 E-MapReduce 控制臺> 進入Kafka集群的Kafka服務的配置頁面，查看zookeeper.connect配置項的值。如果您的Kafka集群是自建集群，則您需要根據集群的實際配置來替換--zookeeper參數。

2. 定義數據源的數據格式描述文件（名稱命名為metrics-kafka.json），并放置在當前目錄下（或放置在其他您指定的目錄上）。

   {
        "type": "kafka",
        "dataSchema": {
            "dataSource": "metrics-kafka",
            "parser": {
                "type": "string",
                "parseSpec": {
                    "timestampSpec": {
                        "column": "time",
                        "format": "auto"
                    },
                    "dimensionsSpec": {
                        "dimensions": ["url", "user"]
                    },
                    "format": "json"
                }
            },
            "granularitySpec": {
                "type": "uniform",
                "segmentGranularity": "hour",
                "queryGranularity": "none"
            },
            "metricsSpec": [{
                    "type": "count",
                    "name": "views"
                },
                {
                    "name": "latencyMs",
                    "type": "doubleSum",
                    "fieldName": "latencyMs"
                }
            ]
        },
        "ioConfig": {
            "topic": "metrics",
            "consumerProperties": {
                "bootstrap.servers": "emr-worker-1.cluster-xxxxxxxx:9092(您 Kafka 集群的 bootstrap.servers)",
                "group.id": "kafka-indexing-service",
                "security.protocol": "SASL_PLAINTEXT",
                "sasl.mechanism": "GSSAPI"
            },
            "taskCount": 1,
            "replicas": 1,
            "taskDuration": "PT1H"
        },
        "tuningConfig": {
            "type": "kafka",
            "maxRowsInMemory": "100000"
        }
    }

   說明

   ioConfig.consumerProperties.security.protocol和ioConfig.consumerProperties.sasl.mechanism為安全相關選項（非安全Kafka集群不需要）。

3. 執行如下命令添加Kafka Supervisor。

   curl --negotiate -u:druid -b ~/cookies -c ~/cookies -XPOST -H 'Content-Type: application/json' -d @metrics-kafka.json http://emr-header-1.cluster-1234:18090/druid/indexer/v1/supervisor

   其中--negotiate、-u、-b和-c是針對安全E-MapReduce Druid集群的選項。

4. 在Kafka集群上開啟一個Console Producer。

   # 如果開啟了Kafka高安全：
   export KAFKA_OPTS="-Djava.security.auth.login.config=/etc/ecm/kafka-conf/kafka_client_jaas.conf"
   echo -e "security.protocol=SASL_PLAINTEXT\nsasl.mechanism=GSSAPI" > /tmp/kafka-producer.conf
   
   kafka-console-producer.sh --producer.config /tmp/kafka-producer.conf --broker-list emr-header-1:9092,emr-header-2:9092,emr-header-3:9092 --topic metrics

   其中，--producer.config /tmp/kafka-producer.conf是針對安全Kafka集群的選項。

5. 在Kafka-console-producer.sh的命令提示符下輸入數據。

   {"time": "2018-03-06T09:57:58Z", "url": "/foo/bar", "user": "alice", "latencyMs": 32}
   {"time": "2018-03-06T09:57:59Z", "url": "/", "user": "bob", "latencyMs": 11}
   {"time": "2018-03-06T09:58:00Z", "url": "/foo/bar", "user": "bob", "latencyMs": 45}

   時間戳可用如下Python命令生成。

   python -c 'import datetime; print(datetime.datetime.utcnow().strftime("%Y-%m-%dT%H:%M:%SZ"))'

6. 準備名為metrics-search.json的查詢文件。

   {
        "queryType" : "search",
        "dataSource" : "metrics-kafka",
        "intervals" : ["2018-03-02T00:00:00.000/2018-03-08T00:00:00.000"],
        "granularity" : "all",
        "searchDimensions": [
            "url",
            "user"
        ],
        "query": {
            "type": "insensitive_contains",
            "value": "bob"
        }
    }

7. 在E-MapReduce Druid集群的Master節點上執行如下命令。

   curl --negotiate -u:Druid -b ~/cookies -c ~/cookies -XPOST -H 'Content-Type: application/json' -d @metrics-search.json http://emr-header-1.cluster-1234:18082/druid/v2/?pretty

   其中--negotiate、-u、-b和-c是針對安全 E-MapReduce Druid集群的選項。

   返回結果示例如下。

   [ {
      "timestamp" : "2018-03-06T09:00:00.000Z",
      "result" : [ {
        "dimension" : "user",
        "value" : "bob",
        "count" : 2
      } ]
    } ]