有赞 Flink 实时任务资源优化探索与实践

简介： 目前有赞实时计算平台对于 Flink 任务资源优化探索已经走出第一步。

随着 Flink K8s 化以及实时集群迁移完成，有赞愈来愈多的 Flink 实时任务运行在 K8s 集群上，Flink K8s 化提高了实时集群在大促时弹性扩缩容能力，更好的下降大促期间机器扩缩容的成本。同时，因为 K8s 在公司内部有专门的团队进行维护， Flink K8s 化也可以更好的减低公司的运维成本。

不过当前 Flink K8s 任务资源是用户在实时平台端进行配置，用户自己对于实时任务具体配置多少资源经验较少，因此存在用户资源配置较多，但实际使用不到的情形。好比一个 Flink 任务实际上 4 个并发可以知足业务处理需求，结果用户配置了 16 个并发，这种状况会致使实时计算资源的浪费，从而对于实时集群资源水位以及底层机器成本，都有必定影响。基于这样的背景，本文从 Flink 任务内存以及消息能力处理方面，对 Flink 任务资源优化进行探索与实践。

1、Flink 计算资源类型与优化思路

1.1 Flink 计算资源类型

一个 Flink 任务的运行，所须要的资源我认为可以分为 5 类：

1. 内存资源
2. 本地磁盘（或云盘）存储
3. 依赖的外部存储资源。好比 HDFS、S3 等（任务状态/数据），HBase、MySQL、Redis 等（数据）
4. CPU 资源
5. 网卡资源

目前 Flink 任务使用最主要的仍是内存和 CPU 资源，本地磁盘、依赖的外部存储资源以及网卡资源通常都不会是瓶颈，因此本文咱们是从 Flink 任务的内存和 CPU 资源，两个方面来对 Flink 实时任务资源进行优化。

1.2 Flink 实时任务资源优化思路

对于 Flink 实时任务资源分析思路，咱们认为主要包含两点：

• 一是从任务内存视角，从堆内存方面对实时任务进行分析。
• 另外一方面则是从实时任务消息处理能力入手，保证知足业务方数据处理需求的同时，尽量合理使用 CPU 资源。

以后再结合实时任务内存分析所得相关指标、实时任务并发度的合理性，得出一个实时任务资源预设值，在和业务方充分沟通后，调整实时任务资源，最终达到实时任务资源配置合理化的目的，从而更好的下降机器使用成本。

■ 1.2.1 任务内存视角

那么如何分析 Flink 任务的堆内存呢？这里咱们是结合 Flink 任务 GC 日志来进行分析。GC 日志包含了每次 GC 堆内不一样区域内存的变化和使用状况。同时根据 GC 日志，也可以获取到一个 Taskmanager 每次 Full GC 后，老年代剩余空间大小。能够说，获取实时任务的 GC 日志，使咱们进行实时任务内存分析的前提。

GC 日志内容分析，这里咱们借助开源的 GC Viewer 工具来进行具体分析，每次分析完，咱们可以获取到 GC 相关指标，下面是经过 GC Viewer 分析一次 GC 日志的部分结果：

上面经过 GC 日志分析出单个 Flink Taskmanager 堆总大小、年轻代、老年代分配的内存空间、Full GC 后老年代剩余大小等，固然还有不少其余指标，相关指标定义能够去 Github 具体查看。

这里最重要的仍是Full GC 后老年代剩余大小这个指标，按照《Java 性能优化权威指南》这本书 Java 堆大小计算法则，设 Full GC 后老年代剩余大小空间为 M，那么堆的大小建议 3 ~ 4倍 M，新生代为 1 ~ 1.5 倍 M，老年代应为 2 ~ 3 倍 M，固然，真实对内存配置，你能够按照实际状况，将相应比例再调大些，用以防止流量暴涨情形。

因此经过 Flink 任务的 GC 日志，咱们能够计算出实时任务推荐的堆内存总大小，当发现推荐的堆内存和实际实时任务的堆内存大小相差过大时，咱们就认为可以去下降业务方实时任务的内存配置，从而下降机器内存资源的使用。

■ 1.2.2 任务消息处理能力视角

对于 Flink 任务消息处理能力分析，咱们主要是看实时任务消费的数据源单位时间的输入，和实时任务各个 Operator / Task 消息处理能力是否匹配。Operator 是 Flink 任务的一个算子，Task 则是一个或者多个算子 Chain 起来后，一块儿执行的物理载体。

