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.length
val avgCollisionRate = results.map(_._2).sum / results.length
val 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 1
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(_._2).sum / results.length // What is the ratio of executors / output files
val avgCollisionRate = results.map(_._3).sum / results.length // What is the average collision rate
val 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 rate
iteration(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等多个维度进行分区。在应用分区策略时，还可以通过一些优化措施来进一步提高分区的性能和效率，例如合理设置分区数、避免过多的分区列、减少重复数据等。
总之，分区是大数据处理和分布式计算中非常重要的技术，可以帮助我们更好的管理和处理大规模的数据，提高数据处理的效率和性能，进而帮助我们更好的应对数据分析和业务应用的挑战。
参考：
