大数据学习（124）-spark数据倾斜

🍋🍋大数据学习🍋🍋

🔥系列专栏： 👑哲学语录: 用力所能及，改变世界。
💖如果觉得博主的文章还不错的话，请点赞👍+收藏⭐️+留言📝支持一下博主哦🤞

---

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结合专门处理倾斜键的系统

三、最佳实践建议

1. 预防为主：

   • 在ETL阶段就识别并处理可能的倾斜键
   • 对业务数据分布有充分了解

2. 监控常态化：

   • 建立作业性能基线
   • 监控关键Stage的Shuffle数据量

3. 参数调优：

   properties

   	# 常见相关参数
   	spark.sql.shuffle.partitions=200 # 默认200，根据集群规模调整
   	spark.default.parallelism=200
   	spark.sql.adaptive.coalescePartitions.enabled=true
   	spark.sql.adaptive.coalescePartitions.minPartitionNum=10

4. 测试验证：

   • 在开发环境使用生产规模数据测试
   • 比较不同方案的性能差异