IoT 数据倾斜如何解决

在开发实时 Flink 任务的时候，可能会出现任务业务延时较高，超出预期的时间范围，导致数据无法及时正确产出的情景。比如摘要介绍的数据倾斜场景。这个时候就要从多个维度对任务进行调优。下面将介绍实时任务调优的一些方法。

定位性能瓶颈 Vertex

首先要定位出现问题的 Vertex。在运行信息 > Vertex 拓扑中查看节点的信息。每个 Vertex 会有自己的健康分，当某个 Vertex 健康分过低时，该 Vertex 可能出现问题，从而影响整个链路。

图片 1.png

链路中的 Vertex 也会出现反压现象。反压是指当下游的 Vertex 处理数据的能力不足时，上游会停止发送数据，避免数据丢失。定位反压节点可以看 IN_Q 和 OUT_Q 指标。每个 Vertex 的信息中，有 IN_Q 和 OUT_Q 指标，分别表示进入队列和输出队列中元素个数/队列总长度。当一个 Vertex 的 IN_Q 满，OUT_Q 空，则该节点有性能问题，会对上游的节点造成反压。另外一种方式是，在节点信息中，查看 BackPressure > Status，如果是绿色 ok，则不存在反压，如果是红色 high，则存在反压，该节点的下游存在性能瓶颈。

图片 2.png

找到性能出现瓶颈的节点后，开始对节点进行调优。

资源和参数调优

任务出现性能瓶颈一个很大的原因是资源分配不足，这个时候就需要对资源配置，例如并发数（Parallelism）、CPU（Core）和堆内存（Heap Memory）进行调优。其中 core 的默认值是 0.1，一般建议配置为 0.25,除了当处理数据源较小，且任务的计算逻辑不复杂的时候，可以为任务统一分配 1CU,此时每个节点的 core 随机根据节点的复杂度分配。

在 core 配置一定的情况下，调整节点的并发数，为节点分配资源。如果该节点是 source 节点，考虑数据源表的大小和数据源表分区的多少。根据数据源表的大小分配并发数，且并发数不超过源表分区的数量。例如数据源表是 sls 源表，只有两个分区时，只用一个并发就够了，最多两个。如果该节点是中间的计算节点，预算计算节点的复杂度，根据复杂度分配并发数。尽量把含有复杂运算的节点拆解出来，为其独立分配并发。

参数调优分为上下游参数的调优和作业参数的调优。针对上下游参数的调优主要是设置 batchsize,因为 flink 处理流式数据，每条数据均会触发上下读写,  为了降低上下游读写的压力，设置 batchsize 可以批量读写数据。该参数在 DDL 的 with 参数中设置。

当任务涉及到聚合逻辑(group by, partition by 等)时，需要加入任务调优参数。任务调优参数在作业参数中设置。

代码调优

调整资源和参数后，如果任务仍存在性能瓶颈，需要优化代码。

数据倾斜问题

判断数据倾斜：在运行信息 > Vertex 拓扑中选中一个节点后，在右侧的详细页面中选择 SubTasks,查看是否出现数据倾斜问题。观察每个 task 的 In Queue, Bytes Received 和 Records Received。如果某一两个 Task 的 In Queue 百分比或者 Bytes Received 特别高，和其他的 task 的 In Queue 形成明显差别，尤其当某个 task In Queue 的百分比长期为 100%，其他的为 0%，则出现了数据倾斜。

图片 3.png

解决方案

1. 利用 hashcode 和 MD5 手动打散数据，尽量使数据均匀分布。IoT 数据的一大特点就是数据倾斜比较严重，用户的数据量差异较大，在进行聚合运算时，经常会出现 key 聚集的情况，这种现象反应到 flink 的任务划分上，就是同一个 Vertex 拓扑中，部分 subtasks 资源过载，无法快速处理完数据，造成整个任务的反压现象。解决这个现象的办法就是进行数据的打散。

图片 4.png

1. 使用 LocalGroupAggregate

在 group by 的时候开启 local aggregate。其中注意统计函数用 sum 的时候要配合 filter,才能开启 local group aggregate。

分组排名问题 TOPN

1. TOPN 一共有三种算法，性能从高到低分别是：UpdateFastRank、 UnaryUpdateRank 和 RetractRank。涉及 TOP N 函数时，内部算法尽量使用 UpdateFastRank。会使性能更好。下图分别打散和不打散数据的性能对比和使用 UpdateFastRank 和 RetractRank 算法时的性能对比。（上下两个任务流程的数据源完全相同）

图片 5.png

1. 在多个维度对指标进行排名时，尽量使程序并行化，多使用 union all。例如 IoT 的数据层级细分很细，有云账号，租户，实例，产品, 设备的层级概念，这些数据还具有地域特性，地域的粒度也可以层级细分到国家，省份，城市等。所以当需要在不同维度对指标进行不同地域粒度的排名时，可以分别进行类目处理，并行计算，再用 union all 的方式整合，提高数据处理效率。