数据源咱们内部通常使用 Kafka，Kafka Topic 的单位时间输入能够经过调用 Kafka Broker JMX 指标接口进行获取，固然你也能够调用 Flink Rest Monitoring 相关 API 获取实时任务全部 Kafka Source Task 单位时间输入，而后相加便可。不过因为反压可能会对 Source 端的输入有影响，这里咱们是直接使用 Kafka Broker 指标 JMX 接口获取 Kafka Topic 单位时间输入。

在获取到实时任务 Kafka Topic 单位时间输入后，下面就是判断实时任务的消息处理能力是否与数据源输入匹配。一个实时任务总体的消息处理能力，会受处处理最慢的 Operator / Task 的影响。打个比方，Flink 任务消费的 Kafka Topic 输入为 20000 Record / S，可是有一个 Map 算子，其并发度为 10 ，Map 算子中业务方调用了 Dubbo，一个 Dubbo 接口从请求到返回为 10 ms，那么 Map 算子处理能力 1000 Record / S (1000 ms / 10 ms * 10 )，从而实时任务处理能力会降低为 1000 Record / S。

因为一条消息记录的处理会在一个 Task 内部流转，因此咱们试图找出一个实时任务中，处理最慢的 Task 逻辑。若是 Source 端到 Sink 端所有 Chain 起来的话，咱们则是会找出处理最慢的 Operator 的逻辑。在源码层，咱们针对 Flink Task 以及 Operator 增长了单条记录处理时间的自定义 Metric，以后该 Metric 能够经过 Flink Rest API 获取。咱们会遍历一个 Flink 任务中全部的 Task , 查询处理最慢的 Task 所在的 JobVertex（JobGraph 的点），而后获取到该 JobVertex 全部 Task 的总输出，最终会和 Kafka Topic 单位时间输入进行比对，判断实时任务消息处理能力是否合理。

设实时任务 Kafka Topic 单位时间的输入为 S，处理最慢的 Task 表明的 JobVertex 的并发度为 P，处理最慢的 Task 所在的 JobVertex 单位时间输出为 O，处理最慢的 Task 的最大消息处理时间为 T，那么经过下面逻辑进行分析：

1. 当 O 约等于 S，且 1 second / T * P 远大于 S 时，会考虑减少任务并发度。
2. 当 O 约等于 S，且 1 second / T * P 约等于 S 时，不考虑调整任务并发度。
3. 当 O 远小于 S，且 1 second / T * P 远小于 S 时，会考虑增长任务并发度。

目前主要是 1 这种状况在 CPU 使用方面不合理，固然，因为不一样时间段，实时任务的流量不一样，因此咱们会有一个周期性检测的的任务，若是检测到某个实时任务连续屡次都符合 1 这种状况时，会自动报警提示平台管理员进行资源优化调整。
下图是从 Flink 任务的内存以及消息处理能力两个视角分析资源逻辑图：

2、从内存视角对 Flink 分析实践

2.1 Flink 任务垃圾回收器选择

Flink 任务本质仍是一个 Java 任务，因此也就会涉及到垃圾回收器的选择。选择垃圾回收器通常须要从两个角度进行参考：

1. 吞吐量，即单位时间内，任务执行时间 / (任务执行时间 + 垃圾回收时间)，固然并非说下降 GC 停顿时间就能提高吞吐量，由于下降 GC 停顿时间，你的 GC 次数也会上升。
2. 延迟。若是你的 Java 程序涉及到与外部交互，延迟会影响外部的请求使用体验。

Flink 任务我认为仍是偏重吞吐量的一类 Java 任务，因此会从吞吐量角度进行更多的考量。固然并非说彻底不考虑延迟，毕竟 JobManager、TaskManager、ResourceManager 之间存在心跳，延迟过大，可能会有心跳超时的可能性。

目前咱们 JDK 版本为内部 JDK 1.8 版本，新生代垃圾回收器使用 Parallel Scavenge，那么老年代垃圾回收器只能从 Serial Old 或者 Parallel Old 中选择。因为咱们 Flink k8s 任务每一个 Pod 的 CPU 限制为 0.6 - 1 core ，最大也只能使用 1 个 core，因此老年代的垃圾回收器咱们使用的是 Serial Old ，多线程垃圾回收在单 Core 之间，可能会有线程切换的消耗。

2.2 实时任务 GC 日志获取

