spark 数据倾斜解决方案

产生的原因

• 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操作

解决问题思路

解决问题的本质办法：

1. 预聚合，相当于hadoop map 的 Combiner，在map端进行预聚合
2. 打散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 SQL SET 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方式