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

 

定位性能瓶颈Vertex

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

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

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

 

资源和参数调优

任务出现性能瓶颈一个很大的原因是资源分配不足，这个时候就需要对资源配置，例如并发数（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%，则出现了数据倾斜。

 

解决方案

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

1. 使用LocalGroupAggregate

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

 

分组排名问题TOPN

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

 

 

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