对 pyspark 中的列进行重新分区如何影响分区数？

Question

我有一个包含一百万条记录的数据框。它看起来像这样 -

df.show()

+--------------------+--------------------++-------------
|            feature1|            feature2| domain    |
+--------------------+--------------------++-------------
|[2.23668528E8, 1....|[2.23668528E8, 1....| domain1   | 
|[2.23668528E8, 1....|[2.23668528E8, 1....| domain2   |
|[2.23668528E8, 1....|[2.23668528E8, 1....| domain1   |
|[2.23668528E8, 1....|[2.23668528E8, 1....| domain2   |
|[2.23668528E8, 1....|[2.23668528E8, 1....| domain1   |

Spark 中的理想分区大小为 128 MB，假设域列有两个唯一值（域 1 和域 2），考虑到这一点，我有两个问题 -

1. 如果我这样做 df.repartition("domain") 并且如果一个分区无法容纳特定域密钥的所有数据，应用程序会失败还是会根据数据自动创建适合的分区？

2. 假设在上面的数据中，基于域键的重新分区已经发生，所以会有两个分区（唯一的键是 domain1 和 domain2）。现在假设 domain1 和 domain2 重复了 1000000 次，我将基于域进行自加入。因此，对于每个域，我将获得大约 10^12 条记录。考虑到我们有两个分区，并且在连接过程中分区的数量没有变化，那么这两个新的分区是否能够处理 1000000 条记录？

Answer 1

答案取决于数据的大小。当一个分区不能保存属于一个分区值的所有数据时（例如domain1），将创建更多的分区，最多spark.sql.shuffle.partitions许多。如果您的数据太大，即一个分区将超过 2GB 的限制（另请参阅 Why does Spark RDD partition has 2GB limit for HDFS? 以获取有关说明），重新分区将导致 OutOfMemoryError。
就像提供完整答案的旁注一样：能够将数据放入一个分区中并不一定意味着只为一个分区值生成一个分区。这取决于 - 其中包括 - 执行者的数量以及数据之前的分区方式。 Spark 会尽量避免不必要的洗牌，因此可以为一个分区值生成多个分区。

因此，为防止作业失败，您应该调整spark.sql.shuffle.partitions 或将所需的分区数与分区列一起传递给repartition。