Spark 按列重新分区，每列动态分区数

如何根据列中的项目数对 DataFrame 进行分区。假设我们有一个包含 100 人的 DataFrame（列是first_name and country），我们希望为一个国家/地区的每 10 个人创建一个分区。

如果我们的数据集包含 80 个中国人、15 个法国人和 5 个古巴人，那么我们需要 8 个中国分区、2 个法国分区和 1 个古巴分区。

这是不起作用的代码：

• df.repartition($"country")：这将为中国创建 1 个分区，为法国创建 1 个分区，为古巴创建 1 个分区
• df.repartition(8, $"country", rand)：这将为每个国家创建最多 8 个分区，因此应该为中国创建 8 个分区，但法国和古巴分区未知。法国可以分为 8 个分区，古巴最多可以分为 5 个分区。看这个答案 https://stackoverflow.com/questions/53037124/partitioning-a-large-skewed-dataset-in-s3-with-sparks-partitionby-method更多细节。

这是repartition()文档：

当我看着repartition()方法，我什至没有看到带有三个参数的方法，所以看起来有些行为没有记录。

有没有办法动态设置每列的分区数量？这将使创建分区数据集变得更加容易。

由于 Spark 数据分区的方式，您将无法完全实现这一点。 Spark 获取您在重新分区中指定的列，将该值散列为 64b 长，然后将该值除以分区数。这样分区的数量是确定的。它以这种方式工作的原因是，除了确保两侧的散列相同之外，连接还需要连接左侧和右侧的分区数量匹配。

“我们希望为一个国家的每 10 个人创建一个分区。”

您到底想在这里完成什么？一个分区中只有 10 行可能会降低性能。您是否正在尝试创建一个分区表，其中分区中的每个文件都保证只有 x 行数？

"df.repartition($"country"): 这将为中国创建 1 个分区，为法国创建 1 个分区，为古巴创建 1 个分区"

这实际上会创建一个数据框，其中包含按国家/地区散列的默认洗牌分区数

  def repartition(partitionExprs: Column*): Dataset[T] = {
    repartition(sparkSession.sessionState.conf.numShufflePartitions, partitionExprs: _*)
  }

"df.repartition(8, $"country", rand)：这将为每个国家创建最多 8 个分区，因此应该为中国创建 8 个分区，但法国和古巴分区未知。法国可能有 8 个分区古巴最多可以分为 5 个分区。有关更多详细信息，请参阅此答案。”

同样，这是微妙的错误。只有 8 个分区，而国家/地区基本上是在这 8 个分区中随机排列的。

编辑：最后一点澄清。数据帧重新分区的工作方式与编写分区时执行的partitionBy(...) 方法不同。 partitionBy 操作 Spark 首先获取所有 Spark 分区，而不是每个 Spark 分区。是将其切片为表分区，然后将每个分区写入与分区列对应的文件夹中。