Spark dataframe groupby均值和中位数未完成

Question

我正在使用 Spark sql 数据帧执行 groupby 操作，然后计算每个组的数据的平均值和中位数。原始数据量约为 1 TB。

val df_result = df.filter($"DayOfWeek" <= 5).groupBy("id").agg(
        count("Error").as("Count"), 
        avg("Error").as("MeanError"), 
        callUDF("percentile_approx", col("Error"), lit(0.05)).as("5thError"), 
        callUDF("percentile_approx", col("Error"), lit(0.5)).as("MedianError"), 
        callUDF("percentile_approx", col("Error"), lit(0.95)).as("95thError")).
    filter($"Count" > 1000)


df_result.orderBy(asc("MeanError")).limit(5000)
    .write.format("csv").option("header", "true").save("/user/foo.bar/result.csv")

当我运行该查询时，我的工作被卡住并且没有完成。我该如何调试问题？是否存在导致groupby() 卡住的关键不平衡？

Answer 1

在 cmets 中已经有很多明智的建议，但我的想法值得在这里：

1) df.count 有效吗？如果没有，您的问题出在您发布的代码之前（如 cmets 中所建议）

2) 查看 Spark UI（如 cmets 中所建议的） - 大多数任务是否快速完成，而少数任务需要很长时间/似乎卡住了？如果是这样，偏斜可能是您的问题

3) 您可能会重写您的查询，首先只找到每个“id”的“计数”。接下来过滤您的原始 df 以仅包含通过广播（以避免 df 的洗牌）内部连接（如果没有太多 1000 次以上的 id）出现超过 1000 次的行。然后聚合这个较小的数据框并计算所有统计数据。如果计数聚合有效，则输出还应显示是否存在任何明显的数据倾斜！

4) 有时将计算分解为更小的步骤，然后写入然后立即从磁盘读取，这帮助我过去完成了一些尴尬的工作。如果一开始生成 df 的成本很高，也可以加快调试速度。

5) 绝对值得升级 spark.sql.shuffle.partitions（如 cmets 中所建议的）； 2001 是 spark 中的一个神奇数字 (What should be the optimal value for spark.sql.shuffle.partitions or how do we increase partitions when using Spark SQL?)

6）我也会尝试改变数据量，如果你只使用星期几 = 1（如 cmets 中的建议），它是否有效

7) 查询是否在没有 percentile_approx 的情况下运行？