🍋🍋大数据学习🍋🍋
🔥系列专栏： 👑哲学语录: 用力所能及，改变世界。
💖如果觉得博主的文章还不错的话，请点赞👍+收藏⭐️+留言📝支持一下博主哦🤞
1. 倾斜表现
- 某些Task执行时间远超其他Task（通常超过平均时间10倍以上）
- 集群资源利用率不均（部分Executor负载过高）
- 作业执行时间主要由少数几个Task决定
- 频繁出现GC超时或OOM错误
2. 诊断方法
scala
// 通过Spark UI观察Stage和Task的执行时间 |
|
// 查看Shuffle Read/Write数据量分布 |
|
// 使用以下代码定位倾斜键： |
|
val skewedKeys = rdd.mapPartitions(iter => { |
|
val counts = scala.collection.mutable.HashMap[String, Int]() |
|
iter.foreach(x => counts.put(getKey(x), counts.getOrElse(getKey(x), 0) + 1)) |
|
counts.filter(_._2 > threshold).iterator |
|
}).collect() |
二、核心解决方案
1. 数据预处理策略
(1) 过滤无效倾斜数据
scala
// 过滤掉明显异常的倾斜键（如空值、默认值等） |
|
val filteredData = rawData.filter(row => !isSkewKey(getKey(row))) |
(2) 数据采样分析
scala
// 采样分析数据分布 |
|
val sampledData = rawData.sample(false, 0.1) |
|
val keyDistribution = sampledData.map(row => (getKey(row), 1)) |
|
.reduceByKey(_ + _) |
|
.collect() |
|
.sortBy(-_._2) |
2. Join操作优化
(1) 广播小表（Broadcast Join）
scala
// 当小表小于spark.sql.autoBroadcastJoinThreshold（默认10MB）时自动触发 |
|
// 可手动设置： |
|
val smallDF = spark.table("small_table").cache() |
|
val largeDF = spark.table("large_table") |
|
val result = largeDF.join(broadcast(smallDF), "join_key") |
(2) 盐值加盐法（Salting）
scala
// 对倾斜键添加随机前缀分散数据 |
|
import org.apache.spark.sql.functions._ |
|
// 生成随机盐值（0-99） |
|
val saltedDF = largeDF.withColumn("salted_key", |
|
concat(lit("salt_"), (rand() * 100).cast("int")), |
|
col("key"))) |
|
// 同样处理小表 |
|
val saltedSmallDF = smallDF.withColumn("salted_key", |
|
concat(lit("salt_"), (rand() * 100).cast("int")), |
|
col("key"))) |
|
// 执行Join后去除盐值 |
|
val joined = saltedDF.join(saltedSmallDF, "salted_key") |
|
.drop("salted_key") |
|
.groupBy("key").agg(...) // 可能需要聚合去除重复 |
(3) 倾斜键单独处理
scala
// 分离倾斜键和非倾斜键分别处理 |
|
val (skewKeys, nonSkewKeys) = getSkewKeys(largeDF) // 自定义方法获取倾斜键 |
|
// 处理非倾斜键 |
|
val nonSkewJoin = largeDF.filter(!col("key").isin(skewKeys:_*)) |
|
.join(smallDF, "key") |
|
// 处理倾斜键（可能使用更细粒度分区或特殊处理） |
|
val skewJoin = largeDF.filter(col("key").isin(skewKeys:_*)) |
|
.repartition(100, col("key")) // 增加分区数 |
|
.join(smallDF.repartition(100, col("key")), "key") |
|
// 合并结果 |
|
val result = nonSkewJoin.union(skewJoin) |
3. 聚合操作优化
(1) 两阶段聚合
scala
// 第一阶段：添加随机前缀分散数据 |
|
val firstStage = df.withColumn("prefix", (rand() * 100).cast("int")) |
|
.groupBy("prefix", "key").agg(...) |
|
// 第二阶段：去除前缀聚合 |
|
val secondStage = firstStage.groupBy("key").agg(...) |
(2) 自定义分区器
scala
// 实现自定义分区器，将倾斜键分散到不同分区 |
|
class SkewAwarePartitioner(partitions: Int) extends Partitioner { |
|
override def numPartitions: Int = partitions |
|
override def getPartition(key: Any): Int = { |
|
val strKey = key.toString |
|
if (isSkewKey(strKey)) { |
|
// 对倾斜键进行哈希分散 |
|
math.abs(strKey.hashCode) % partitions |
|
} else { |
|
// 非倾斜键使用默认分区 |
|
math.abs(strKey.hashCode) % (partitions / 10) // 减少非倾斜键分区数 |
|
} |
|
} |
|
} |
|
// 使用自定义分区器 |
|
val partitionedRDD = rdd.partitionBy(new SkewAwarePartitioner(100)) |
4. Shuffle相关优化
(1) 调整并行度
scala
// 增加Shuffle时的并行度 |
|
val repartitionedDF = df.repartition(200, col("skew_key")) |
|
// 或在join时指定 |
|
df1.join(df2, Seq("key"), "inner").repartition(200) |
(2) 使用SkewJoin优化器（Spark 3.0+）
scala
// Spark 3.0+ 自动检测并优化倾斜Join |
|
spark.conf.set("spark.sql.adaptive.enabled", "true") |
|
spark.conf.set("spark.sql.adaptive.skewJoin.enabled", "true") |
|
spark.conf.set("spark.sql.adaptive.skewJoin.skewedPartitionFactor", "5") // 倾斜阈值 |
|
spark.conf.set("spark.sql.adaptive.skewJoin.skewedPartitionThresholdInBytes", "256MB") |
5. 其他高级技术
(1) 增量计算
scala
// 对倾斜键采用增量计算方式，分批处理 |
|
val batchSize = 10000 |
|
val skewKeys = getSkewKeys(df) // 获取所有倾斜键 |
|
val results = skewKeys.grouped(batchSize).flatMap { batch => |
|
val batchDF = df.filter(col("key").isin(batch:_*)) |
|
// 处理当前批次 |
|
processBatch(batchDF) |
|
}.toDF() |
(2) 外部系统辅助
scala
// 对极端倾斜数据，可考虑将数据导出到外部系统（如Redis、HBase）处理 |
|
// 或使用Spark结合专门处理倾斜键的系统 |
三、最佳实践建议
预防为主：
- 在ETL阶段就识别并处理可能的倾斜键
- 对业务数据分布有充分了解
监控常态化：
- 建立作业性能基线
- 监控关键Stage的Shuffle数据量
参数调优：
properties# 常见相关参数spark.sql.shuffle.partitions=200 # 默认200，根据集群规模调整spark.default.parallelism=200spark.sql.adaptive.coalescePartitions.enabled=truespark.sql.adaptive.coalescePartitions.minPartitionNum=10测试验证：
- 在开发环境使用生产规模数据测试
- 比较不同方案的性能差异