将DataFrame写入Parquet或Delta似乎没有并行化-耗时太长

Question

问题陈述

我已将分区的 CSV 文件读入 Spark 数据帧。

为了利用Delta表的改进，我尝试将其作为Delta导出到Azure Data Lake Storage Gen2内的一个目录中。我在Databricks笔记本中使用以下代码：

%scala

df_nyc_taxi.write.partitionBy("year", "month").format("delta").save("/mnt/delta/")

整个数据帧大约有160 GB。

硬件规格

我正在使用具有12个内核和42 GB RAM的集群运行此代码。

然而，看起来整个写作过程都由Spark/Database ricks顺序处理，例如非并行方式：

DAG 可视化效果如下所示：

总而言之，这看起来需要1-2个小时才能执行。

问题

• 有没有办法让Spark并行写入不同的分区？
• 可能是问题在于我尝试将增量表直接写入 Azure 数据湖存储吗？

Answer 1

这与另一个答案类似，但是，我在重新分区后添加了一个持久化

from pyspark.sql import functions as F
df_nyc_taxi.repartition(1250,F.col("year"), col("month"))\
.persist(StorageLevel.MEMORY_AND_DISK).write.partitionBy("year", "month")\
.format("delta").save("/mnt/delta/")

Answer 2

要跟进@eliasah的评论，也许你可以试试这个:

import org.apache.spark.sql.functions
df_nyc_taxi.repartition(col("year"), col("month"), lit(rand() * 200)).write.partitionBy("year", "month").format("delta").save("/mnt/delta/")

@eliasah的回答很有可能是每个目录只创建一个文件“/mnt/delta/year=XX/month=XX”，每个文件只由一个工人写数据。额外的列将进一步分割数据(在这种情况下，我将每个原始文件中的数据分成200个更小的分区，如果愿意，您可以编辑它)，以便更多的工作人员可以并发写入。

附言：对不起，我还没有足够的代表来评论：'D