一、数据倾斜定位
通过 Web UI 各个 SubTask 的 Records Sent 和 Records Received 来确认，另外，还可以通过 Checkpoint detail 里不同的 SubTask 的 State Size 来判断是否数据倾斜。
例如上图，节点 2 的数据量明显高于其他节点的数据量，数据发生了很严重的倾斜问题。
二、数据倾斜的解决办法 2.1 keyBy 前的数据倾斜这种情况，需要视数据来源而定。比如从 kafka 读取数据，根据可能存在的不同场景，采取不同的方案。
2.1.1 程序并行度 != kafka 分区数这种情况下，就会导致有些 task 对接的 partition 数量多，有些 task 对接的 partition 数量少；最好保证程序的并行度和 kafka 分区数保持一致，每个 task 对应一个 partition。
2.1.2 kafka 分区数据本身存在倾斜如果能在 producer 端解决，并且不会对性能造成大的影响，最好在 producer 端解决。这样，不仅能解决 flink 数据倾斜的问题，也会解决 kafka 分区数据倾斜的问题，为 kafka 后续的操作便利。
如果不能在 producer 端解决数据倾斜的问题，只能在 flink 端解决，那么在接收到数据之后，需要对数据进行重分区，让 flink 任务强制 shuffle。使用 shuffle、rebalance、rescale 算子，都可以将数据打散重新均匀分布，从而解决数据倾斜问题。推荐使用 rescale，效率更高。
2.2 keyBy 后的数据倾斜 2.2.1 二次聚合存在的问题有些小伙伴可能想到，或许可以像 Spark 一样，采用二次聚合解决数据倾斜的问题。
比如，对商品按照 goodsId 进行 keyBy 后产生数据倾斜，计算商品量。
?采用二次聚合，第一次聚合时，按照 goodsId + "_" + 一定范围内的随机数(通常是并行度范围内的随机数)，得到一个聚合后的中间结果；
?
?第二次聚合时，第一次聚合的计算结果，按照 goodsId 进行聚合，再次计算，得到最终结果。
?通过两次聚合，把数据均匀分散到不同的 task ；第一次聚合，把数据打散，每一个批次计算得到一个中间值；第二次聚合，对第一次的中间值进行计算，数据量小很多，比如上图中，第二次聚合，每个task 收到来自不同上游 task 的数量只有 3 条。
但是，在 Flink 中，这种方式确实不可行的。因为 Flink 是流式计算，上游来一条数据，就会往下游发送一条数据，即使像 Spark 那样，先按照 goodsId+*聚合一次，再按照 goodsId 聚合一次，两次聚合后的 task ，收到的数据量是一样的，最终执行汇总计算的数据量还是和之前一样，数据倾斜没有得到解决。同时，还有可能导致数据在两次聚合中被重复计算。
?这个时候，就需要考虑其他的方案。
2.2.2 自定义分区打散直接落库如果是计算结果落到数据库，并且要求实时性高，需要数据实时落库，那么就不适合才去窗口的方式，会产生延迟。可以参考上面的二次聚合的操作，只不过只计算到第一次加了自定义分区 id 之后的聚合结果，把结果存到库里。
这种方案的影响就是，会造成数据库的数据量增大。本来一个 goodsId 对应一条数据，现在一个 goodsId 对应 3（跟自定义分区个数有关） 条数据。把 flink 中产生对性能的严重影响，转移到数据库中，同时可能对数据库的性能影响不会很大。当然，要结合自己的业务，综合评估，对数据库的影响是否可接收。
2.2.3 二次聚合 + 窗口如果对计算结果的实时性要求不是特别高，可以接收一点延迟，那么可以考虑 二次聚合 + 窗口 的方案。
类似于 Spark 中的二次聚合。不同的是，Spark 本身是批次操作，一个批次聚合后产生一个结果流向下游，所以第一次聚合后，数据被打散，一个批次只计算出来一个结果，流向下游；在 Flink 中，需要在聚合后，自己定义窗口，或者计时器，在一定的时间范围内，计算得到一个结果，流向下游。
?
?
比如上面的案例中，在第一次聚合后使用窗口操作，一个窗口产生一个结果，发送到下游。这样，在第二次聚合的时候，数据量就会大大减少。
窗口的大小，可以结合自己的业务确定，在满足时效性的同时，减轻数据倾斜带来的影响。
注：这里也可以使用定时器来取代窗口，某些场景下，定时器更优于窗口，两者的区别，可以参考Flink timer定时器使用
 1.本站遵循行业规范，任何转载的稿件都会明确标注作者和来源；2.本站的原创文章，会注明原创字样,如未注明都非原创，如有侵权请联系删除！；3.作者投稿可能会经我们编辑修改或补充；4.本站不提供任何储存功能只提供收集或者投稿人的网盘链接。 |
