Hive on Spark：入门

Hive on Spark 使 Hive 能够利用Apache Spark作为其执行引擎。

set hive.execution.engine=spark;

Hive on Spark 已在HIVE-7292中添加。

Version Compatibility

Hive on Spark 仅在特定版本的 Spark 上进行了测试，因此，给定版本的 Hive 仅能与特定版本的 Spark 一起使用。其他版本的 Spark 可以与给定版本的 Hive 一起使用，但这不能保证。以下是 Hive 版本及其对应的兼容 Spark 版本的列表。

Spark Installation

按照说明安装 Spark：

纱线模式：http://spark.apache.org/docs/latest/running-on-yarn.html
独立模式：https://spark.apache.org/docs/latest/spark-standalone.html

Hive on Spark 默认支持纱上的火花模式。

对于安装，请执行以下任务：

• 安装 Spark(下载预构建的 Spark，或从源代码构建程序集)。

• 安装/构建兼容版本。Hive 根pom.xml的\ <spark.version>定义了构建/测试的 Spark 版本。

• 安装/构建兼容的发行版。每个版本的 Spark 都有多个发行版，分别对应于不同的 Hadoop 版本。

• 一旦安装了 Spark，请找到并记下\ <spark-assembly-*.jar>的位置。

• 请注意，您必须具有不包含 Hive 罐子的 Spark 版本。意思是不是使用 Hive 配置文件构建的。如果要使用 Parquet 表，建议也启用“ parquet 提供的”配置文件。否则，Parquet 依赖项可能会发生冲突。要从安装中删除 Hive jar，只需在 Spark 存储库下使用以下命令：

在 Spark 2.0.0 之前：

./make-distribution.sh --name "hadoop2-without-hive" --tgz "-Pyarn,hadoop-provided,hadoop-2.4,parquet-provided"

从 Spark 2.0.0 开始：

./dev/make-distribution.sh --name "hadoop2-without-hive" --tgz "-Pyarn,hadoop-provided,hadoop-2.7,parquet-provided"

从 Spark 2.3.0 开始：

./dev/make-distribution.sh --name "hadoop2-without-hive" --tgz "-Pyarn,hadoop-provided,hadoop-2.7,parquet-provided,orc-provided"

• 启动 Spark 集群

• 记下\ 。可以在 Spark 主 WebUI 中找到。

Configuring YARN

不是capacity scheduler，而是fair scheduler。这样可以公平分配 YARN 群集中作业的资源。

yarn.resourcemanager.scheduler.class=org.apache.hadoop.yarn.server.resourcemanager.scheduler.fair.FairScheduler

Configuring Hive

• 要将 Spark 依赖项添加到 Hive：

• 在 Hive 2.2.0 之前，请将火花装配罐链接到HIVE_HOME/lib。

• 从 Hive 2.2.0 开始，Spark 上的 Hive 在 Spark 2.0.0 及更高版本上运行，后者没有程序集 jar。

• 要以 YARN 模式(yarn-client 或 yarn-cluster)运行，请将以下 jar 链接到HIVE_HOME/lib。

• scala-library

  • spark-core

  • spark-network-common

  • 要以 LOCAL 模式运行(仅用于调试)，除了上面的 jar 之外，还将以下 jar 链接到HIVE_HOME/lib。

• chill-java chill jackson-module-paranamer jackson-module-scala jersey-container-servlet-core

  • 泽西服务器 json4s-ast kryoshade 的 minlog scala-xml spark-launcher

  • spark-network-shuffle spark-unsafe xbean-asm5-shaded

• 配置 Hive 执行引擎以使用 Spark：

set hive.execution.engine=spark;

有关配置 Hive 和 Remote Spark Driver 的其他属性，请参见Hive 配置属性的 Spark 部分。

• 为 Hive 配置 Spark 应用程序配置。参见：http://spark.apache.org/docs/latest/configuration.html。这可以通过将具有这些属性的文件“ spark-defaults.conf”添加到 HiveClasspath 中，或通过在 Hive 配置(hive-site.xml)上进行设置来完成。例如：

set spark.master=<Spark Master URL>
set spark.eventLog.enabled=true;
set spark.eventLog.dir=<Spark event log folder (must exist)>
set spark.executor.memory=512m;              
set spark.serializer=org.apache.spark.serializer.KryoSerializer;

配置属性详细信息

• spark.executor.memory：每个执行程序进程要使用的内存量。

• spark.executor.cores：每个执行程序的核心数。

• spark.yarn.executor.memoryOverhead：在 Yarn 上运行 Spark 时，将为每个执行程序分配的堆外内存量(以兆字节为单位)。这是用于解决 VM 开销，内部字符串，其他本机开销等问题的内存。除了执行程序的内存外，启动执行程序的容器还需要一些额外的内存用于系统进程，这就是这些开销对于。

• spark.executor.instances：分配给每个应用程序的执行程序的数量。

• spark.driver.memory：分配给远程 Spark 上下文(RSC)的内存量。我们建议使用 4GB。

• spark.yarn.driver.memoryOverhead：我们建议 400(MB)。

• 允许 Yarn 在节点上缓存必需的 spark 依赖项 jar，以便不需要在每次运行应用程序时将其分发。

