使用 Pyspark 覆盖 Parquet 文件

Question

我正在尝试使用 Pyspark 覆盖 S3 中的 Parquet 文件。已为存储桶启用版本控制。

我正在使用以下代码：

编写 v1：

df_v1.repartition(1).write.parquet(path='s3a://bucket/file1.parquet')

更新 v2：

df_v1 = spark.read.parquet("s3a://bucket/file1.parquet")
df_v2 = df_v1.... <- transform
df_v2.repartition(1).write.mode("overwrite").parquet('s3a://bucket/file1.parquet')

但是当我读取 df_v2 时，它包含来自两次写入的数据。此外，当编写 df_v1 时，我可以看到一个 part-xxx.snappy.parquet 文件，在编写 df_v2 之后，我可以看到两个。它表现为追加而不是覆盖。

我错过了什么？谢谢

火花 = 2.4.4 Hadoop = 2.7.3

Answer 1

问题可能来自您使用的是 S3。 在 S3 中，文件系统是基于键/值的，这意味着没有名为 file1.parquet 的物理文件夹，只有键类似于 s3a://bucket/file1.parquet/part-XXXXX-b1e8fd43-ff42-46b4-a74c-9186713c26c6-c000.parquet 的文件（这只是一个示例）。

因此，当您“覆盖”时，您应该覆盖无法检测到的文件夹。因此，spark 会创建新的键：它就像一个“附加”模式。

您可能需要编写自己的函数来覆盖“文件夹” - 删除名称中包含该文件夹的所有键。