基于Hadoop的摄入

Apache Druid当前支持通过一个Hadoop摄取任务来支持基于Apache Hadoop的批量索引任务， 这些任务被提交到 Druid Overlord的一个运行实例上。详情可以查看 基于Hadoop的摄取vs基于本地批摄取的对比 来了解基于Hadoop的摄取、本地简单批摄取、本地并行摄取三者的比较。

运行一个基于Hadoop的批量摄取任务，首先需要编写一个如下的摄取规范， 然后提交到Overlord的 druid/indexer/v1/task 接口，或者使用Druid软件包中自带的 bin/post-index-task 脚本。

教程

本章包括了基于Hadoop摄取的参考文档，对于粗略的查看，可以查看 从Hadoop加载数据 教程。

任务符号

以下为一个示例任务：

{
  "type" : "index_hadoop",
  "spec" : {
    "dataSchema" : {
      "dataSource" : "wikipedia",
      "parser" : {
        "type" : "hadoopyString",
        "parseSpec" : {
          "format" : "json",
          "timestampSpec" : {
            "column" : "timestamp",
            "format" : "auto"
          },
          "dimensionsSpec" : {
            "dimensions": ["page","language","user","unpatrolled","newPage","robot","anonymous","namespace","continent","country","region","city"],
            "dimensionExclusions" : [],
            "spatialDimensions" : []
          }
        }
      },
      "metricsSpec" : [
        {
          "type" : "count",
          "name" : "count"
        },
        {
          "type" : "doubleSum",
          "name" : "added",
          "fieldName" : "added"
        },
        {
          "type" : "doubleSum",
          "name" : "deleted",
          "fieldName" : "deleted"
        },
        {
          "type" : "doubleSum",
          "name" : "delta",
          "fieldName" : "delta"
        }
      ],
      "granularitySpec" : {
        "type" : "uniform",
        "segmentGranularity" : "DAY",
        "queryGranularity" : "NONE",
        "intervals" : [ "2013-08-31/2013-09-01" ]
      }
    },
    "ioConfig" : {
      "type" : "hadoop",
      "inputSpec" : {
        "type" : "static",
        "paths" : "/MyDirectory/example/wikipedia_data.json"
      }
    },
    "tuningConfig" : {
      "type": "hadoop"
    }
  },
  "hadoopDependencyCoordinates": <my_hadoop_version>
}

还要注意，Druid会自动计算在Hadoop集群中运行的Hadoop作业容器的类路径。但是，如果Hadoop和Druid的依赖项之间发生冲突，可以通过设置 druid.extensions.hadoopContainerDruidClasspath属性。请参阅 基本druid配置中的扩展配置 。

dataSchema

该字段是必须的。 详情可以查看摄取页中的 dataSchema 部分来看它应该包括哪些部分。

ioConfig

该字段是必须的。

inputSpec

有多种类型的inputSec：

static

一种inputSpec的类型，该类型提供数据文件的静态路径。

例如，以下例子使用了静态输入路径：

"paths" : "hdfs://path/to/data/is/here/data.gz,hdfs://path/to/data/is/here/moredata.gz,hdfs://path/to/data/is/here/evenmoredata.gz"

也可以从云存储直接读取数据，例如AWS S3或者谷歌云存储。 前提是需要首先的所有Druid MiddleManager进程或者Indexer进程的类路径下安装必要的依赖库。对于S3，需要通过以下命令来安装 Hadoop AWS 模块

java -classpath "${DRUID_HOME}lib/*" org.apache.druid.cli.Main tools pull-deps -h "org.apache.hadoop:hadoop-aws:${HADOOP_VERSION}";
cp ${DRUID_HOME}/hadoop-dependencies/hadoop-aws/${HADOOP_VERSION}/hadoop-aws-${HADOOP_VERSION}.jar ${DRUID_HOME}/extensions/druid-hdfs-storage/

