repartition导致的广播失败，关于错误Failed to get broadcast_544_piece0 of broadcast_544

提示：扫一扫查出行
【扫一扫了解最新限行尾号】
复制提示

今天一个生产环境任务出现了性能问题，，经过仔细检查发现是在一个join操作时，原设定广播右表数据广播失败，导致后续步骤进行缓慢，，报错信息

java.io.IOException: org.apache.spark.SparkException：Failed to get broadcast_544_piece0 of broadcast_544

源代码大概是这个样子（变量全部用xx、yy代替了，不影响整个结构）

val Site = draftedSite.join(broadcast(toSite), Seq("joinCon"))
.withColumn("xxx", distanceUDF($"yy", $"yy", $"yy", $"yy"))
.withColumn("xxx", defineSiteDistanceUDF($"yy", $"yy", $"yy", $"yy"))
.filter("xx> 0 and xx< yy")
.withColumn("deleteSite", expr(
"""
|case
|when xx!= xx then if (xx< xx, xx, xx)
|when xx!= xx then if(xx< xx, xx, xx)
|else if(xx> xx, xx, xx)
|end
""".stripMargin)).repartition(xx).cache()

一开始查询网上，大致都是一种说法，类似https://issues.apache.org/jira/browse/SPARK-5594中的sparkContect中的残留信息数据导致不成功，这明显不是我这个问题，我每次都是新起动一个sparkContect的。

后来公司的大神看了这段代码之后，指出可能是repartition导致的广播失败，去掉repartition（xx），之后任务成功执行。

在key值不够的情况下，强制repartition可能会导致生成一部分空分区，空分区导致了广播的失败。

另外在数据量不定的情况下不建议使用强制广播，建议将tosite注册为临时表之后cache，有spark根据数据量自动判断是否广播

最终修改之后结果如下：

toSite.createOrReplaceTempView("temp")
spark.catalog.cacheTable("temp")
val temp= spark.sql("select * from temp")val Site = draftedSite.join(toSite, Seq("joinCon"))
.withColumn("xxx", distanceUDF($"yy", $"yy", $"yy", $"yy"))
.withColumn("xxx", defineSiteDistanceUDF($"yy", $"yy", $"yy", $"yy"))
.filter("xx> 0 and xx< yy")
.withColumn("deleteSite", expr(
"""
|case
|when xx!= xx then if (xx< xx, xx, xx)
|when xx!= xx then if(xx< xx, xx, xx)
|else if(xx> xx, xx, xx)
|end
""".stripMargin)).cache()