• 在 Hive 2.2.0 之前，请将 spark-assembly jar 上传到 hdfs 文件(例如：hdfs://xxxx:8020/spark-assembly.jar)，然后在 hive-site.xml 中添加以下内容

<property>
  <name>spark.yarn.jar</name>
  <value>hdfs://xxxx:8020/spark-assembly.jar</value>
</property>

• Hive 2.2.0，将$ SPARK_HOME/jars 中的所有 jar 都上传到 hdfs 文件夹(例如：hdfs:///xxxx:8020/spark-jars)，然后在 hive-site.xml 中添加以下内容

<property>
  <name>spark.yarn.jars</name>
  <value>hdfs://xxxx:8020/spark-jars/*</value>
</property>

Configuring Spark

设置执行程序的内存大小比简单地设置尽可能大要复杂得多。有几件事情需要考虑：

• 更多的执行程序内存意味着可以为更多查询启用 mapjoin 优化。

• 另一方面，从 GC 的角度来看，更多的执行者 Memory 变得笨拙。

• 一些实验表明，HDFSClient 端不能很好地处理并发编写器，因此，如果执行者内核过多，它可能会面临竞争状态。

需要针对集群调整以下设置，这些设置也可能适用于在 Hive on Spark 之外提交 Spark 作业：

Tuning Details

在 YARN 模式下运行 Spark 时，我们通常建议将 spark.executor.cores 设置为 5、6 或 7，这取决于可被典型节点整除的对象。例如，如果 yarn.nodemanager.resource.cpu-vcores 为 19，则 6 是一个更好的选择(所有执行程序只能具有相同数量的内核，在这里，如果我们选择 5，则每个执行程序只能获得 3 个内核；如果我们选择选择 7，则仅使用 2 个执行程序，并且将浪费 5 个内核)。如果是 20，那么 5 是一个更好的选择(因为这样，您将获得 4 个执行程序，并且不会浪费任何核心资源)。

对于spark.executor.memory，我们建议计算出 yarn.nodemanager.resource.memory-mb *(spark.executor.cores/yarn.nodemanager.resource.cpu-vcores)，然后将其在 spark.executor.memory 和spark.yarn.executor.memoryOverhead之间进行分割。根据我们的实验，我们建议将spark.yarn.executor.memoryOverhead设置为总内存的 15-20％。

在确定每个执行程序可以接收多少内存之后，您需要确定将为查询分配多少个执行程序。在 GA 版本中，将支持 Spark 动态执行程序分配。但是，对于此 Beta 版，只能使用静态资源分配。根据每个节点中的物理内存以及spark.executor.memory和spark.yarn.executor.memoryOverhead的配置，您将需要选择实例数并设置spark.executor.instances。

现在是一个真实的例子。假设每个节点具有 12 个虚拟内核(例如yarn.nodemanager.resource.cpu-vcores=12)的 10 个节点具有 64GB 的内存。一个节点将用作主节点，因此群集将具有 9 个从节点。我们将spark.executor.cores配置为 6.假设 ram yarn.nodemanager.resource.memory-mb的 64GB 为 50GB。我们将确定每个 Actuator 的内存量，如下所示：50GB *(6/12)= 25GB。我们将为spark.yarn.executor.memoryOverhead或 5120 分配 20％，为spark.executor.memory或 20GB 分配 80％。

在这 9 个节点的群集上，每个主机将有两个执行程序。这样，我们可以将spark.executor.instances配置为 2 到 18 之间的某个值。值 18 会利用整个群集。

常见问题(绿色已解决，将从此列表中删除)

Recommended Configuration

有关这些设置的详细信息，请参见HIVE-9153。

mapreduce.input.fileinputformat.split.maxsize=750000000
hive.vectorized.execution.enabled=true

hive.cbo.enable=true
hive.optimize.reducededuplication.min.reducer=4
hive.optimize.reducededuplication=true
hive.orc.splits.include.file.footer=false
hive.merge.mapfiles=true
hive.merge.sparkfiles=false
hive.merge.smallfiles.avgsize=16000000
hive.merge.size.per.task=256000000
hive.merge.orcfile.stripe.level=true
hive.auto.convert.join=true
hive.auto.convert.join.noconditionaltask=true
hive.auto.convert.join.noconditionaltask.size=894435328
hive.optimize.bucketmapjoin.sortedmerge=false
hive.map.aggr.hash.percentmemory=0.5
hive.map.aggr=true
hive.optimize.sort.dynamic.partition=false
hive.stats.autogather=true
hive.stats.fetch.column.stats=true
hive.vectorized.execution.reduce.enabled=false
hive.vectorized.groupby.checkinterval=4096
hive.vectorized.groupby.flush.percent=0.1
hive.compute.query.using.stats=true
hive.limit.pushdown.memory.usage=0.4
hive.optimize.index.filter=true
hive.exec.reducers.bytes.per.reducer=67108864
hive.smbjoin.cache.rows=10000
hive.exec.orc.default.stripe.size=67108864
hive.fetch.task.conversion=more
hive.fetch.task.conversion.threshold=1073741824
hive.fetch.task.aggr=false
mapreduce.input.fileinputformat.list-status.num-threads=5
spark.kryo.referenceTracking=false
spark.kryo.classesToRegister=org.apache.hadoop.hive.ql.io.HiveKey,org.apache.hadoop.io.BytesWritable,org.apache.hadoop.hive.ql.exec.vector.VectorizedRowBatch

有关其他属性，请参见配置页面的 Spark 部分。

Design documents

• Hive on Spark：总体设计来自HIVE-7272

• Hive on Spark：加入设计(HIVE-7613)

• Hive on Spark 配置(HIVE-9449)

• Hive on Spark 解释计划