作者：vivo 互联网搜索团队- Deng Jie

随着技术的不断的发展，大数据领域对于海量数据的存储和处理的技术框架越来越多。在离线数据处理生态系统最具代表性的分布式处理引擎当属 Hive 和 Spark，它们在分区策略方面有着一些相似之处，但也存在一些不同之处。

一、概述

随着技术的不断的发展，大数据领域对于海量数据的存储和处理的技术框架越来越多。在离线数据处理生态系统最具代表性的分布式处理引擎当属 Hive 和 Spark，它们在分区策略方面有着一些相似之处，但也存在一些不同之处。本篇文章将分析 Hive 与 Spark 分区策略的异同点、它们各自的优缺点，以及一些优化措施。

二、Hive 和 Spark 分区概念

在了解 Hive 和 Spark 分区内容之前，首先，我们先来回顾一下 Hive 和 Spark 的分区概念。在 Hive 中，分区是指将表中的数据划分为不同的目录或者子目录，这些目录或子目录的名称通常与表的列名相关联。比如，一个名为“t_orders_name”的表可以按照日期分为多个目录，每个目录名称对应一个日期值。这样做的好处是可以大大提高查询效率，因为只有涉及到特定日期的查询才需要扫描对应的目录，而不需要去扫描整个表。Spark 的分区概念与 Hive 类似，但是有一些不同之处，我们将在后文中进行讨论。

在 Hive 中，分区可以基于多个列进行，这些列的值组合形成目录名称。例如，如果我们将“t_orders_name”表按照日期和地区分区，那么目录的名称将包含日期和地区值的组合。在 Hive 中，数据存储在分区的目录下，而不是存储在表的目录下。这使得 Hive 可以快速访问需要的数据，而不必扫描整个表。另外，Hive 的分区概念也可以用于数据分桶，分桶是将表中的数据划分为固定数量的桶，每个桶包含相同的行。

而与 Hive 不同的是，Spark 的分区是将数据分成小块以便并行计算处理。在 Spark 中，分区的数量由 Spark 执行引擎根据数据大小和硬件资源自动计算得出。Spark 的分区数越多，可以并行处理的数据也就越多，因此也能更快的完成计算任务。但是，如果分区数太多，将会导致过多的任务调度和数据传输开销，从而降低整体的性能。因此，Spark 分区数的选择应该考虑数据大小、硬件资源和计算任务复杂度等因素。

三、Hive 和 Spark 分区的应用场景

在了解 Hive 和 Spark 的分区概念之后，接下来，我们来看看 Hive 和 Spark 分区在不同的应用场景中有哪些不同的优势。

3.1 Hive 分区

Hive 分区适用于大数据场景，可以对数据进行多级分区，以便更细粒度地划分数据，提高查询效率。例如，在游戏平台的充值数据中，可以按照道具购买日期、道具付款状态、游戏用户 ID 等多个维度进行分区。这样可以方便的进行数据统计、分析和查询操作，同时避免单一分区数据过大导致的性能问题。

3.2 Spark 分区

Spark 分区适用于大规模数据处理场景，可以充分利用集群资源进行并行计算处理。比如，在机器学习算法的训练过程中，可以将大量数据进行分区，然后并行处理每个分区的数据，从而提高算法的训练速度和效率。另外，Spark 的分布式计算引擎也可以支持在多个节点上进行数据分区和计算，从而提高整个集群的计算能力和效率。

简而言之，Hive 和 Spark 分区在大数据处理和分布式计算场景这都有广泛的应用，可以通过选择合适的分区策略和优化措施，进一步提高数据处理的效率和性能。

四、如何选择分区策略

在熟悉了 Hive 和 Spark 的分区概念以及应用场景后。接下来，我们来看看在 Hive 和 Spark 中如何选择分区策略。分区策略的选择对数据处理的效率和性能有着重要的影响。下面将分别阐述 Hive 和 Spark 分区策略的优缺点以及如何选择分区策略。

4.1 Hive 分区策略

优点：

Hive 的分区策略可以提高查询效率和数据处理性能，特别是在大数据集上表现突出。另外，Hive 还支持多级分区，允许更细粒度的数据划分。

