火花数据帧按平均值和中值分组不完整

Question

我使用Spark sql dataframes执行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

评论中已经有很多合理的建议，但值得一提的是我的想法：

1） 执行df。计算工时？如果不是，您的问题就出现在您发布的代码之前（如注释中所示）

2） 查看Spark UI（如评论中所建议的）-大多数任务完成得很快，而少数任务需要很长时间/看起来卡住了吗？如果是这样，歪斜很可能是你的问题

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

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

5） 绝对值得点燃火花。sql。洗牌分区（如注释所示）；2001是spark中的一个神奇数字（spark.sql.shuffle.partitions的最佳值应该是多少，或者在使用spark sql时我们如何增加分区？）

6）我也会尝试改变数据量，如果你只使用一周中的一天=1，它是否有效（如评论中建议的）

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