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

Posted 2022-08-23 九师兄

1.概述

转载：有赞 Flink 实时任务资源优化探索与实践 这里是一个产品功能思路，觉得比较和我们相似，我们的是直接界面能查看所有节点的信息。这里可以借鉴一下。

首先，我们在源码层增加了 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 整体步骤如下：

• 获取一个实时任务所有的 Vertexs
• 遍历每个 Vertex，然后获取这个 Vertex 所有并发度 Task 的 taskOneRecordDealTime，并且记录其最大值
• 所有 Vertex 单条记录处理 Metric 最大值进行对比，找出处理时间最慢的 Vertex。

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