如何处理生产环境中的数据倾斜问题 ？

分析&回答

1、flink数据倾斜的表现：

任务节点频繁出现反压，增加并行度也不能解决问题

部分节点出现OOM异常，是因为大量的数据集中在某个节点上，导致该节点内存被爆，任务失败重启

2、数据倾斜产生的原因：

业务上有严重的数据热点，比如滴滴打车的订单数据中北京、上海等几个城市的订单量远远超过其他地区；

技术上大量使用了 KeyBy、GroupBy 等操作，错误的使用了分组 Key，人为产生数据热点。

3、解决问题的思路：

业务上要尽量避免热点 key 的设计，例如我们可以把北京、上海等热点城市分成不同的区域，并进行单独处理；

技术上出现热点时，要调整方案打散原来的 key，避免直接聚合；此外 Flink 还提供了大量的功能可以避免数据倾斜。

3、Flink 任务数据倾斜场景和解决方案

A、两阶段聚合解决 KeyBy 热点：

• 首先把分组的 key 打散，比如加随机后缀；
• 对打散后的数据进行聚合；
• 把打散的 key 还原为真正的 key；
• 二次 KeyBy 进行结果统计，然后输出。

DataStream sourceStream = ...;
resultStream = sourceStream
     .map(record -> {
        Record record = JSON.parseObject(record, Record.class);
        String type = record.getType();
        record.setType(type + "#" + new Random().nextInt(100));
        return record;
      })
      .keyBy(0)
      .window(TumblingEventTimeWindows.of(Time.minutes(1)))
      .aggregate(new CountAggregate())
      .map(count -> {
        String key = count.getKey.substring(0, count.getKey.indexOf("#"));
        return RecordCount(key,count.getCount);
      })
      //二次聚合
      .keyBy(0)
      .process(new CountProcessFunction);
resultStream.sink()...

env.execute()...

B、GroupBy + Aggregation 分组聚合热点问题:

将SQL 拆成了内外两层，第一层通过随机打散 100 份的方式减少数据热点，当然这个打散的方式可以根据业务灵活指定。

select date,
       type,
       sum(pv) as pv
from(
  select
        date,
        type,
        sum(count) as pv
  from table
        group by
        date,
        type,
        floor(rand()*100) --随机打散成100份 
    )
    group by 
    date,
    type;

C、Flink 消费 Kafka 上下游并行度不一致导致的数据倾斜

Flink 消费 Kafka 的数据时，是推荐上下游并行度保持一致，即 Kafka 的分区数等于 Flink Consumer 的并行度。

但是会有一种情况，为了加快数据的处理速度，来设置 Flink 消费者的并行度大于 Kafka 的分区数。如果你不做任何的设置则会导致部分 Flink Consumer 线程永远消费不到数据。需要设置 Flink 的 Redistributing，也就是数据重分配。

dataStream
        .setParallelism(2)
        // 采用REBALANCE分区策略重分区
        .rebalance() //.rescale()
        .print()
        .setParallelism(4);

• Rebalance 分区策略，数据会以 round-robin 的方式对数据进行再次分区，可以全局负载均衡。
• Rescale 分区策略基于上下游的并行度，会将数据以循环的方式输出到下游的每个实例中

反思&扩展