产生的原因
- shuffle操作之后，1个key有80万数据，其它key8万，这就会导致某个reducetask上被分配了88万数据执行，两外两个task完成之后等待这个task完成
- 在业务层面，产生的原因一般是网站被刷
造成影响
该作业执行非常慢，或者直接OOM
定位问题
- 观察spark ui 发现大部分task都执行非常快，刷刷刷，剩下几个task，执行的特别特别慢，前面的task，一般1s可以执行完5个；最后发现1000个task，998，999 task，要执行1个小时，2个小时才能执行完一个task刷刷刷，突然OOM了
- 找代码哪些地方有shuffle操作
解决问题思路
解决问题的本质办法：
- 预聚合，相当于hadoop map 的 Combiner，在map端进行预聚合
- 打散key，二次聚合
1、过滤异常数据
countByKey然后对这些 key 对应的记录进行分析:
- 空值或者异常值之类的，大多是这个原因引起（网站被刷，生产环境经常遇到）
在hue上写spark sql，执行left join 操作，大量空值会产生数据倾斜，改为union ，优化sql - 无效数据，大量重复的测试数据或是对结果影响不大的有效数据
- 有效数据，业务导致的正常数据分布
正对以上前两种情况，直接过滤掉，第三种情况业务数据分布本身就倾斜，怎么办？
2、业务导致的正常数据分布倾斜
提高 shuffle 并行度
- Spark SQL，还可通过
SET spark.sql.shuffle.partitions=[num_tasks]设置并行度 - RDD 操作 可在需要 Shuffle 的操作算子上直接设置并行度或者使用 spark.default.parallelism 设置。如果是
解决：大量不同的 Key 被分配到了相同的 Task 造成该 Task 数据量过大。
自定义 Partitioner
.groupByKey(new Partitioner() {
@Override
public int numPartitions() {
return 12;
}
@Override
public int getPartition(Object key) {
int id = Integer.parseInt(key.toString());
if(id >= 9500000 && id <= 9500084 && ((id - 9500000) % 12) == 0) {
return (id - 9500000) / 12;
} else {
return id % 12;
}
}
})
解决：使用自定义的 Partitioner 实现类代替默认的 HashPartitioner，尽量将所有不同的 Key 均匀分配到不同的 Task 中。
对源数据进行预聚合操作
- spark sql
执行，优化key（一大一小表）
某个key对应的80万数据，某些key对应几百条，某些key对应几十条；现在，咱们直接在生成hive表的hive etl中，对数据进行聚合。比如按key来分组，将key对应的所有的values，全部用一种特殊的格式，拼接到一个字符串里面去，比如“key=sessionid, value: action_seq=1|user_id=1|search_keyword=火锅|category_id=001;action_seq=2|user_id=1|search_keyword=涮肉|category_id=001”。
对key进行group，在spark中，拿到key=sessionid，values<Iterable>；hive etl中，直接对key进行了聚合。那么也就意味着，每个key就只对应一条数据。在spark中，就不需要再去执行groupByKey+map这种操作了。直接对每个key对应的values字符串，map操作，进行你需要的操作即可。key,values串。
Spark SQLSET spark.sql.autoBroadcastJoinThreshold=10485760 (10m)可以设置为20m - rdd 执行（一大一小表）
使用广播变量，进行map端join，小表join大表转为小表broadcast+map大表实现,例如几百MB或者1~2GB
拆分 join 再 union 两大表join
- spark sql
拆分sql优化为union all 方式 - rdd
想办法转为大小表，过滤掉不需要的数据，然后再使用broadcast+map方式
