建个好网站,国内代理ip地址 免费,免费推广原因,印度人通过什么网站做国际贸易简介#xff1a; 目前有赞实时计算平台对于 Flink 任务资源优化探索已经走出第一步。 随着 Flink K8s 化以及实时集群迁移完成#xff0c;有赞越来越多的 Flink 实时任务运行在 K8s 集群上#xff0c;Flink K8s 化提升了实时集群在大促时弹性扩缩容能力#xff0c;更好的降… 简介 目前有赞实时计算平台对于 Flink 任务资源优化探索已经走出第一步。 随着 Flink K8s 化以及实时集群迁移完成有赞越来越多的 Flink 实时任务运行在 K8s 集群上Flink K8s 化提升了实时集群在大促时弹性扩缩容能力更好的降低大促期间机器扩缩容的成本。同时由于 K8s 在公司内部有专门的团队进行维护 Flink K8s 化也能够更好的减低公司的运维成本。
不过当前 Flink K8s 任务资源是用户在实时平台端进行配置用户本身对于实时任务具体配置多少资源经验较少所以存在用户资源配置较多但实际使用不到的情形。比如一个 Flink 任务实际上 4 个并发能够满足业务处理需求结果用户配置了 16 个并发这种情况会导致实时计算资源的浪费从而对于实时集群资源水位以及底层机器成本都有一定影响。基于这样的背景本文从 Flink 任务内存以及消息能力处理方面对 Flink 任务资源优化进行探索与实践。
一、Flink 计算资源类型与优化思路
1.1 Flink 计算资源类型
一个 Flink 任务的运行所需要的资源我认为能够分为 5 类
内存资源本地磁盘或云盘存储依赖的外部存储资源。比如 HDFS、S3 等任务状态/数据HBase、MySQL、Redis 等数据CPU 资源网卡资源目前 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 起来后一起执行的物理载体。
数据源我们内部一般使用 KafkaKafka 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 所在的 JobVertexJobGraph 的点然后获取到该 JobVertex 所有 Task 的总输出最终会和 Kafka Topic 单位时间输入进行比对判断实时任务消息处理能力是否合理。
设实时任务 Kafka Topic 单位时间的输入为 S处理最慢的 Task 代表的 JobVertex 的并发度为 P处理最慢的 Task 所在的 JobVertex 单位时间输出为 O处理最慢的 Task 的最大消息处理时间为 T那么通过下面逻辑进行分析
当 O 约等于 S且 1 second / T * P 远大于 S 时会考虑减小任务并发度。当 O 约等于 S且 1 second / T * P 约等于 S 时不考虑调整任务并发度。当 O 远小于 S且 1 second / T * P 远小于 S 时会考虑增加任务并发度。
目前主要是 1 这种情况在 CPU 使用方面不合理当然由于不同时间段实时任务的流量不同所以我们会有一个周期性检测的的任务如果检测到某个实时任务连续多次都符合 1 这种情况时会自动报警提示平台管理员进行资源优化调整。 下图是从 Flink 任务的内存以及消息处理能力两个视角分析资源逻辑图 二、从内存视角对 Flink 分析实践
2.1 Flink 任务垃圾回收器选择
Flink 任务本质还是一个 Java 任务所以也就会涉及到垃圾回收器的选择。选择垃圾回收器一般需要从两个角度进行参考
吞吐量即单位时间内任务执行时间 / (任务执行时间 垃圾回收时间)当然并不是说降低 GC 停顿时间就能提升吞吐量因为降低 GC 停顿时间你的 GC 次数也会上升。延迟。如果你的 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 表示日志分析的结果。下面是我们平台对于某个实时任务内存分析的结果 下面是上面截图中部分参数说明
RunHoursFlink 任务运行小时数YGSize一个 TaskManager 新生代堆内存最大分配量单位兆YGUsePC一个 TaskManager 新生代堆最大使用率OGSize一个 TaskManager 老年代堆内存最大分配量单位兆OGUsePC一个 TaskManager 老生代堆最大使用率YGCoun一个 TaskMnager Young GC 次数YGPerTime一个 TaskMnager Young GC 每次停顿时间单位秒FGCount一个 TaskMnager Full GC 次数FGAllTime一个 TaskMnager Full GC 总时间单位秒ThroughtTask Manager 吞吐量AVG PT分析结果 avgPromotion 参数平均每次 Young GC 晋升到老年代的对象大小Rec Heap推荐的堆大小RecNewHeap推荐的新生代堆大小RecOldHeap推荐的老年代堆大小
上述大部分内存分析结果通过 GC Viewer 分析都能得到不过推荐堆大小、推荐新生代堆大小、推荐老年代堆大小则是根据 1.2.1 小节的内存优化规则来设置。
三、从消息处理视角对 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 任务自动构建其 PackagedProgramPackagedProgram 是 Flink 内部的一个类然后通过 PackagedProgram 我们可以获取一个 Flink Jar 任务的 StreamGraphStreamGraph 里面有 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?get0.Filter.numRecordsOut0 表示该 Vertex Task 的 idFilter.numRecordsOut 则表示具体的指标名称。我们内部使用 taskOneRecordDealTime 表示Task 处理单条记录时间 Metric然后用 0.taskOneRecordDealTime 去获取某个 Task 的单条记录处理时间的指标。上面接口支持多个指标查询即 get 后面使用逗号隔开即可。
最终自动化检测 Flink 消息处理最慢 Task 整体步骤如下
获取一个实时任务所有的 Vertexs遍历每个 Vertex然后获取这个 Vertex 所有并发度 Task 的 taskOneRecordDealTime并且记录其最大值所有 Vertex 单条记录处理 Metric 最大值进行对比找出处理时间最慢的 Vertex。
下面是我们实时平台对于一个 Flink 实时任务分析的结果 四、有赞 Flink 实时任务资源优化实践
既然 Flink 任务的内存以及消息处理能力分析的方式已经有了那接下来就是在实时平台端进行具体实践。我们实时平台每天会定时扫描所有正在运行的 Flink 任务在任务内存方面我们能够结合 实时任务 GC 日志同时根据内存优化规则计算出 Flink 任务推荐的堆内存大小并与实际分配的 Flink 任务的堆内存进行比较如果两者相差的倍数过大时我们认为 Flink 任务的内存配置存在浪费的情况接下来我们会报警提示到平台管理员进行优化。
平台管理员再收到报警提示后同时也会判定实时任务消息能力是否合理如果消息处理最慢的 Vertex (某段实时逻辑)其所有 Task 单位时间处理消息记录数的总和约等于实时任务消费的 Kafka Topic 单位时间的输入但通过 Vertex 的并发度以及单条消息处理 Metric 算出该 Vertex 单位时间处理的消息记录数远大于 Kafka Topic 的单位输入时则认为 Flink 任务可以适当调小并发度。具体调整多少会和业务方沟通之后在进行调整。整体 Flink 任务资源优化操作流程如下 五、总结
目前有赞实时计算平台对于 Flink 任务资源优化探索已经走出第一步。通过自动化发现能够优化的实时任务然后平台管理员介入分析最终判断是否能够调整 Flink 任务的资源。在整个实时任务资源优化的链路中目前还是不够自动化因为在后半段还需要人为因素。未来我们计划 Flink 任务资源的优化全部自动化会结合实时任务历史不同时段的资源使用情况自动化推测和调整实时任务的资源配置从而达到提升整个实时集群资源利用率的目的。
同时未来也会和元数据平台的同学进行合作一起从更多方面来分析实时任务是否存在资源优化的可能性他们在原来离线任务资源方面积攒了很多优化经验未来也可以参考和借鉴应用到实时任务资源的优化中。
当然最理想化就是实时任务的资源使用能够自己自动弹性扩缩容之前听到过社区同学有这方面的声音同时也欢迎你能够和我一起探讨。
作者沈磊
原文链接
本文为阿里云原创内容未经允许不得转载
