如何在pyspark中并行编写多个镶木地板文件？

Question

我想将数据帧拆分为两个数据帧，并将它们写入两个单独的 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 列的分区写入单个镶木地板文件。

谢谢，

Answer 1

spark 写入时，它为每个数据帧并行写入（基于分区数）。所以你基本上是在做两个串行的并行任务（即那不应该有太大的影响）。 主要问题是目前您正在重新计算 df 两次。

这样做的原因是DAG是为每个动作单独计算的（写作是一个动作）。

如果你有足够的内存，你可以通过在第一次写入之前执行 df.cache() 和在第二次写入之后执行 df.unpersist 来改善这一点。这将在可能的情况下将 df 的计算保留在内存中（即有足够的内存）。

如果您没有足够的内存并且 process_attachment 真的很长，您可以尝试使用带有 MEMORY_AND_DISK 选项的持久化，如果它太大，则会将计算溢出到磁盘（即您将从磁盘重新加载而不是重新计算）。