缺点：

在 Hive 中，分区是以目录的形式存在的，这会导致大量的目录和子目录，如果分区过多，将会占用过多的存储空间。此外，Hive 的分区策略需要在创建表时进行设置，如果数据分布出现变化，需要重新设置分区策略。

4.2 Spark 分区策略

优点：

Spark 的分区策略可以根据数据大小和硬件资源自动计算分区数，这使得计算任务可以并行计算处理，从而提高了处理效率和性能。

缺点：

如果分区数设置不当，将会导致过多的任务调度和数据传输开销，从而影响整体性能。此外，Spark 的分区策略也需要根据数据大小、硬件资源和计算任务复杂度等因素进行调整。

4.3 分区策略选择

在实际项目开发使用中，选择合适的分区策略可以显著提高数据处理的效率和性能。但是，如何选择分区策略需要根据具体情况进行考虑，这里总结了一些分区策略选择的场景：

数据集大小：如果数据集较大，可以考虑使用 Hive 的多级划分策略，以便更细粒度的划分数据，提高查询效率。如果数据集较小，可以使用 Spark 自动计算分区策略，以便充分利用硬件资源并提高计算效率。

计算任务复杂度：如果计算任务比较复杂，例如需要进行多个 JOIN 操作，可以使用 Hive 的分桶策略，以便加快数据访问速度，减少 JOIN 操作的开销。

硬件资源：分区策略的选择也需要考虑硬件资源的限制。如果硬件资源比较充足，可以增加分区数以提高计算效率。如果硬件资源比较紧张，需要减少分区数以避免任务调度和数据传输的开销。

综上所述，选择合适的分区策略需要根据具体的情况进行考虑，包括数据集大小、计算任务复杂度和硬件资源等因素。在实际使用中，可以通过实验和调试来找到最佳的分区策略。

五、如何优化分区性能

除了选择合适的分区策略之外，还可以通过一些优化措施来进一步提高分区的性能。在 Spark 中，大多数的 Spark 任务可以通过三个阶段来表述，它们分别是读取输入数据、使用 Spark 处理、保持输出数据。Spark 虽然实际数据处理主要发生在内存中，但是 Spark 使用的是存储在 HDFS 上的数据来作为输入和输出，任务的调度执行会使用大量的 I/O，存在性能瓶颈。

而 Hive 分区数据是存储在 HDFS 上的，然而 HDFS 对于大量小文件支持不太友好，因为在每个 NameNode 内存中每个文件大概有 150 字节的存储开销，而整个 HDFS 集群的 IOPS 数量是有上限的。当文件写入达到峰值时，会对 HDFS 集群的基础架构的某些部分产生性能瓶颈。

5.1 通过减少 I/O 带宽来优化性能

在 Hadoop 集群中，它依靠大规模并行 I/O 来支持数千个并发任务。比如现有一个大小为 96TB 的数据节点，磁盘的大小有两种，它们分别是 8TB 和 16TB。具有 8TB 磁盘的数据节点有 12 块这样的磁盘，而具有 16TB 磁盘的数据节点有 6 块这样的磁盘。我们可以假设每个磁盘的平均读写吞吐量约为 100MB/s，而这两种不同的磁盘分布，它们对应的带宽和 IOPS，具体详情如下表所示：

5.2 通过设置参数来优化性能

在 Hadoop 集群中，每个数据节点为每个卷运行一个卷扫描器，用于扫描块的状态。由于卷扫描器与应用程序竞争磁盘资源，因此限制其磁盘带宽很重要。配置 dfs.block.scanner.volume.bytes.per.second 属性值来定义卷扫描器每秒可以扫描的字节数，默认为 1MB/s。

比如设置带宽为 5MB/s，扫描 12TB 所需要的时间为

12TB / 5MBps = (12 * 1024 * 1024 / (3600 * 24)) = 29.13 天。

5.3 通过优化 Spark 处理分区任务来提升性能

假如，现在需要重新计算历史分区的数据表，这种场景通常用于修复错误或者数据质量问题。在处理包含一年数据的大型数据集（比如 1TB 以上）时，可能会将数据分成几千个 Spark 分区来进行处理。虽然，从表面上看，这种处理方法并不是最合适的，使用动态分区并将数据结果写入按照日期分区的 Hive 表中将产生多达上百万个文件。

