在大数据处理领域，数据倾斜是一个非常常见的问题，今天我们就简单讲讲在flink中如何处理流式数据倾斜问题。

1.数据倾斜的原理和影响

1.1 原理

数据倾斜就是数据的分布严重不均，造成一部分数据很多，一部分数据很少的局面。

数据分布理论上都是倾斜的，符合“二八原理”：例如80%的财富集中在20%的人手中、80%的用户只使用20%的功能、20%的用户贡献了80%的访问量。

数据倾斜的现象，如下图所示。

Flink数据倾斜概述与优化-风君雪科技博客

1.2 影响

（1）单点问题

数据集中在某些分区上（Subtask），导致数据严重不平衡。

（2）GC 频繁

过多的数据集中在某些 JVM（TaskManager），使得JVM 的内存资源短缺，导致频繁 GC。

（3）吞吐下降、延迟增大

数据单点和频繁 GC 导致吞吐下降、延迟增大。

（4）系统崩溃

严重情况下，过长的 GC 导致 TaskManager 失联，系统崩溃。

Flink数据倾斜概述与优化-风君雪科技博客

2.Flink 如何定位数据倾斜？

步骤1：定位反压

定位反压有2种方式：Flink Web UI 自带的反压监控（直接方式）、Flink Task Metrics（间接方式）。通过监控反压的信息，可以获取到数据处理瓶颈的 Subtask。

参考：【Flink 精选】如何分析及处理反压?

步骤2：确定数据倾斜

Flink Web UI 自带Subtask 接收和发送的数据量。当 Subtasks 之间处理的数据量有较大的差距，则该 Subtask 出现数据倾斜。如下图所示，红框内的 Subtask 出现数据热点。

Flink数据倾斜概述与优化-风君雪科技博客

3.Flink 如何处理常见数据倾斜？

Flink数据倾斜概述与优化-风君雪科技博客

场景一：数据源 source 消费不均匀

解决思路：通过调整并发度，解决数据源消费不均匀或者数据源反压的情况。

例如kafka数据源，可以调整 KafkaSource 的并发度解决消费不均匀。

调整并发度的原则：KafkaSource 并发度与 kafka 分区数是一样的，或者 kafka 分区数是KafkaSource 并发度的整数倍。

场景二：key 分布不均匀的无统计场景

说明：key 分布不均匀的无统计场景，例如上游数据分布不均匀，使用keyBy来打散数据。

解决思路： 通过添加随机前缀，打散 key 的分布，使得数据不会集中在几个 Subtask。

 Flink数据倾斜概述与优化-风君雪科技博客

 具体措施：
① 在原来分区 key/uid 的基础上，加上随机的前缀或者后缀。
② 使用数据到达的顺序seq，作为分区的key。

场景三：key 分布不均匀的统计场景

解决思路：聚合统计前，先进行预聚合，例如两阶段聚合（加盐局部聚合+去盐全局聚合）。

Flink数据倾斜概述与优化-风君雪科技博客

 两阶段聚合的具体措施：
① 预聚合：加盐局部聚合，在原来的 key 上加随机的前缀或者后缀。
② 聚合：去盐全局聚合，删除预聚合添加的前缀或者后缀，然后进行聚合统计。 

4.一个SQL演示案例

我们先来看一个可能产生数据倾斜的sql

select TUMBLE_END(proc_time, INTERVAL '1' MINUTE) as winEnd,plat,count(*) as pv  from source_kafka_table 
group by TUMBLE(proc_time, INTERVAL '1' MINUTE) ,plat

在这个sql里，我们统计一个网站各个端的每分钟的pv，从kafka消费过来的数据首先会按照端进行分组，然后执行聚合函数count来进行pv的计算。

如果某一个端产生的数据特别大，比如我们的微信小程序端产生数据远远大于其他app端的数据，那么把这些数据分组到某一个算子之后，由于这个算子的处理速度跟不上，就会产生数据倾斜。

 查看flink的ui，会看到如下的场景。

Flink数据倾斜概述与优化-风君雪科技博客

对于这种简单的数据倾斜，我们可以通过对分组的key加上随机数，再次打散，分别计算打散后不同的分组的pv数，然后在最外层再包一层，把打散的数据再次聚合，这样就解决了数据倾斜的问题。

优化后的sql如下：

select winEnd,split_index(plat1,'_',0) as plat2,sum(pv) from (

  select TUMBLE_END(proc_time, INTERVAL '1' MINUTE) as winEnd,plat1,count(*) as pv from (

    -- 最内层，将分组的key，也就是plat加上一个随机数打散
    select plat || '_' || cast(cast(RAND()*100 as int) as string) as plat1 ,proc_time from source_kafka_table 

) group by TUMBLE(proc_time, INTERVAL '1' MINUTE), plat1

) group by winEnd,split_index(plat1,'_',0)

在这个sql的最内层，将分组的key，也就是plat加上一个随机数打散，然后求打散后的各个分组（也就是sql中的plat1）的pv值，然后最外层，将各个打散的pv求和。

注意：最内层的sql，给分组的key添加的随机数，范围不能太大，也不能太小，太大的话，分的组太多，增加checkpoint的压力，太小的话，起不到打散的作用。

在我的测试中，一天大概十几亿的数据量，5个并行度，随机数的范围在100范围内，就可以正常处理了。

修改后我们看到各个子任务的数据基本均匀了。

 Flink数据倾斜概述与优化-风君雪科技博客

https://www.cnblogs.com/qiu-hua/p/14056747.html

https://www.cnblogs.com/qiu-hua/p/14056780.html

欢迎关注微信公众号：大数据从业者