一旦在所有的MiddleManager和Indexer进程中安装了Hadoop AWS模块，即可将S3路径放到 inputSpec 中，同时需要有任务属性。 对于更多配置，可以查看 Hadoop AWS 模块

"paths" : "s3a://billy-bucket/the/data/is/here/data.gz,s3a://billy-bucket/the/data/is/here/moredata.gz,s3a://billy-bucket/the/data/is/here/evenmoredata.gz"

"jobProperties" : {
  "fs.s3a.impl" : "org.apache.hadoop.fs.s3a.S3AFileSystem",
  "fs.AbstractFileSystem.s3a.impl" : "org.apache.hadoop.fs.s3a.S3A",
  "fs.s3a.access.key" : "YOUR_ACCESS_KEY",
  "fs.s3a.secret.key" : "YOUR_SECRET_KEY"
}

对于谷歌云存储，需要将 GCS connector jar 安装到所有MiddleManager或者Indexer进程的 ${DRUID_HOME}/hadoop-dependencies。 一旦在所有的MiddleManager和Indexer进程中安装了GCS连接器jar包，即可将谷歌云存储路径放到 inputSpec 中，同时需要有任务属性。对于更多配置，可以查看 instructions to configure Hadoop, GCS core default 和 GCS core template.

"paths" : "gs://billy-bucket/the/data/is/here/data.gz,gs://billy-bucket/the/data/is/here/moredata.gz,gs://billy-bucket/the/data/is/here/evenmoredata.gz"

"jobProperties" : {
  "fs.gs.impl" : "com.google.cloud.hadoop.fs.gcs.GoogleHadoopFileSystem",
  "fs.AbstractFileSystem.gs.impl" : "com.google.cloud.hadoop.fs.gcs.GoogleHadoopFS"
}

granularity

一种inputSpec类型，该类型期望数据已经按照日期时间组织到对应的目录中，路径格式为： y=XXXX/m=XX/d=XX/H=XX/M=XX/S=XX (其中日期用小写表示，时间用大写表示)。

例如， 如果示例配置具有 2012-06-01/2012-06-02 时间间隔，则数据期望的路径是：

s3n://billy-bucket/the/data/is/here/y=2012/m=06/d=01/H=00
s3n://billy-bucket/the/data/is/here/y=2012/m=06/d=01/H=01
...
s3n://billy-bucket/the/data/is/here/y=2012/m=06/d=01/H=23

dataSource

一种inputSpec的类型, 该类型读取已经存储在Druid中的数据。 该类型被用来"re-indexing"(重新索引)数据和下边描述 multi 类型 inputSpec 的 "delta-ingestion"(增量摄取)。

下表中为ingestionSpec中的一些选项：

示例：

"ioConfig" : {
  "type" : "hadoop",
  "inputSpec" : {
    "type" : "dataSource",
    "ingestionSpec" : {
      "dataSource": "wikipedia",
      "intervals": ["2014-10-20T00:00:00Z/P2W"]
    }
  },
  ...
}

multi

这是一个组合类型的 inputSpec, 来组合其他 inputSpec。此inputSpec用于增量接收。您还可以使用一个 multi 类型的inputSpec组合来自多个数据源的数据。但是，每个特定的数据源只能指定一次。注意，"useNewAggs"必须设置为默认值false以支持增量摄取。

示例：

"ioConfig" : {
  "type" : "hadoop",
  "inputSpec" : {
    "type" : "multi",
    "children": [
      {
        "type" : "dataSource",
        "ingestionSpec" : {
          "dataSource": "wikipedia",
          "intervals": ["2012-01-01T00:00:00.000/2012-01-03T00:00:00.000", "2012-01-05T00:00:00.000/2012-01-07T00:00:00.000"],
          "segments": [
            {
              "dataSource": "test1",
              "interval": "2012-01-01T00:00:00.000/2012-01-03T00:00:00.000",
              "version": "v2",
              "loadSpec": {
                "type": "local",
                "path": "/tmp/index1.zip"
              },
              "dimensions": "host",
              "metrics": "visited_sum,unique_hosts",
              "shardSpec": {
                "type": "none"
              },
              "binaryVersion": 9,
              "size": 2,
              "identifier": "test1_2000-01-01T00:00:00.000Z_3000-01-01T00:00:00.000Z_v2"
            }
          ]
        }
      },
      {
        "type" : "static",
        "paths": "/path/to/more/wikipedia/data/"
      }
    ]
  },
  ...
}

