我想重新分区/合并我的数据，以便将其保存到每个分区的一个 Parquet 文件中。我还想使用 Spark SQL partitionBy API。所以我可以这样做：
df.coalesce(1)
.write
.partitionBy("entity", "year", "month", "day", "status")
.mode(SaveMode.Append)
.parquet(s"$location")
我已经对此进行了测试，但它似乎表现不佳。这是因为数据集中只有一个分区可以处理，文件的所有分区、压缩和保存都必须由一个 CPU 内核完成。
我可以在调用 coalesce.
但是使用标准 Spark SQL API 是否有更好的方法来做到这一点？
【问题讨论】：
标签： apache-spark apache-spark-sql
我遇到了完全相同的问题，我找到了一种使用DataFrame.repartition() 的方法。使用coalesce(1) 的问题是你的并行度下降到1，它可能会很慢，最坏的情况是出错。增加这个数字也无济于事——如果你这样做coalesce(10)，你会得到更多的并行度，但最终每个分区有 10 个文件。
要在不使用coalesce() 的情况下为每个分区获取一个文件，请将repartition() 与您希望输出分区的相同列一起使用。所以在你的情况下，这样做：
import spark.implicits._
df.repartition($"entity", $"year", $"month", $"day", $"status").write.partitionBy("entity", "year", "month", "day", "status").mode(SaveMode.Append).parquet(s"$location")
一旦我这样做了，我会为每个输出分区获得一个 parquet 文件，而不是多个文件。
我在 Python 中对此进行了测试，但我认为在 Scala 中应该是相同的。
【讨论】：
$"entity"。
根据定义：
coalesce(numPartitions: Int): 数据帧 返回一个新的 DataFrame，它正好有 numPartitions 个分区。
您可以通过 numPartitions 参数使用它来减少 RDD/DataFrame 中的分区数。它对于过滤大型数据集后更有效地运行操作很有用。
关于您的代码，它表现不佳，因为您实际上在做的是：
将所有内容放入 1 个分区，这会使驱动程序过载，因为它将所有数据拉入驱动程序上的 1 个分区（这也不是一个好习惯）
coalesce 实际上是对网络上的所有数据进行洗牌，这也可能导致性能损失。
shuffle 是 Spark 用于重新分配数据的机制，以便跨分区以不同方式分组。这通常涉及跨执行器和机器复制数据，这使得 shuffle 成为一项复杂且成本高昂的操作。
shuffle 概念对于管理和理解非常重要。总是最好尽可能地洗牌，因为这是一项昂贵的操作，因为它涉及磁盘 I/O、数据序列化和网络 I/O。为了组织 shuffle 的数据，Spark 生成任务集 - 映射任务来组织数据，以及一组 reduce 任务来聚合它。此命名法来自 MapReduce，与 Spark 的 map 和 reduce 操作没有直接关系。
在内部，各个地图任务的结果会保存在内存中，直到无法容纳为止。然后，这些根据目标分区排序并写入单个文件。在 reduce 方面，任务读取相关的排序块。
关于分区 parquet，我建议您阅读答案 here 关于 Spark DataFrames with Parquet Partitioning 以及 Spark Programming Guide for Performance Tuning 中的 section。
我希望这会有所帮助！
【讨论】：