下面，我们将任务分区数缩小，现有一个包含 3 个分区的 Spark 任务，并且想将数据写入到包含 3 个分区的 Hive 表。在这种情况下，希望发送的是将 3 个文件写入到 HDFS 中，所有数据都存储在每个分区的单个文件中。最终会生成 9 个文件，并且每个文件都有 1 个记录。使用动态分区写入 Hive 表时，每个 Spark 分区都由执行程序来并行处理。

处理 Spark 分区数据时，每次执行程序在给定的 Spark 分区中遇到新的分区时，它都会打开一个新文件。默认情况下，Spark 对数据会使用 Hash 或者 Round Robin 分区器。当应用于任意数据时，可以假设这两种方法在整个 Spark 分区中相对均匀且随机分布数据。如下图所示：

理想情况下，目标文件大小应该大约是 HDFS 块大小的倍数，默认情况下是 128MB。在 Hive 中，提供了一些配置参数来自动将结果写入到合理大小的文件中，从开发者的角度来看几乎是透明的，比如设置属性 hive.merge.smallfiles.avgsize 和 hive.merge.size.per.task 。但是，Spark 中不存在此类功能，因此，我们需要自己开发实现，来确定一个数据集，应该写入多少文件。

5.3.1 基于大小的计算

理论上，这是最直接的方法，设置目标大小，估算数据的大小，然后进行划分。但是，在很多情况下，文件被写入磁盘时会进行压缩，并且其格式与存储在 Java 堆中的记录格式有所不同。这意味着估算写入磁盘时内存的记录大小不是一件容易的事情。虽然可以使用 Spark SizeEstimator 应用程序通过内存中的数据的大小进行估算。但是，SizeEstimator 会考虑数据帧、数据集的内部消耗，以及数据的大小。总体来说，这种方式不太容易准确实现。

5.3.2 基于行数的计算

这种方法是设置目标行数，计算数据集的大小，然后执行除法来估算目标。我们的目标行数可以通过多种方式确定，或者通过为所有数据集选择一个静态数字，或者通过确定磁盘上单个记录的大小并执行必要的计算。哪种方式最优，取决于你的数据集数量及其复杂性。计算相对来说成本较低，但是需要在计算前缓存以避免重新计算数据集。

5.3.3 静态文件计算

最简单的解决方案是，只要求开发者在每个写入任务的基础上，告诉 Spark 总共应该写入多少个文件。这种方式需要给开发者一些其他方法来获取具体的数字，可以通过这种方式来替代昂贵的计算。

5.4. 优化 Spark 分发数据方式来提升性能

即使我们知道了如何将文件写入磁盘，但是，我们仍须让 Spark 以符合实际的方式来构建我们的分区。在 Spark 中，它提供了许多工具来确定数据在整个分区中的分布方式。但是，各种功能中隐藏着很多复杂性，在某些情况下，它们的含义并不明显，下面将介绍 Spark 提供的一些选项来控制 Spark 输出文件的数量。

5.4.1 合并

Spark Coalesce 是一个特殊版本的重新分区，它只允许减少总的分区，但是不需要完全的 Shuffle，因此比重新分区要快得多。它通过有效的合并分区来实现这一点。如下图所示：

Coalesce 在某些情况下看起来是不错的，但是也有一些问题。首先，Coalesce 有一个难以使用的行为，以一个非常基础的 Spark 应用程序为例，代码如下所示：

Spark

load().map(…).filter(…).save()

比如，设置的并行度为 1000，但是最终只想写入 10 个文件，可以设置如下：

Spark

load().map(…).filter(…).coalesce(10).save()

但是，Spark 会尽可能早的有效的将合并操作下推，因此这将执行为如下代码：

Spark

load().coalesce(10).map(…).filter(…).save()

有效的解决这种问题的方法是在转换和合并之间强制执行，代码如下所示：

Spark

val df = load().map(…).filter(…).cache()df.count()df.coalesce(10)