强烈建议显式地在 dataSource 中的 inputSpec 中提供段列表，以便增量摄取任务是幂等的。您可以通过对Coordinator进行以下调用来获取该段列表，POST /druid/coordinator/v1/metadata/datasources/{dataSourceName}/segments?full, 请求体：[interval1，interval2，…]， 例如["2012-01-01T00:00:00.000/2012-01-03T00:00:00.000"，"2012-01-05T00:00:00.000/2012-01-07T00:00:00.000"]

tuningConfig

tuningConfig 是一个可选项，如果未指定的话，则使用默认的参数。

jobProperties

   "tuningConfig" : {
     "type": "hadoop",
     "jobProperties": {
       "<hadoop-property-a>": "<value-a>",
       "<hadoop-property-b>": "<value-b>"
     }
   }

Hadoop的 MapReduce文档 列出来了所有可能的配置参数。

在一些Hadoop分布式环境中，可能需要设置 mapreduce.job.classpath 或者 mapreduce.job.user.classpath.first 来避免类加载相关的问题。 更多详细信息可以参见 使用不同Hadoop版本的文档

partitionsSpec

段总是基于时间戳进行分区（根据 granularitySpec），并且可以根据分区类型以其他方式进一步分区。Druid支持两种类型的分区策略：hashed（基于每行中所有维度的hash）和 single_dim（基于单个维度的范围）。

在大多数情况下，建议使用哈希分区，因为相对于单一维度分区，哈希分区将提高索引性能并创建更统一大小的数据段。

基于哈希的分区

  "partitionsSpec": {
     "type": "hashed",
     "targetRowsPerSegment": 5000000
   }

哈希分区的工作原理是首先选择多个段，然后根据每一行中所有维度的哈希对这些段中的行进行分区。段的数量是根据输入集的基数和目标分区大小自动确定的。

配置项为：

单一维度范围分区

  "partitionsSpec": {
     "type": "single_dim",
     "targetRowsPerSegment": 5000000
   }

单一维度范围分区的工作原理是首先选择要分区的维度，然后将该维度分隔成连续的范围，每个段将包含该维度值在该范围内的所有行。例如，可以在维度"host"上对段进行分区,范围为"a.example.com"到"f.example.com"和"f.example.com"到"z.example.com"。 默认情况下，将自动确定要使用的维度，但可以使用特定维度替代它。

配置项为：

远程Hadoop集群

如果已经有了一个远程的Hadoop集群，确保在Druid的 _common 配置目录中包含 *.xml 文件。

如果Hadoop与Druid的版本存在依赖等问题，请查看 这些文档

Elastic MapReduce

如果集群运行在AWS上，可以使用Elastic MapReduce(EMR)来从S3中索引数据。需要以下几步：

• 创建一个 持续运行的集群
• 创建集群时，请输入以下配置。如果使用向导，则应在"编辑软件设置"下处于高级模式：

classification=yarn-site,properties=[mapreduce.reduce.memory.mb=6144,mapreduce.reduce.java.opts=-server -Xms2g -Xmx2g -Duser.timezone=UTC -Dfile.encoding=UTF-8 -XX:+PrintGCDetails -XX:+PrintGCTimeStamps,mapreduce.map.java.opts=758,mapreduce.map.java.opts=-server -Xms512m -Xmx512m -Duser.timezone=UTC -Dfile.encoding=UTF-8 -XX:+PrintGCDetails -XX:+PrintGCTimeStamps,mapreduce.task.timeout=1800000]

