29 大表Join大表(二):什么是负隅顽抗的调优思路?

你好,我是吴磊。

在上一讲,我们说了应对“大表Join大表”的第一种调优思路是分而治之,也就是把一个庞大而又复杂的Shuffle Join转化为多个轻量的Broadcast Joins。这一讲,我们接着来讲第二种调优思路:负隅顽抗。

负隅顽抗指的是,当内表没法做到均匀拆分,或是外表压根就没有分区键,不能利用DPP,只能依赖Shuffle Join,去完成大表与大表的情况下,我们可以采用的调优方法和手段。这类方法比较庞杂,适用的场景各不相同。从数据分布的角度出发,我们可以把它们分两种常见的情况来讨论,分别是数据分布均匀和数据倾斜。

我们先来说说,在数据分布均匀的情况下,如何应对“大表Join大表”的计算场景。

数据分布均匀

在第27讲的最后,我们说过,当参与关联的大表与小表满足如下条件的时候,Shuffle Hash Join的执行效率,往往比Spark SQL默认的Shuffle Sort Merge Join更好。

实际上,这个调优技巧同样适用于“大表Join大表”的场景,原因其实很简单,这两个条件与数据表本身的尺寸无关,只与其是否分布均匀有关。不过,为了确保Shuffle Hash Join计算的稳定性,我们需要特别注意上面列出的第二个条件,也就是内表所有的数据分片都能够放入内存。

那么问题来了,我们怎么确保第二个条件得以成立呢?其实,只要处理好并行度、并发度与执行内存之间的关系,我们就可以让内表的每一个数据分片都恰好放入执行内存中。简单来说,就是先根据并发度与执行内存,计算出可供每个Task消耗的内存上下限,然后结合分布式数据集尺寸与上下限,倒推出与之匹配的并行度。更详细的内容你可以去看看第14讲

那我们该如何强制Spark SQL在运行时选择Shuffle Hash Join机制呢?答案就是利用Join Hints。这个技巧我们讲过很多次了,所以这里,我直接以上一讲中的查询为例,把它的使用方法写在了下面,方便你复习。

//查询语句中使用Join hints
select /*+ shuffle_hash(orders) */ sum(tx.price * tx.quantity) as revenue, o.orderId
from transactions as tx inner join orders as o
on tx.orderId = o.orderId
where o.status = ‘COMPLETE’
and o.date between ‘2020-01-01’ and ‘2020-03-31’
group by o.orderId

数据倾斜

接下来,我们再说说,当参与Join的两张表存在数据倾斜问题的时候,我们该如何应对“大表Join大表”的计算场景。对于“大表Join大表”的数据倾斜问题,根据倾斜位置的不同,我们可以分为3种情况来讨论。

其实,不管哪种表倾斜,它们的调优技巧都是类似的。因此,我们就以第一种情况为例,也就是外表倾斜、内表分布均匀的情况,去探讨数据倾斜的应对方法。

以Task为粒度解决数据倾斜

学过AQE之后,要应对数据倾斜,想必你很快就会想到AQE的特性:自动倾斜处理。给定如下配置项参数,Spark SQL在运行时可以将策略OptimizeSkewedJoin插入到物理计划中,自动完成Join过程中对于数据倾斜的处理。

Join过程中的自动倾斜处理如上图所示,当AQE检测到外表存在倾斜分区之后,它会以spark.sql.adaptive.advisoryPartitionSizeInBytes配置的数值为拆分粒度,把倾斜分区拆分为多个数据分区。与此同时,AQE还需要对内表中对应的数据分区进行复制,来保护两表之间的关联关系。

有了AQE的自动倾斜处理特性,在应对数据倾斜问题的时候,我们确实能够大幅节省开发成本。不过,天下没有免费的午餐,AQE的倾斜处理是以Task为粒度的,这意味着原本Executors之间的负载倾斜并没有得到根本改善。这到底是什么意思呢?

我们来举个例子,假设某张表在Shuffle过后有两个倾斜分区如上图,它们又刚好都被Shuffle到了同一个执行器:Executor 0。在AQE的自动倾斜处理机制下,两个倾斜分区分别被拆分变成了4个尺寸适中的数据分区。如此一来,Executor 0中所有Task的计算负载都得到了平衡。但是,相比Executor 1,Executor 0整体的计算负载还是那么多,并没有因为AQE的自动处理而得到任何缓解。

以Executor为粒度解决数据倾斜

你也许会说:“哪会那么凑巧,倾斜的分区刚好全都落在同一个Executor上?”确实,刚才的例子主要是为了帮你解释清楚倾斜粒度这个概念,如果实际应用中倾斜分区在集群中的分布比较平均的话,AQE的自动倾斜处理机制确实就是开发者的“灵丹妙药”。

