Spark Join大大表_大表join大表优化

Join 大大表 :

• Join 的两张体量较大的事实表，尺寸相差在 3 倍内，且无法广播变量
• 用大表 Join 大表才能实现业务逻辑，说明 : 数据仓库在设计初时，考虑不够完善
• 大表 Join 大表的调优思路：分而治之/ 负隅顽抗

分而治之

分而治之的调优思路：把复杂任务拆解成多个简单任务，再合并多个简单任务的计算结果

分而治之的计算过程：

1. 根据两表的大小区分：外表/ 内表(较小)
2. 对内表进行过滤，并把内表划分为多个不重复的完整子集
3. 把外表和这些子集做关联，得到部分计算结果
4. 最后用 Union 把所有结果合并到一起，得到完整的计算结果

拆分内表

内表拆分：要求每个子表的尺寸相对均匀， 且都小到进行广播变量

拆分的关键 : 选取的列，要让子表足够小 :

• 基数低 : 内表的拆分列是性别(男/女)，性别基数是 2。这拆出来的子表还是很大，远远超出广播阈值
• 基数大 : 拆分：身份证号。缺点：不易拆分，开发成本高 ; 过滤条件很难触发优化机制(谓词下推)
• 拆分：时间。一般事实表都与时间相关。既能享受分区剪裁（Partition Pruning），又降低开发成本

外表的重复扫描

外表的重复扫描 ：

• 内表拆分后，外表会和所有子表做关联，但每次关联都要重新扫描外表的全量数据
• 外表扫描的次数 = 内表拆分的份数

解决数据重复扫描：

• Cache，要求资源非常吊
• 用 DPP 机制，对外表进行分区过滤

DPP 机制：

• 每个子查询只扫描外表的子集，把所有子集加起来，就是外表的全量数据

案例

orders 和 transactions 都是事实表，都是 TB 级别 :

//orders 订单表
orderId: Int
customerId: Int
status: String
date: Date //分区键

//lineitems 交易明细表
orderId: Int //分区键
txId: Int
itemId: Int
price: Float
quantity: Int
1
2
3
4
5
6
7
8
9
10
11
12

每隔一段时间 ，计算上个季度所有订单的交易额 :

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
"
1
2
3
4
5
6
7
8
9

transactions 是外表，orders 是内表(较小)

• 对 date 字段进行以天拆分

//以date字段拆分内表
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 = '2020-01-01'
  group by o.orderId
"
1
2
3
4
5
6
7
8
9
10

内表拆分后，外表与所有子表做关联，把全部子关联的结果合并

//循环遍历 dates
val dates: Seq[String] = Seq("2020-01-01", "2020-01-02",..."2020-03-31")

for (date <- dates) {
  val query: String = s"
    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 = ${date}
    group by o.orderId
  " 
  val file: String = s"${outFile}/${date}"
  spark.sql(query).save.parquet(file)
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16

负隅顽抗

负隅顽抗 : 当内表没法均匀拆分，或外表没有分区键，不能利用 DPP，只能依赖 Shuffle Join，来完成 Join 大大表

数据分布均匀

默认 Shuffle Sort Merge Join 转为 Shuffle Hash Join 条件：

• 两表数据分布均匀
• 内表所有数据分片，能放入内存

每个数据分片的切分 :

• 根据并发度/执行内存，计算每个 Task 消耗的内存上下限
• 结合分布式数据集尺寸与上下限，计算出并行度

利用 Join Hints 选择 Shuffle Hash Join

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
1
2
3
4
5
6
7
8
9

数据倾斜

Join 大大表数据倾斜情况 ：

Task 数据倾斜

利用 AQE 解决自动倾斜处理。配置参数 ：

• spark.sql.adaptive.skewJoin.skewedPartitionFactor : 判定倾斜的膨胀系数
• spark.sql.adaptive.skewJoin.skewedPartitionThresholdInBytes : 判定倾斜的最低阈值
• spark.sql.adaptive.advisoryPartitionSizeInBytes : 定义拆分粒度 (字节)

AQE 自动倾斜处理 :

• 外表的倾斜分区，以 spark.sql.adaptive.advisoryPartitionSizeInBytes 把倾斜分区拆分为多个数据分区
• 对内表的数据分区进行复制

Task 的负载均衡 :

• AQE 只能处理 Task 倾斜，Executors 的负载倾斜并没有改善

Executor 数据倾斜

解决 Executors 的数据倾斜的方法 ：分而治之/ 两阶段 Shuffle

分而治之 :

• 对外表中所有 Join Keys，按照倾斜分为两组 (倾斜的 Join Keys / 分布均匀的 Join Keys)
• 按照两组 Join Keys，对内表分为两份
• 对内外表的两组数据，分别用不同方法做关联计算
• 分布均匀的数据，把 Shuffle Sort Merge Join 转为 Shuffle Hash Join
• 倾斜数据，用两阶段 Shuffle，平衡 Executors 之间的工作负载
• 再把两个关联结果集进行 Union

两阶段 Shuffle

两阶段 Shuffle：

• 第一阶段 ：目的将数据打散、平衡计算负载，通过 加盐、Shuffle、关联、聚合
• 第二阶段 ：计算结果，通过 去盐化、Shuffle、聚合
• 不破坏原有关联关系下，平衡 Executors 之间的计算负载

第一阶段：对倾斜 Join Keys 加盐 (粒度 : Executors 总数)

• 外表/内表做不同加盐处理

对外表进行随机加盐 :

• 对倾斜的 Join Key，都 + 随机后缀 (1 - #N)

内表进行复制加盐 ：

• 对倾斜的 Join Key，把原数据复制 (Executors 总数 – 1)，得到 (Executors 总数) 份数据副本
• 对每份副本，按 Join Key +固定后缀 (1 - #N) ，与打散后的外表数据保持一致

第二阶段 :

• 把每个 Join Key 的后缀去掉 (去盐化)
• 按照原来的 Join Key 进行 Shuffle 和聚合计算，得出倾斜组的计算结果
• 将倾斜的结果和均匀的结果进行合并

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)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22

把倾斜的 orderId 保存在数组 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"))
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17

对均匀数据，转为 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)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15

对外表做随机加盐，对内表做复制加盐

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)))
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18

对加盐的两张表，进行查询 ：

//将加盐后的数据分别注册为临时表
saltedSkewTx.createOrReplaceTempView(“saltedSkewTx”)
saltedskewOrders.createOrReplaceTempView(“saltedskewOrders”)

val skewQuery: String = "
  select /*+ shuffle_hash(orders) */ 
		sum(tx.price * tx.quantity) as initialReven,
  	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)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18

去盐化目的 ：把计算的粒度，从加盐 joinKey 恢复为原来的 orderId

• 只要在 orderId 上进行聚合，就能去盐化

val skewResults: DataFrame = 
  skewInitialResults.select("initialRevenue", "orderId")
  .groupBy(col("orderId"))
  .agg(sum(col("initialRevenue")).alias("revenue"))
1
2
3
4

把倾斜结果和均匀结果进行合并，就能平衡 Executors 计算负载

evenResults union skewResults
1