python之如何在pyspark中并行写入多个parquet文件

我想将数据帧拆分为两个数据帧并将它们写入两个单独的 Parquet 文件，如下所示

df = attachment_df.flatMap(process_attachment).toDF() 
 
large_df = df.filter(df.is_large_file == True) 
small_df = df.filter(df.is_large_file == False) 
 
(large_df.write 
    .mode("overwrite") 
    .parquet('large_dummy')) 
 
(small_df.write 
    .mode("overwrite") 
    .parquet('small_dummy')) 

但是，上面的代码将按顺序编写，看起来每个附件都会调用 process_attachment 函数两次。我真的很想避免重复计算，因为处理附件非常昂贵。

有没有办法避免重复处理附件并并行写入？我不想使用按 is_large_file 列的分区写入单个 Parquet 文件。

谢谢，

请您参考如下方法:

当 spark 写入时，它会为每个数据帧并行写入(基于分区数)。因此，您基本上是在连续执行两个并行任务(即应该不会产生太大影响)。 主要问题是目前您正在重新计算 df 两次。

这样做的原因是 DAG 是针对每个 Action 单独计算的(写是一个 Action )。

如果你有足够的内存，你可以简单地通过在第一次写入之前执行 df.cache() 和在第二次写入之后执行 df.unpersist 来改进这一点。这将在可能的时候(即有足够的内存)将 df 的计算保存在内存中。

如果您没有足够的内存并且 process_attachment 真的很长，您可以尝试使用 persist 和 MEMORY_AND_DISK 选项，如果它太大，计算将溢出到磁盘(即您将从磁盘重新加载而不是重新计算)。