• 按照 Hadoop连接配置 指导，使用EMR master中 /etc/hadoop/conf 的XML文件。

Kerberized Hadoop集群

默认情况下，druid可以使用本地kerberos密钥缓存中现有的TGT kerberos票证。虽然TGT票证的生命周期有限，但您需要定期调用 kinit 命令以确保TGT票证的有效性。为了避免这个额外的外部cron作业脚本周期性地调用 kinit，您可以提供主体名称和keytab位置，druid将在启动和作业启动时透明地执行身份验证。

从具有EMR的S3加载

• 在Hadoop索引任务中 tuningConfig 部分的 jobProperties 字段中添加一下内容：

"jobProperties" : {
   "fs.s3.awsAccessKeyId" : "YOUR_ACCESS_KEY",
   "fs.s3.awsSecretAccessKey" : "YOUR_SECRET_KEY",
   "fs.s3.impl" : "org.apache.hadoop.fs.s3native.NativeS3FileSystem",
   "fs.s3n.awsAccessKeyId" : "YOUR_ACCESS_KEY",
   "fs.s3n.awsSecretAccessKey" : "YOUR_SECRET_KEY",
   "fs.s3n.impl" : "org.apache.hadoop.fs.s3native.NativeS3FileSystem",
   "io.compression.codecs" : "org.apache.hadoop.io.compress.GzipCodec,org.apache.hadoop.io.compress.DefaultCodec,org.apache.hadoop.io.compress.BZip2Codec,org.apache.hadoop.io.compress.SnappyCodec"
}

注意，此方法使用Hadoop的内置S3文件系统，而不是Amazon的EMRFS，并且与Amazon的特定功能（如S3加密和一致视图）不兼容。如果您需要使用这些特性，那么您将需要通过 其他Hadoop发行版 一节中描述的机制之一，使Amazon EMR Hadoop JARs对Druid可用。

使用其他的Hadoop

Druid在许多Hadoop发行版中都是开箱即用的。

如果Druid与您当前使用的Hadoop版本发生依赖冲突时，您可以尝试在 Druid用户组 中搜索解决方案， 或者阅读 Druid不同版本Hadoop文档

命令行版本

运行：

java -Xmx256m -Duser.timezone=UTC -Dfile.encoding=UTF-8 -classpath lib/*:<hadoop_config_dir> org.apache.druid.cli.Main index hadoop <spec_file>

可选项

• "--coordinate" - 提供要使用的Apache Hadoop版本。此属性将覆盖默认的Hadoop。一旦指定，Apache Druid将从 druid.extensions.hadoopDependenciesDir 位置寻找Hadoop依赖。
• "--no-default-hadoop" - 不要下拉默认的hadoop版本

规范文件

spec文件需要包含一个JSON对象，其中的内容与Hadoop索引任务中的"spec"字段相同。有关规范格式的详细信息，请参见 Hadoop批处理摄取。

另外， metadataUpdateSpec 和 segmentOutputPath 字段需要被添加到ioConfig中：

      "ioConfig" : {
        ...
        "metadataUpdateSpec" : {
          "type":"mysql",
          "connectURI" : "jdbc:mysql://localhost:3306/druid",
          "password" : "druid",
          "segmentTable" : "druid_segments",
          "user" : "druid"
        },
        "segmentOutputPath" : "/MyDirectory/data/index/output"
      },

同时， workingPath 字段需要被添加到tuningConfig：

  "tuningConfig" : {
   ...
    "workingPath": "/tmp",
    ...
  }

Metadata Update Job Spec

这是一个属性规范，告诉作业如何更新元数据，以便Druid集群能够看到输出段并加载它们。

这些属性应该模仿您为 Coordinator 配置的内容。

segmentOutputPath配置

workingPath配置

请注意，命令行Hadoop indexer不具备索引服务的锁定功能，因此如果选择使用它，则必须注意不要覆盖由实时处理创建的段（如果设置了实时管道）。