Spark最终任务比前199任务长100倍,如何改进

Question

我在使用数据帧运行查询时看到了一些性能问题.我在研究中已经看到,长期运行最终任务可以表明数据没有被最佳干扰,但是没有找到解决此问题的详细过程.

我开始将两个表作为数据帧加载,然后我在一个字段上加入这些表.我试图通过(重新分区)添加分发,并按顺序排序,以提高性能,但我仍然看到这个单一的长期运行的最终任务.这是我的代码的简单版本,请注意查询一和二实际上并不简单,并使用UDF来计算某些值.

我尝试过几种不同的设置spark.sql.shuffle.我已经尝试了100,但它失败了(说实话我没有真正调试这个).我尝试了300,4000和8000.性能随着每次增加而下降.我选择一天的数据,每个文件是一小时.

val df1 = sqlContext.sql("Select * from Table1")
val df2 = sqlContext.sql("Select * from Table2")

val distributeDf1 = df1
    .repartition(df1("userId"))
    .sortWithinPartitions(df1("userId"))

val distributeDf2 = df2
    .repartition(df2("userId"))
    .sortWithinPartitions(df2("userId"))

distributeDf1.registerTempTable("df1")
distributeDf2.registerTempTable("df2")

val df3 = sqlContext
  .sql("""
    Select 
      df1.* 
    from 
      df1 
    left outer join df2 on 
      df1.userId = df2.userId""")

由于看起来userId的分区并不理想,我可以通过时间戳进行分区.如果我这样做,我应该只做日期+小时吗？如果我有不到200个独特的组合,我会有空执行器吗？

Answer 1

你明显有一个巨大的权利数据偏差的问题.让我们来看看你提供的统计数据:

df1 = [mean=4.989209978967438, stddev=2255.654165352454, count=2400088] 
df2 = [mean=1.0, stddev=0.0, count=18408194]

平均值大约为5,标准差超过2000,你得到一个长尾巴.

由于在重新分区之后某些键比其他键更频繁,因此一些执行器将比剩余的键有更多的工作要做.

此外,您的描述表明问题可能是单个或几个键散列到同一个分区.

那么,让我们首先确定异常值(伪代码):

val mean = 4.989209978967438 
val sd = 2255.654165352454

val df1 = sqlContext.sql("Select * from Table1")
val counts = df.groupBy("userId").count.cache

val frequent = counts
  .where($"count" > mean + 2 * sd)  // Adjust threshold based on actual dist.
  .alias("frequent")
  .join(df1, Seq("userId"))

其余的:

val infrequent = counts
  .where($"count" <= mean + 2 * sd)
  .alias("infrequent")
  .join(df1, Seq("userId"))

这真的是值得期待的吗？如果没有,请尝试确定上游问题的来源.

如果是预期的,您可以尝试:

• 广播小表:

  val df2 = sqlContext.sql("Select * from Table2")
  df2.join(broadcast(df1), Seq("userId"), "rightouter")
  

• 分裂,统一(union)和频繁播放:

  df2.join(broadcast(frequent), Seq("userId"), "rightouter")
    .union(df2.join(infrequent, Seq("userId"), "rightouter"))
  

• userId用一些随机数据腌制

但你不应该:

• 重新分配所有数据并在本地排序(尽管单独在本地排序不应该是一个问题)
• 对完整数据执行标准散列连接.