如何处理 Flink 作业中的数据倾斜问题？

分析&回答

什么是数据倾斜？

由于数据分布不均匀，造成数据大量的集中到一点，造成数据热点。

举例：一个 Flink 作业包含 200 个 Task 节点，其中有 199 个节点可以在很短的时间内完成计算。但是有一个节点执行时间远超其他结果，并且随着数据量的持续增加，导致该计算节点挂掉，从而整个任务失败重启。我们可以在 Flink 的管理界面中看到任务的某一个 Task 数据量远超其他节点。

大数据框架的特性

不怕数据大，怕数据倾斜。
jobs数比较多的作业运行效率相对比较低，如子查询比较多。
sum,count,max,min等聚集函数，不会有数据倾斜问题

容易数据倾斜情况

group by
count(distinct ),在数据量大的情况下，容易数据倾斜，因为count(distinct)是按group by 字段分组，按distinct字段排序。
小表关联超大表

优化常用的手段

Flink 任务出现数据倾斜的直观表现是任务节点频繁出现反压，但是增加并行度后并不能解决问题；部分节点出现 OOM 异常，是因为大量的数据集中在某个节点上，导致该节点内存被爆，任务失败重启。

产生数据倾斜的原因主要有 2 个方面：

业务上有严重的数据热点，比如滴滴打车的订单数据中北京、上海等几个城市的订单量远远超过其他地区；
技术上大量使用了 KeyBy、GroupBy 等操作，错误的使用了分组 Key，人为产生数据热点。

因此解决问题的思路也很清晰：

业务上要尽量避免热点 key 的设计，例如我们可以把北京、上海等热点城市分成不同的区域，并进行单独处理；
技术上出现热点时，要调整方案打散原来的 key，避免直接聚合；此外 Flink 还提供了大量的功能可以避免数据倾斜。

解决数据倾斜问题

减少job数（合并MapReduce，用Multi-group by）
设置合理的mapreduce的task数，能有效提升性能。
数据量较大的情况下，慎用count(distinct)。
对小文件进行合并，针对文件数据源。

优化案例

join原则将条目少的表/子查询放在 Join的左边。原因是在 Join 操作的 Reduce 阶段，位于 Join左边的表的内容会被加载进内存，将条目少的表放在左边，可以有效减少发生内存溢出的几率。

当一个小表关联一个超大表时，容易发生数据倾斜，可以用MapJoin把小表全部加载到内存在map端进行join，避免reducer处理。

如：SELECT /*+ MAPJOIN(user)*/  l.session_id, u.username from user u join page_views lon (u. id=l.user_id) ;
复制代码

笛卡尔积当Hive设定为严格模式（hive.mapred.mode=strict）时，不允许在HQL语句中出现笛卡尔积。

当无法躲避笛卡尔积时，采用MapJoin，会在Map端完成Join操作，将Join操作的一个或多个表完全读入内存。

MapJoin的用法是在查询/子查询的SELECT关键字后面添加/*+MAPJOIN(tablelist) */提示优化器转化为MapJoin 。

其中tablelist可以是一个表，或以逗号连接的表的列表。tablelist中的表将会读入内存，应该将小表写在这里

控制Map数

同时可执行的map数是有限的。

通常情况下，作业会通过input的目录产生一个或者多个map任务

主要的决定因素有： input的文件总个数，input的文件大小。

举例：a) 假设input目录下有1个文件a,大小为780M,那么hadoop会将该文件a分隔成7个块（block为128M,6个128m的块和1个12m的块），从而产生7个map数b) 假设input目录下有3个文件a,b,c,大小分别为10m，20m，130m，那么hadoop会分隔成4个块（10m,20m,128m,2m）,从而产生4个map数
复制代码

两种方式控制Map数：即减少map数和增加map数

减少map数可以通过合并小文件来实现，这点是对文件数据源来讲。
增加map数的可以通过控制上一个job的reduer数来控制

反思&扩展

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

通常我们在使用 Flink 处理实时业务时，上游一般都是消息系统，Kafka 是使用最广泛的大数据消息系统。当使用 Flink 消费 Kafka 数据时，也会出现数据倾斜。

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

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

这时候你需要设置 Flink 的 Redistributing，也就是数据重分配。

GroupBy + Aggregation 分组聚合热点问题

业务上通过 GroupBy 进行分组，然后紧跟一个 SUM、COUNT 等聚合操作是非常常见的。我们都知道 GroupBy 函数会根据 Key 进行分组，完全依赖 Key 的设计，如果 Key 出现热点，那么会导致巨大的 shuffle，相同 key 的数据会被发往同一个处理节点；如果某个 key 的数据量过大则会直接导致该节点成为计算瓶颈，引起反压。