设置完垃圾回收器后，下一步就是获取 Flink 任务的 GC 日志。Flink 任务构成通常是单个 JobManager + 多个 TaskManger ，这里须要获取到 TaskManager 的 GC 日志进行分析。那是否是要对全部 TaskManager 进行获取呢。这里咱们按照 TaskManager 的 Young GC 次数，按照次数大小进行排序，取排名前 16 的 TaskManager 进行分析。YoungGC 次数能够经过 Flink Rest API 进行获取。

Flink on Yarn 实时任务的 GC 日志，直接点开 TaskManager 的日志连接就可以看到，而后经过 HTTP 访问，就能下载到本地。Flink On k8s 任务的 GC 日志，会先写到 Pod 所挂载的云盘，基于 k8s hostpath volume 进行挂载。咱们内部使用 Filebeat 进行日志文件变动监听和采集，最终输出到下游的 Kafka Topic。咱们内部会有自定义日志服务端，它会消费 Kafka 的日志记录，自动进行落盘和管理，同时向外提供日志下载接口。经过日志下载的接口，便可以下载到须要分析的 TaskManager 的 GC 日志。

2.3 基于 GC Viewer 分析 Flink 任务内存

GC Viewer 是一个开源的 GC 日志分析工具。使用 GC Viewer 以前，须要先把 GC Viewer 项目代码 clone 到本地，而后进行编译打包，就可使用其功能。

在对一个实时任务堆内存进行分析时，先把 Flink TaskManager 的日志下载到本地，而后经过 GC Viewer 对日志进行。若是你以为多个 Taskmanager GC 日志分析较慢时，可使用多线程。上面全部这些操做，能够将其代码化，自动化产出分析结果。下面是经过 GC Viewer 分析的命令行：

java -jar gcviewer-1.37-SNAPSHOT.jar gc.log summary.csv

上面参数 gc.log 表示一个 Taskmanager 的 GC 日志文件名称，summary.csv 表示日志分析的结果。下面是咱们平台对于某个实时任务内存分析的结果：

下面是上面截图中，部分参数说明：

1. RunHours，Flink 任务运行小时数
2. YGSize，一个 TaskManager 新生代堆内存最大分配量，单位兆
3. YGUsePC，一个 TaskManager 新生代堆最大使用率
4. OGSize，一个 TaskManager 老年代堆内存最大分配量，单位兆
5. OGUsePC，一个 TaskManager 老生代堆最大使用率
6. YGCoun，一个 TaskMnager Young GC 次数
7. YGPerTime，一个 TaskMnager Young GC 每次停顿时间，单位秒
8. FGCount，一个 TaskMnager Full GC 次数
9. FGAllTime，一个 TaskMnager Full GC 总时间，单位秒
10. Throught，Task Manager 吞吐量
11. AVG PT（分析结果 avgPromotion 参数），平均每次 Young GC 晋升到老年代的对象大小
12. Rec Heap，推荐的堆大小
13. RecNewHeap，推荐的新生代堆大小
14. RecOldHeap，推荐的老年代堆大小

上述大部份内存分析结果，经过 GC Viewer 分析都能获得，不过推荐堆大小、推荐新生代堆大小、推荐老年代堆大小则是根据 1.2.1 小节的内存优化规则来设置。

3、从消息处理视角对 Flink 分析实践

3.1 实时任务 Kafka Topic 单位时间输入获取

想要对 Flink 任务的消息处理能力进行分析，第一步即是获取该实时任务的 Kafka 数据源 Topic，目前若是数据源不是 Kafka 的话，咱们不会进行分析。Flink 任务整体分为两类：Flink Jar 任务和 Flink SQL 任务。Flink SQL 任务获取 Kafka 数据源比较简单，直接解析 Flink SQL 代码，而后获取到 With 后面的参数，在过滤掉 Sink 表以后，若是 SQLCreateTable 的 Conector 类型为 Kafka，就可以经过 SQLCreateTable with 后的参数，拿到具体 Kafka Topic。

Flink Jar 任务的 Kafka Topic 数据源获取相对繁琐一些，咱们内部有一个实时任务血缘解析服务，经过对 Flink Jar 任务自动构建其 PackagedProgram，PackagedProgram 是 Flink 内部的一个类，而后经过 PackagedProgram ，咱们能够获取一个 Flink Jar 任务的 StreamGraph，StreamGraph 里面有 Source 和 Sink 的全部 StreamNode，经过反射，咱们能够获取 StreamNode 里面具体的 Source Function，若是是 Kafka Source Sunction，咱们就会获取其 Kafka Topic。下面是 StreamGraph 类截图：

