python - Spark 合并与收集，哪个更快？

Question

我正在使用 pyspark 使用 AWS EMR 和 ~15 m4.large 内核处理 50Gb 数据。

数据的每一行都包含一天中特定时间的一些信息。我正在使用以下 for 循环来提取和汇总每小时的信息。最后我union 数据，因为我希望我的结果保存在一个 csv 文件中。

# daily_df is a empty pyspark DataFrame
for hour in range(24):
    hourly_df = df.filter(hourFilter("Time")).groupby("Animal").agg(mean("weights"), sum("is_male"))
    daily_df = daily_df.union(hourly_df)

据我所知，我必须执行以下操作以强制 pyspark.sql.Dataframe 对象保存到 1 个 csv 文件(大约 1Mb)而不是 100 多个文件:

daily_df.coalesce(1).write.csv("some_local.csv")

完成这个进度似乎花了大约 70 分钟，我想知道我是否可以通过使用 collect() 之类的方法使其更快？

daily_df_pandas = daily_df.collect()
daily_df_pandas.to_csv("some_local.csv")

Answer 1

coalesce(1) 和 collect 通常都非常糟糕，但预期输出大小约为 1MB，这并不重要。它根本不应该成为这里的瓶颈。

一个简单的改进是删除 loop -> filter -> union 并执行单个聚合:

df.groupby(hour("Time"), col("Animal")).agg(mean("weights"), sum("is_male"))

如果这还不够，那么这里的问题很可能是配置(如果您还没有调整 spark.sql.shuffle.partitions，最好的起点是调整)。