一、引子
在用Spark SQL编程时，不论是执行SQL语句，还是编写算子提交SparkSubmit 执行，在DataFrame 上的操作大致都会经历以下过程：
执行语句 ——>SQL解析器——> 语法分析器 ——> 优化器 ——> 逻辑执行计划 ——> RDD运算
在关系型数据库中，如Oracle，DB2 大多采用CBO（cost-based optimizer,基于成本的优化规则）数据库引擎会定期收集数据库表的统计信息，在查询时基于统计信息计算不同执行计划的执行成本，选择最优的执行路径。但在Spark 2.0 -2.2 版本中，目前发现大部分的大表Join都会被解析成 sort-merge Join. 而这种Join 方式非常容易在join key 发生数据倾斜时执行失败。
二、思路
通过Join 列的key 值分析（如果数据量较大，可用抽样函数），分析key 值的分布，问题关键在于如何处理异常Key 值
解决Join key倾斜的思路有几种：
1. 这些key存在是否有意义，在业务上是否允许清洗掉这些key?
2. 能否采用分治的思路？一次Join 变成两步Join
3. 加大内存，加大核数能否解决问题？
一般来说，遇到这种问题，从效率角度和降低复杂度来看，应先分析加大资源能否解决问题，进而，数据清洗能否解决，最后才考虑解决倾斜问题。
三、实施
如果采用分治的思路，可考虑改进的Join算法 skew-join
1. 统计key 的分布，过滤出超过阈值的key
2. 对于需要Join 的表,过滤key 做Join计算
3. 广播超阈值的key, 做广播Join
4. 两次join 结果执行union运算
代码实现：
def skewedJoin[T : TypeTag](
a: DataFrame,
b: DataFrame,
joinCol: String,
hubs: Set[T],
logPrefix: String): DataFrame = {
val sqlContext = a.sqlContext
if (hubs.isEmpty) {
// No skew. Do regular join.
a.join(b, joinCol)
} else {
logger.debug(s"$logPrefix Skewed join with ${hubs.size} high-degree keys.")
val isHub = udf { id: T =>
hubs.contains(id)
}
val hashJoined = a.filter(!isHub(col(joinCol)))
.join(b.filter(!isHub(col(joinCol))), joinCol)
val broadcastJoined = a.filter(isHub(col(joinCol)))
.join(broadcast(b.filter(isHub(col(joinCol)))), joinCol)
hashJoined.unionAll(broadcastJoined)
}
}
四、总结
Spark作为新兴技术，还有许多待改进的地方。在实际应用中发现，如果不对倾斜的数据处理，一个10G 的表Join 一个100G的表跑了56个小时，其中一个任务55小时30分钟未结束。
