将文件从一个 parquet 分区移动到另一个

Question

我的 S3 存储桶中有大量数据，由 MODULE 和 DATE 两列分区 这样我的镶木地板的文件结构是：

s3://my_bucket/path/file.parquet/MODULE='XYZ'/DATE=2020-01-01

我有 7 个MODULE，DATE 的范围从 2020-01-01 到 2020-09-01。 我发现数据存在差异，需要更正其中一个模块的 MODULE 条目。基本上我需要将属于MODULEXYZ 的特定索引号的所有数据更改为MODULEABC。 我可以在 pyspark 中通过加载数据框并执行以下操作来做到这一点：

df=df.withColumn('MODULE', when(col('index')==34, "ABC").otherwise(col('MODULE')))

但是我该如何重新分区，以便只有那些被更改的条目才能移动到ABC MODULE 分区？如果我这样做：

df.mode('append').partitionBy('MODULE','DATE').parquet(s3://my_bucket/path/file.parquet")

我会将数据与错误的MODULE 数据一起添加。另外，我有将近一年的数据，不想重新分区整个数据集，因为这需要很长时间。

有没有办法做到这一点？

Answer 1

如果我理解得很好，您应该将分区 MODULE=XYZ 中的数据移动到 MODULE=ABC。

首先，确定受影响的文件。

from pyspark.sql import functions as F

file_list = df.where(F.col("index") == 34).select(
    F.input_file_name()
).distinct().collect()

然后，您仅基于这些文件创建一个数据框，并使用它来完成MODULE。

df = spark.read.parquet(file_list).withColumn(
    "MODULE", when(col("index") == 34, "ABC").otherwise(col("MODULE"))
)

df.write.parquet(
    "s3://my_bucket/path/ABC/", mode="append", partitionBy=["MODULE", "DATE"]
)

此时 ABC 应该没问题（你刚刚添加了缺失的数据），但是 XYZ 应该是错误的，因为重复的数据。要恢复XYZ，只需删除file_list中的文件列表即可。

Answer 2

IIUC 您可以通过过滤该特定索引的数据然后将该数据与日期一起保存为分区来做到这一点。

df=df.withColumn('MODULE', when(col('index')==34, "ABC").otherwise(col('MODULE')))
df = df.filter(col('index')==34)
df.mode('overwrite').partitionBy('DATE').parquet(s3://my_bucket/path/ABC/")

通过这种方式，您最终只会修改已更改的模块，即 ABC