然而,凡事总有个万一,我们在探讨调优方案的时候,还是要考虑周全:如果你的场景就和咱们的例子一样,倾斜分区刚好落在集群中少数的Executors上,你该怎么办呢?答案是:“分而治之”和“两阶段Shuffle”。

这里的分而治之与上一讲的分而治之在思想上是一致的,都是以任务分解的方式来解决复杂问题。区别在于我们今天要讲的,是以Join Key是否倾斜为依据来拆解子任务。具体来说,对于外表中所有的Join Keys,我们先按照是否存在倾斜把它们分为两组。一组是存在倾斜问题的Join Keys,另一组是分布均匀的Join Keys。因为给定两组不同的Join Keys,相应地我们把内表的数据也分为两份。

那么,分而治之的含义就是,对于内外表中两组不同的数据,我们分别采用不同的方法做关联计算,然后通过Union操作,再把两个关联计算的结果集做合并,最终得到“大表Join大表”的计算结果,整个过程如上图所示。

对于Join Keys分布均匀的数据部分,我们可以沿用把Shuffle Sort Merge Join转化为Shuffle Hash Join的方法。对于Join Keys存在倾斜问题的数据部分,我们就需要借助“两阶段Shuffle”的调优技巧,来平衡Executors之间的工作负载。那么,什么是“两阶段Shuffle”呢?

如何理解“两阶段Shuffle”?

用一句话来概括,“两阶段Shuffle”指的是,通过“加盐、Shuffle、关联、聚合”与“去盐化、Shuffle、聚合”这两个阶段的计算过程,在不破坏原有关联关系的前提下,在集群范围内以Executors为粒度平衡计算负载 。

我们先来说说第一阶段,也就是“加盐、Shuffle、关联、聚合”的计算过程。显然,这个阶段的计算分为4个步骤,其中最为关键的就是第一步的加盐。加盐来源于单词Salting,听上去挺玄乎,实际上就是给倾斜的Join Keys添加后缀。加盐的核心作用就是把原本集中倾斜的Join Keys打散,在进行Shuffle操作的时候,让原本应该分发到某一个Executor的倾斜数据,均摊到集群中的多个Executors上,从而以这种方式来消除倾斜、平衡Executors之间的计算负载。

对于加盐操作,我们首先需要确定加盐的粒度,来控制数据打散的程度,粒度越高,加盐后的数据越分散。由于加盐的初衷是以Executors为粒度平衡计算负载,因此通常来说,取Executors总数#N作为加盐粒度往往是一种不错的选择。其次,为了保持内外表的关联关系不被破坏,外表和内表需要同时做加盐处理,但处理方法稍有不同。

外表的处理称作“随机加盐”,具体的操作方法是,对于任意一个倾斜的Join Key,我们都给它加上1到#N之间的一个随机后缀。以Join Key = ‘黄小乙’来举例,假设N = 5,那么外表加盐之后,原先Join Key = ‘黄小乙’的所有数据记录,就都被打散成了Join Key为(‘黄小乙_1’,‘黄小乙_2’,‘黄小乙_3’,‘黄小乙_4’,‘黄小乙_5’)的数据记录。

