scala 大数据 hdfs java Spark Join大大表

Spark Join大大表

分而治之拆分内表外表的重复扫描案例

负隅顽抗数据分布均匀数据倾斜Task 数据倾斜Executor 数据倾斜两阶段 ShuffleExecutors 调优案例

Join 大大表 :

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

分而治之

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

分而治之的计算过程：

根据两表的大小区分：外表/ 内表(较小)对内表进行过滤，并把内表划分为多个不重复的完整子集把外表和这些子集做关联，得到部分计算结果最后用 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

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

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

"

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

"

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

//循环遍历 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)

}

负隅顽抗

负隅顽抗 : 当内表没法均匀拆分，或外表没有分区键，不能利用 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

数据倾斜

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)

把倾斜的 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"))

对均匀数据，转为 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)

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

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)))

}

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

//将加盐后的数据分别注册为临时表

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)

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

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

val skewResults: DataFrame =

skewInitialResults.select("initialRevenue", "orderId")

.groupBy(col("orderId"))

.agg(sum(col("initialRevenue")).alias("revenue"))

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

evenResults union skewResults

参考链接

 您阅读本篇文章共花了：