Spark镶木地板零件编号不连续？答案

【问题标题】：Spark parquet part number not sequential?Spark镶木地板零件编号不连续？
【发布时间】：2020-03-19 04:29:22
【问题描述】：

我正在使用 spark 3.0 对数据进行转换和排序。让我担心的是，我将数据分成 250 个桶，排序然后保存。我在输出目录中看到的通常是 160 个文件，文件名以 part-0000-** 开头，但编号顺序不一致：

part-00000-f667f314-69f2-40d0-ba8a-b5b934650158-c000.gz.parquet
part-00002-f667f314-69f2-40d0-ba8a-b5b934650158-c000.gz.parquet
part-00005-f667f314-69f2-40d0-ba8a-b5b934650158-c000.gz.parquet

那么第 1、3 和 4 部分发生了什么？这发生在整个文件中。我是在丢失数据还是由于某种原因正在合并分区？代码是这样的：

 df = df.withColumn("bucket", F.abs(F.col("Id")) % F.lit(250))
 df = df.repartition(250,"bucket")
 df = df.sortWithinPartitions("id")
 df.write.option("compression", "gzip").parquet(outputPath)

【问题讨论】：

标签： apache-spark pyspark parquet partition

【解决方案1】：

您需要查看 id 列的分布。在第一行中，如果 id 使用 250 进行模运算，这意味着如果值 1、3,4 等不存在，则不会创建这些 id。在下一步中，您将使用 repartition 函数创建 250 个分区。其中一些分区可能是空的。

您可以尝试执行df.repartition("bucket")，这将创建至少 200 个分区，这是默认分区值。如果您想没有间隙，那么您可以进行计数并在此基础上使用重新分区。

【讨论】：

df.repartition("bucket") 和df.repartition(250, "bucket") 之间的唯一区别是前者使用默认的分区数，除非另有配置，否则为 200。它不会改变许多分区可能是空的并且可能（并且很可能会）在 ID 中存在间隙的事实。编写测试代码很简单。
谢谢。这确实有道理。有 250 万个唯一 ID，ID 实际上是一个 64 位哈希。我希望分布均匀，但正如你所说，不能保证每个存储桶都有 id，但我很容易测试
@HristoIliev - 是的，你是对的，我忘了提到 .write.partitionBy(COL) 目录的数量是根据 COL 的唯一值创建的。
@Jayadeep Jayaraman 我不写分区的原因是这似乎破坏了排序，这是该过程中最重要的部分。
是的，我看到了您最初的帖子，并看到您正在写入非分区表。如果这个答案有用，请接受我的回答。

【解决方案2】：

如果您想获得没有间隙的分区 ID，则必须确保分区数匹配（或通常小于或等于）唯一分区键值的数量：

df = df.withColumn("bucket", F.abs(F.col("Id")) % F.lit(250))
nkeys = df.select("bucket").distinct().count()
df = df.repartition(nkeys, "bucket")
df = df.sortWithinPartitions("id")
df.write.option("compression", "gzip").parquet(outputPath)

另外，请注意，缺少某些分区 ID 意味着这些分区是空的，这意味着您的 id 值具有特定的分布。根据您要对数据重新分区的原因，使用 id 列 (df.repartition(250, "id")) 更有可能为您提供 250 个大小几乎相等的分区。这是因为 Spark 实际上使用了分区键的哈希，而哈希函数更有可能将原始分布转化为均匀分布。

【讨论】：

感谢您的回复。无论如何，ID实际上是一个xx64（spark 3.0）哈希，所以我希望它无论如何都是统一的。我将不得不研究为什么 250 万个 xx64 哈希的分布不那么均匀。我实际上并不关心顺序文件名，我只是担心丢失数据。这反过来又让我质疑 id 哈希。