获取到 Flink 任务的 Kafka Topic 数据源以后，下一步即是获取该 Topic 单位时间输入的消息记录数，这里能够经过 Kafka Broker JMX Metric 接口获取，咱们则是经过内部 Kafka 管理平台提供的外部接口进行获取。

3.2 自动化检测 Flink 消息处理最慢 Task

首先，咱们在源码层增长了 Flink Task 单条记录处理时间的 Metric，这个 Metric 能够经过 Flink Rest API 获取。接下来就是借助 Flink Rest API，遍历要分析的 Flink 任务的全部的 Task。Flink Rest Api 有这样一个接口：

base_flink_web_ui_url/jobs/:jobid

这个接口可以获取一个任务的全部 Vertexs，一个 Vertex 能够简单理解为 Flink 任务 JobGraph 里面的一个 JobVertex。JobVertex 表明着实时任务中一段执行逻辑。

获取完 Flink 任务全部的 Vertex 以后，接下来就是获取每一个 Vertex 具体 Task 处理单条记录的 metric，可使用下面的接口：

须要在上述 Rest API 连接 metrics 以后添加 ?get=（具体meitric ），好比：metrics?get=0.Filter.numRecordsOut，0 表示该 Vertex Task 的 id，Filter.numRecordsOut 则表示具体的指标名称。咱们内部使用 taskOneRecordDealTime 表示Task 处理单条记录时间 Metric，而后用 0.taskOneRecordDealTime 去获取某个 Task 的单条记录处理时间的指标。上面接口支持多个指标查询，即 get 后面使用逗号隔开便可。

最终自动化检测 Flink 消息处理最慢 Task 总体步骤以下：

1. 获取一个实时任务全部的 Vertexs
2. 遍历每一个 Vertex，而后获取这个 Vertex 全部并发度 Task 的 taskOneRecordDealTime，而且记录其最大值
3. 全部 Vertex 单条记录处理 Metric 最大值进行对比，找出处理时间最慢的 Vertex。

下面是咱们实时平台对于一个 Flink 实时任务分析的结果：

4、有赞 Flink 实时任务资源优化实践

既然 Flink 任务的内存以及消息处理能力分析的方式已经有了，那接下来就是在实时平台端进行具体实践。咱们实时平台天天会定时扫描全部正在运行的 Flink 任务，在任务内存方面，咱们可以结合 实时任务 GC 日志，同时根据内存优化规则，计算出 Flink 任务推荐的堆内存大小，并与实际分配的 Flink 任务的堆内存进行比较，若是二者相差的倍数过大时，咱们认为 Flink 任务的内存配置存在浪费的状况，接下来咱们会报警提示到平台管理员进行优化。

平台管理员再收到报警提示后，同时也会断定实时任务消息能力是否合理，若是消息处理最慢的 Vertex (某段实时逻辑)，其全部 Task 单位时间处理消息记录数的总和约等于实时任务消费的 Kafka Topic 单位时间的输入，但经过 Vertex 的并发度，以及单条消息处理 Metric ，算出该 Vertex 单位时间处理的消息记录数远大于 Kafka Topic 的单位输入时，则认为 Flink 任务能够适当调小并发度。具体调整多少，会和业务方沟通以后，在进行调整。总体 Flink 任务资源优化操做流程以下：

5、总结

目前有赞实时计算平台对于 Flink 任务资源优化探索已经走出第一步。经过自动化发现可以优化的实时任务，而后平台管理员介入分析，最终判断是否可以调整 Flink 任务的资源。在整个实时任务资源优化的链路中，目前仍是不够自动化，由于在后半段还须要人为因素。将来咱们计划 Flink 任务资源的优化所有自动化，会结合实时任务历史不一样时段的资源使用状况，自动化推测和调整实时任务的资源配置，从而达到提高整个实时集群资源利用率的目的。

同时将来也会和元数据平台的同窗进行合做，一块儿从更多方面来分析实时任务是否存在资源优化的可能性，他们在原来离线任务资源方面积攒了不少优化经验，将来也能够参考和借鉴，应用到实时任务资源的优化中。

固然，最理想化就是实时任务的资源使用可以本身自动弹性扩缩容，以前听到过社区同窗有这方面的声音，同时也欢迎你可以和我一块儿探讨。

做者：沈磊
原文连接本文为阿里云原创内容，未经容许不得转载