在 Spark 中，缓存是必须的，否则，你将不得不重新计算数据，这可能会重新消耗计算资源。然后，缓存是需要消费一定资源的，如果你的数据集无法放入内存中，或者无法释放内存，将数据有效的存储在内存中两次，那么必须使用磁盘缓存，这有其自身的局限性和显著的性能损失。

此外，正如我们看到的，通常需要执行 Shuffle 来获得我们想要的更复杂的数据集结果。因此，Coalesce 仅适用于特定的情况，比如如下场景：

• 保证只写入一个 Hive 分区；

• 目标文件数少于你用于处理数据的 Spark 分区数；

• 有充足的缓存资源。

5.4.2 简单重新分区

在 Spark 中，一个简单的重新分区，可以通过设置参数来实现，比如 df.repartition(100)。在这种情况下，使用循环分区器，这意味着唯一的保证是输出数据具有大致相同大小的 Spark 分区，这种分区仅适用于以下情况：

• 保证只需要写入一个 Hive 分区；

• 正在写入的文件数大于你的 Spark 分区数，或者由于某些原因你无法使用合并。

5.4.3 按列重新分区

按列重新分区接收目标 Spark 分区计数，以及要重新分区的列序列，例如，df.repartition(100,$"date")。这对于强制要求 Spark 将具有相同键的数据，分发到同一个分区很有用。一般来说，这对许多 Spark 操作（比如 JOIN）很有用。

按列重新分区使用 HashPartitioner，将具有相同值的数据，分发给同一个分区，实际上，它将执行以下操作：

但是，这种方法只有在每个分区键都可以安全的写入到一个文件时才有效。这是因为无论有多少特定的 Hash 值，它们最终都会在同一个分区中。按列重新分区仅在你写入一个或者多个小的 Hive 分区时才有效。在任何其他情况下，它都是无效的，因为每个 Hive 分区最终都会生成一个文件，仅适用于最小的数据集。

5.4.4 按具有随机因子的列重新分区

我们可以通过添加约束的随机因子来按列修改重新分区，具体代码如下：

Spark

df.withColumn("rand", rand() % filesPerPartitionKey).repartition(100, $"key", $"rand")

理论上，只要满足以下条件，这种方法应该会产生排序规则的数据和大小均匀的文件：

• Hive 分区的大小大致相同；

• 知道每个 Hive 分区的目标文件数并且可以在运行时对其进行编码。

但是，即使我们满足上述这些条件，还有另外一个问题：散列冲突。假设，现在正在处理一年的数据，日期作为分区的唯一键。如果每个分区需要 5 个文件，可以执行如下代码操作：

Spark

df.withColumn("rand", rand() % 5).repartition(5*365, $"date", $"rand")

在后台，Scala 将构造一个包含日期和随机因子的键，例如（,<0-4>）。然后，如果我们查看 HashPartitioner 代码，可以发现它将执行以下操作：

Spark

class HashPartitioner(partitions: Int) extends Partitioner {    def getPartition(key: Any): Int = key match {        case null => 0        case _ => Utils.nonNegativeMod(key.hashCode, numPartitions)    }}

实际上，这里面所做的事情，就是获取关键元组的散列，然后使用目标数量的 Spark 分区获取它的 mod。我们可以分析一下在这种情况下我们的数据将如何实现分布，具体代码如下：

Spark