内表的处理称为“复制加盐”,具体的操作方法是,对于任意一个倾斜的Join Key,我们都把原数据复制(#N – 1)份,从而得到#N份数据副本。对于每一份副本,我们为其Join Key追加1到#N之间的固定后缀,让它与打散后的外表数据保持一致。对于刚刚Join Key = ‘黄小乙’的例子来说,在内表中,我们需要把‘黄小乙’的数据复制4份,然后依次为每份数据的Join Key追加1到5的固定后缀,如下图所示。

内外表分别加盐之后,数据倾斜问题就被消除了。这个时候,我们就可以使用常规优化方法,比如,将Shuffle Sort Merge Join转化为Shuffle Hash Join,去继续执行Shuffle、关联和聚合操作。到此为止,“两阶段Shuffle” 的第一阶段执行完毕,我们得到了初步的聚合结果,这些结果是以打散的Join Keys为粒度进行计算得到的。

我们刚刚说,第一阶段加盐的目的在于将数据打散、平衡计算负载。现在我们已经得到了数据打散之后初步的聚合结果,离最终的计算结果仅有一步之遥。不过,为了还原最初的计算逻辑,我们还需要把之前加上的“盐粒”再去掉。

第二阶段的计算包含“去盐化、Shuffle、聚合”这3个步骤。首先,我们把每一个Join Key的后缀去掉,这一步叫做“去盐化”。然后,我们按照原来的Join Key再做一遍Shuffle和聚合计算,这一步计算得到的结果,就是“分而治之”当中倾斜部分的计算结果。

经过“两阶段Shuffle”的计算优化,我们终于得到了倾斜部分的关联结果。将这部分结果与“分而治之”当中均匀部分的计算结果合并,我们就能完成存在倾斜问题的“大表Join大表”的计算场景。

以Executors为粒度的调优实战

应该说,以Executors为粒度平衡计算负载的优化过程,是我们学习过的调优技巧中最复杂的。因此,咱们有必要结合实际的应用案例,来详细讲解具体的实现方法。为了方便你对不同的调优方法做对比,我们不妨以上一讲跨境电商的场景为例来讲。

咱们先来回顾一下这家电商的业务需求,给定orders和transactions两张体量都在TB级别的事实表,每隔一段时间就计算一次上一个季度所有订单的交易额,具体的业务代码如下所示。

//统计订单交易额的代码实现
val txFile: String = _
val orderFile: String = _
 
val transactions: DataFrame = spark.read.parquent(txFile)
val orders: DataFrame = spark.read.parquent(orderFile)
 
transactions.createOrReplaceTempView(“transactions”)
orders.createOrReplaceTempView(“orders”)
 
val query: String = “
select sum(tx.price * tx.quantity) as revenue, o.orderId
from transactions as tx inner join orders as o
on tx.orderId = o.orderId
where o.status = ‘COMPLETE’
and o.date between ‘2020-01-01’ and ‘2020-03-31’
group by o.orderId
”
 
val outFile: String = _
spark.sql(query).save.parquet(outFile)

对于这样一个查询语句,我们该如何实现刚刚说过的优化过程呢?首先,我们先遵循“分而治之”的思想,把内外表的数据分为两个部分。第一部分包含所有存在倾斜问题的Join Keys及其对应的Payloads,第二部分保留的是分布均匀的Join Keys和相应的Payloads。假设我们把所有倾斜的orderId,也就是Join Key保存在数组skewOrderIds中,而把分布均匀的orderId保持在数组evenOrderIds中,我们就可以使用这两个数组,把内外表各自拆分为两部分。

//根据Join Keys是否倾斜、将内外表分别拆分为两部分
import org.apache.spark.sql.functions.array_contains
 
//将Join Keys分为两组,存在倾斜的、和分布均匀的
val skewOrderIds: Array[Int] = _
val evenOrderIds: Array[Int] = _
 
val skewTx: DataFrame = transactions.filter(array_contains(lit(skewOrderIds),$"orderId"))
val evenTx: DataFrame = transactions.filter(array_contains(lit(evenOrderIds),$"orderId"))
 
val skewOrders: DataFrame = orders.filter(array_contains(lit(skewOrderIds),$"orderId"))
val evenOrders: DataFrame = orders.filter(array_contains(lit(evenOrderIds),$"orderId"))

拆分完成之后,我们就可以延续“分而治之”的思想,分别对这两部分应用不同的调优技巧。对于分布均匀的部分,我们把Shuffle Sort Merge Join转化为Shuffle Hash Join。

//将分布均匀的数据分别注册为临时表
evenTx.createOrReplaceTempView(“evenTx”)
evenOrders.createOrReplaceTempView(“evenOrders”)
 
val evenQuery: String = “
select /*+ shuffle_hash(orders) */ sum(tx.price * tx.quantity) as revenue, o.orderId
from evenTx as tx inner join evenOrders as o
on tx.orderId = o.orderId
where o.status = ‘COMPLETE’
and o.date between ‘2020-01-01’ and ‘2020-03-31’
group by o.orderId
”
val evenResults: DataFrame = spark.sql(evenQuery)

对于存在倾斜的部分,我们要祭出“两阶段Shuffle”的杀手锏。首先,在第一阶段,我们需要给两张表分别加盐,对外表(交易表)做“随机加盐”,对内表(订单表)做“复制加盐”。

import org.apache.spark.sql.functions.udf
 
//定义获取随机盐粒的UDF
val numExecutors: Int = _
val rand = () => scala.util.Random.nextInt(numExecutors)
val randUdf = udf(rand)
 
//第一阶段的加盐操作。注意:保留orderId字段,用于后期第二阶段的去盐化
 
//外表随机加盐
val saltedSkewTx = skewTx.withColumn(“joinKey”, concat($“orderId”, lit(“_”), randUdf()))
 
//内表复制加盐
var saltedskewOrders = skewOrders.withColumn(“joinKey”, concat($“orderId”, lit(“_”), lit(1)))
for (i <- 2 to numExecutors) {
saltedskewOrders = saltedskewOrders union skewOrders.withColumn(“joinKey”, concat($“orderId”, lit(“_”), lit(i)))
}

两张表分别做完加盐处理之后,我们就可以使用与之前类似的查询语句,对它们执行后续的Shuffle、关联与聚合等操作。

//将加盐后的数据分别注册为临时表
saltedSkewTx.createOrReplaceTempView(“saltedSkewTx”)
saltedskewOrders.createOrReplaceTempView(“saltedskewOrders”)
 
val skewQuery: String = “
select /*+ shuffle_hash(orders) */ sum(tx.price * tx.quantity) as initialRevenue, o.orderId, o.joinKey
from saltedSkewTx as tx inner join saltedskewOrders as o
on tx.joinKey = o.joinKey
where o.status = ‘COMPLETE’
and o.date between ‘2020-01-01’ and ‘2020-03-31’
group by o.joinKey
”
//第一阶段加盐、Shuffle、关联、聚合后的初步结果
val skewInitialResults: DataFrame = spark.sql(skewQuery)

得到第一阶段的初步结果之后,我们就可以开始执行第二阶段的计算了,也就是“去盐化、Shuffle与聚合”这三个操作。去盐化的目的实际上就是把计算的粒度,从加盐的joinKey恢复为原来的orderId。由于在最初加盐的时候,我们对orderId字段进行了保留,因此在第二阶段的计算中,我们只要在orderId字段之上执行聚合操作,就能达到我们想要的“去盐化”效果。

val skewResults: DataFrame = skewInitialResults.select(“initialRevenue”, “orderId”)
.groupBy(col(“orderId”)).agg(sum(col(“initialRevenue”)).alias(“revenue”))

在完成了第二阶段的计算之后,我们拿到了“两阶段Shuffle”的计算结果。最终,只需要把这份结果与先前均匀部分的关联结果进行合并,我们就能实现以Executors为粒度平衡计算负载的优化过程。

evenResults union skewResults

执行性能与开发成本的博弈

你可能会说:“我的天呐!为了优化这个场景的计算,这得花多大的开发成本啊!又是分而治之,又是两阶段Shuffle的,这么大的开发投入真的值得吗?”

这个问题非常好。我们要明确的是,分而治之外加两阶段Shuffle的调优技巧的初衷,是为了解决AQE无法以Executors为粒度平衡计算负载的问题。因此,这项技巧取舍的关键就在于,Executors之间的负载倾斜是否构成整个关联计算的性能瓶颈。如果这个问题的答案是肯定的,我们的投入就是值得的。

小结

今天这一讲,你需要掌握以Shuffle Join的方式去应对“大表Join大表”的计算场景。数据分布不同,应对方法也不尽相同。

当参与Join的两张表数据分布比较均匀,而且内表的数据分片能够完全放入内存,Shuffle Hash Join的计算效率往往高于Shuffle Sort Merge Join,后者是Spark SQL默认的关联机制。你可以使用关键字“shuffle_hash”的Join Hints,强制Spark SQL在运行时选择Shuffle Hash Join实现机制。对于内表数据分片不能放入内存的情况,你可以结合“三足鼎立”的调优技巧,调整并行度、并发度与执行内存这三类参数,来满足这一前提条件。

当参与Join的两张表存在数据倾斜时,如果倾斜的情况在集群内的Executors之间较为均衡,那么最佳的处理方法就是,利用AQE提供的自动倾斜处理机制。你只需要设置好以下三个参数,剩下的事情交给AQE就好了。

但是,如果倾斜问题仅集中在少数的几个Executors中,而且这些负载过高的Executors已然成为性能瓶颈,我们就需要采用“分而治之”外加“两阶段Shuffle”的调优技巧去应对。“分而治之”指的是根据Join Keys的倾斜与否,将内外表的数据分为两部分分别处理。其中,均匀的部分可以使用Shuffle Hash Join来完成计算,倾斜的部分需要用“两阶段Shuffle”进行处理。

两阶段Shuffle的关键在于加盐和去盐化。加盐的目的是打散数据分布、平衡Executors之间的计算负载,从而消除Executors单点瓶颈。去盐化的目的是还原原始的关联逻辑。尽管两阶段Shuffle的开发成本较高,但只要获得的性能收益足够显著,我们的投入就是值得的。

每日一练

  1. 当尝试将Join Keys是否倾斜作为“分而治之”的划分依据时,你觉得我们该依据什么标准把Join Keys划分为倾斜组和非倾斜组呢?
  2. 无论是AQE的自动倾斜处理,还是开发者的“两阶段Shuffle”,本质上都是通过“加盐”与“去盐化”的两步走,在维持关联关系的同时平衡不同粒度下的计算负载。那么,这种“加盐”与“去盐化”的优化技巧,是否适用于所有的关联场景?如果不是,都有哪些场景没办法利用AQE的自动倾斜处理,或是我们的“两阶段Shuffle”呢?

期待在留言区看到你的思考和答案,我们下一讲见!