将文件从一个实木复合地板分区移动到另一个

如何解决将文件从一个实木复合地板分区移动到另一个

我的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条目。基本上，我需要将属于MODULE XYZ的特定索引号的所有数据更改为MODULE ABC。 我可以在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数据一起添加。另外，我有将近一年的数据，并且不想对整个数据集进行分区，因为这将花费很长时间。

有没有办法做到这一点？

解决方法

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

,

据我所知，您在分区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中的文件列表。