import java.time.LocalDate def hashCodeTuple(one: String, two: Int, mod: Int): Int = { val rawMod = (one, two).hashCode % mod rawMod + (if (rawMod < 0) mod else 0)}def hashCodeSeq(one: String, two: Int, mod: Int): Int = { val rawMod = Seq(one, two).hashCode % mod rawMod + (if (rawMod < 0) mod else 0)} def iteration(numberDS: Int, filesPerPartition: Int): (Double, Double, Double) = {  val hashedRandKeys = (0 to numberDS - 1).map(x => LocalDate.of(2019, 1, 1).plusDays(x)).flatMap(    x => (0 to filesPerPartition - 1).map(y => hashCodeTuple(x.toString, y, filesPerPartition*numberDS))  )   hashedRandKeys.size // Number of unique keys, with the random factor   val groupedHashedKeys = hashedRandKeys.groupBy(identity).view.mapValues(_.size).toSeq   groupedHashedKeys.size // number of actual sPartitions used   val sortedKeyCollisions = groupedHashedKeys.filter(_._2 != 1).sortBy(_._2).reverse     val sortedSevereKeyCollisions = groupedHashedKeys.filter(_._2 > 2).sortBy(_._2).reverse   sortedKeyCollisions.size // number of sPartitions with a hashing collision   // (collisions, occurences)  val collisionCounts = sortedKeyCollisions.map(_._2).groupBy(identity).view.mapValues(_.size).toSeq.sortBy(_._2).reverse     (    groupedHashedKeys.size.toDouble / hashedRandKeys.size.toDouble,    sortedKeyCollisions.size.toDouble / groupedHashedKeys.size.toDouble,  sortedSevereKeyCollisions.size.toDouble / groupedHashedKeys.size.toDouble  )} val results = Seq(  iteration(365, 1),  iteration(365, 5),  iteration(365, 10),  iteration(365, 100),  iteration(365 * 2, 100),  iteration(365 * 5, 100),  iteration(365 * 10, 100)) val avgEfficiency = results.map(_._1).sum / results.lengthval avgCollisionRate = results.map(_._2).sum / results.lengthval avgSevereCollisionRate = results.map(_._3).sum / results.length (avgEfficiency, avgCollisionRate, avgSevereCollisionRate) // 63.2%, 42%, 12.6%

上面的脚本计算了 3 个数量：

• 效率：非空的 Spark 分区与输出文件数量的比率；

• 碰撞率：(date,rand)的 Hash 值发送冲突的 Spark 分区的百分比；

• 严重冲突率：同上，但是此键上的冲突次数为 3 或者更多。

冲突很重要，因为它们意味着我们的 Spark 分区包含多个唯一的分区键，而我们预计每个 Spark 分区只有 1 个。我们从分析的结果可知，我们使用了 63%的执行器，并且可能会出现严重的偏差，我们将近一半的执行正在处理比预期多 2 到 3 倍或者在某些情况下高达 8 倍的数据。

现在，有一个解决方法，即分区缩放。在之前示例中，输出的 Spark 分区数量等于预期的总文件数。如果将 N 个对象随机分配给 N 个插槽，可以预期会有多个插槽包含多个对象，并且有几个空插槽。因此，需要解决此问题，必须要降低对象与插槽的比率。

我们通过缩放输出分区计数来实现这一点，通过将输出 Spark 分区数乘以一个大因子，类似于：

Spark

df.withColumn("rand", rand() % 5).repartition(5*365*SCALING_FACTOR, $"date", $"rand")

具体分析代码如下所示：

Spark

import java.time.LocalDate def hashCodeTuple(one: String, two: Int, mod: Int): Int = { val rawMod = (one, two).hashCode % mod rawMod + (if (rawMod < 0) mod else 0)} def hashCodeSeq(one: String, two: Int, mod: Int): Int = { val rawMod = Seq(one, two).hashCode % mod rawMod + (if (rawMod < 0) mod else 0)} def iteration(numberDS: Int, filesPerPartition: Int, partitionFactor: Int = 1): (Double, Double, Double, Double) = {  val partitionCount = filesPerPartition*numberDS * partitionFactor  val hashedRandKeys = (0 to numberDS - 1).map(x => LocalDate.of(2019, 1, 1).plusDays(x)).flatMap(    x => (0 to filesPerPartition - 1).map(y => hashCodeTuple(x.toString, y, partitionCount))  )     hashedRandKeys.size // Number of unique keys, with the random factor   val groupedHashedKeys = hashedRandKeys.groupBy(identity).view.mapValues(_.size).toSeq   groupedHashedKeys.size // number of unique hashes - and thus, sPartitions with > 0 records     val sortedKeyCollisions = groupedHashedKeys.filter(_._2 != 1).sortBy(_._2).reverse     val sortedSevereKeyCollisions = groupedHashedKeys.filter(_._2 > 2).sortBy(_._2).reverse   sortedKeyCollisions.size // number of sPartitions with a hashing collision   // (collisions, occurences)  val collisionCounts = sortedKeyCollisions.map(_._2).groupBy(identity).view.mapValues(_.size).toSeq.sortBy(_._2).reverse     (    groupedHashedKeys.size.toDouble / partitionCount,    groupedHashedKeys.size.toDouble / hashedRandKeys.size.toDouble,    sortedKeyCollisions.size.toDouble / groupedHashedKeys.size.toDouble,    sortedSevereKeyCollisions.size.toDouble / groupedHashedKeys.size.toDouble  )} // With a scale factor of 1val results = Seq(  iteration(365, 1),  iteration(365, 5),  iteration(365, 10),  iteration(365, 100),  iteration(365 * 2, 100),  iteration(365 * 5, 100),  iteration(365 * 10, 100)) val avgEfficiency = results.map(_._2).sum / results.length // What is the ratio of executors / output filesval avgCollisionRate = results.map(_._3).sum / results.length // What is the average collision rateval avgSevereCollisionRate = results.map(_._4).sum / results.length // What is the average collision rate where 3 or more hashes collide (avgEfficiency, avgCollisionRate, avgSevereCollisionRate) // 63.2% Efficiency, 42% collision rate, 12.6% severe collision rate iteration(365, 5, 2) // 37.7% partitions in-use, 77.4% Efficiency, 24.4% collision rate, 4.2% severe collision rateiteration(365, 5, 5)iteration(365, 5, 10)iteration(365, 5, 100)

随着我们的比例因子接近无穷大，碰撞很快接近于 0，效率接近 100%。但是，这会产生另外一个问题，即大量 Spark 分区输出将为空。同时这些空的 Spark 分区也会带来一些资源开销，增加 Driver 的内存大小，会使我们更容易遇到，由于异常错误而导致分区键空间意外增大的问题。

这里的一个常见方法，是在使用这种方法时不显示设置分区（默认并行度和缩放），如果不提供分区计数，则依赖 Spark 默认的 spark.default.parallelism 值。虽然，通常并行度自然高于总输出文件数（因此，隐式提供大于 1 的缩放因子）。如果满足以下条件，这种方式依然是一种有效的方法：

• Hive 分区的文件数大致相等；

• 可以确定平均分区文件数应该是多少；

• 大致知道唯一分区键的总数。

5.4.5 按范围重新分区

按范围重新分区是一个特列，它不使用 RoundRobin 和 Hash Partitioner，而是使用一种特殊的方法，叫做 Range Partitioner。

范围分区器根据某些给定键的顺序在 Spark 分区之间进行拆分行，但是，它不仅仅是全局排序，而且还拥有以下特性：

• 具有相同散列的所有记录将在同一个分区中结束；

• 所有 Spark 分区都将有一个最小值和最大值与之关联；

• 最小值和最大值将通过使用采样来检测关键频率和范围来确定，分区边界将根据这些估计值进行初始设置；

• 分区的大小不能保证完全相等，它们的相等性基于样本的准确性，因此，预测的每个 Spark 分区的最小值和最大值，分区将根据需要增大或缩小来保证前两个条件。

总而言之，范围分区将导致 Spark 创建与请求的 Spark 分区数量相等的 Bucket 数量，然后它将这些 Bucket 映射到指定分区键的范围。例如，如果你的分区键是日期，则范围可能是（最小值 2022-01-01，最大值 2023-01-01）。然后，对于每条记录，将记录的分区键与存储 Bucket 的最小值和最大值进行比较，并相应的进行分配。如下图所示：

六、总结

在选择分区策略时，需要根据具体的应用场景和需求进行选择。常见的分区策略包括按照时间、地域、用户 ID 等多个维度进行分区。在应用分区策略时，还可以通过一些优化措施来进一步提高分区的性能和效率，例如合理设置分区数、避免过多的分区列、减少重复数据等。

总之，分区是大数据处理和分布式计算中非常重要的技术，可以帮助我们更好的管理和处理大规模的数据，提高数据处理的效率和性能，进而帮助我们更好的应对数据分析和业务应用的挑战。

参考：

1. https://github.com/apache/spark
   

2. https://github.com/apache/hive
   

3. https://spark.apache.org/
   

4. https://hive.apache.org/