一口气搞懂「Flink Metrics」监控指标和性能优化，全靠这33张图和7千字（建议收藏）

迷南。 2022-09-14 02:18 801阅读 0赞

# 前言 #

大家好，我是土哥。

最近在公司做 Flink 推理任务的性能测试，要对 job 的`全链路吞吐`、`全链路时延`、`吞吐时延`指标进行监控和调优，其中要使用 Flink Metrics 对指标进行监控。

接下来这篇文章，干货满满，我将带领读者全面了解 Flink Metrics 指标监控，并通过实战案例，对`全链路吞吐`、`全链路时延`、`吞吐时延`的指标进行性能优化，彻底掌握 Flink Metrics 性能调优的方法和 Metrics 的使用。大纲目录如下：

![a2080ab3-df56-4be7-b27e-6acef34ce893.png][]

# 1 Flink Metrics 简介 #

Flink Metrics 是 Flink 集群运行中的各项指标，包含机器系统指标，比如：CPU、内存、线程、JVM、网络、IO、GC 以及任务运行组件（JM、TM、Slot、作业、算子）等相关指标。

Flink Metrics 包含两大作用：

1.  实时采集监控数据。在 Flink 的 UI 界面上，用户可以看到自己提交的任务状态、时延、监控信息等等。
2.  对外提供数据收集接口。用户可以将整个 Flink 集群的监控数据主动上报至第三方监控系统，如：prometheus、grafana 等，下面会介绍。

### 1.1 Flink Metric Types ###

Flink 一共提供了四种监控指标：分别为 Counter、Gauge、Histogram、Meter。

![6a40341b-4ce8-4c83-972b-cd87909e5854.png][]

**1. Count 计数器**

统计一个 指标的总量。写过 MapReduce 的开发人员就应该很熟悉 Counter，其实含义都是一样的，就是对一个计数器进行累加，即对于多条数据和多兆数据一直往上加的过程。其中 Flink 算子的接收记录总数 (numRecordsIn) 和发送记录总数 (numRecordsOut) 属于 Counter 类型。

使用方式：可以通过调用 counter(String name)来创建和注册 MetricGroup

![521340d0-be35-49a6-9fc5-d2e2ea6f6d2e.png][]

**2. Gauge 指标瞬时值**

Gauge 是最简单的 Metrics ，它反映一个指标的瞬时值。比如要看现在 TaskManager 的 JVM heap 内存用了多少，就可以每次实时的暴露一个 Gauge，Gauge 当前的值就是 heap 使用的量。

使用前首先创建一个实现 org.apache.flink.metrics.Gauge 接口的类。返回值的类型没有限制。您可以通过在 MetricGroup 上调用 gauge

![1e051497-f250-41fd-b621-b6e31e7a0f29.png][]

**3. Meter 平均值**

用来记录一个指标在某个时间段内的平均值。Flink 中的指标有 Task 算子中的 numRecordsInPerSecond,记录此 Task 或者算子每秒接收的记录数。

使用方式：通过 markEvent() 方法注册事件的发生。通过markEvent(long n) 方法注册同时发生的多个事件。

![e1b4ed10-4c14-4880-94fc-1b3dd056bd61.png][]

**4. Histogram 直方图**

Histogram 用于统计一些数据的分布，比如说 Quantile、Mean、StdDev、Max、Min 等，其中最重要一个是统计算子的延迟。此项指标会记录数据处理的延迟信息，对任务监控起到很重要的作用。

使用方式：通过调用 histogram(String name, Histogram histogram) 来注册一个 MetricGroup

![c63ef651-7812-4d75-a977-7df485f1824f.png][]

### 1.2 Scope ###

Flink 的指标体系按树形结构划分，域相当于树上的顶点分支，表示指标大的分类。每个指标都会分配一个标识符，该标识符将基于 3 个组件进行汇报：

1.  注册指标时用户提供的名称；
2.  可选的用户自定义域；
3.  系统提供的域。

例如，如果 A.B 是系统域，C.D 是用户域，E 是名称，那么指标的标识符将是 A.B.C.D.E. 你可以通过设置 conf/flink-conf.yam 里面的 metrics.scope.delimiter 参数来配置标识符的分隔符(默认“.”)

举例说明：以算子的指标组结构为例，其默认为：

<host>.taskmanager.<tm_id>.<job_name>.<operator_name>.<subtask_index>

算子的输入记录数指标为：

hlinkui.taskmanager.1234.wordcount.flatmap.0.numRecordsIn

### 1.3 Metrics 运行机制 ###

在生产环境下，为保证对Flink集群和作业的运行状态进行监控，Flink 提供两种集成方式：

**1.3.1 主动方式 MetricReport**

Flink Metrics 通过在 conf/flink-conf.yaml 中配置一个或者一些 reporters，将指标暴露给一个外部系统.这些 reporters 将在每个 job 和 task manager 启动时被实例化。

**1.3.2 被动方式 RestAPI**

通过提供 Rest 接口，被动接收外部系统调用，可以返回集群、组件、作业、Task、算子的状态。Rest API 实现类是 WebMonitorEndpoint

# 2 Flink Metrics 监控系统搭建 #

Flink 主动方式共提供了 8 种 Report。

我们使用 PrometheusPushGatewayReporter 方式 通过 prometheus + pushgateway + grafana 组件搭建 Flink On Yarn 可视化监控。

当 用户 使用 Flink 通过 session 模式向 yarn 集群提交一个 job 后，Flink 会通过 PrometheusPushGatewayReporter 将 metrics push 到 pushgateway 的 9091 端口上，然后使用外部系统 prometheus 从 pushgateway 进行 pull 操作，将指标采集过来，通过 Grafana可视化工具展示出来。 原理图如下：

![feaec9c8-0882-49d0-9677-ec544eb98602.png][]

首先，我们先在 Flink On Yarn 集群中提交一个 Job 任务，让其运行起来，然后执行下面的操作。

### 2.1 配置 Reporter ###

**下面所有工具、jar 包已经全部下载好，需要的朋友在公众号后台回复：08，可以全部获取到**。

![61ad6f82-5a2b-463f-8376-b0386f8490b2.png][]

**2.1.1 导包**

将 flink-metrics-prometheus\_2.11-1.13.2.jar 包导入 flink-1.13.2/bin 目录下

**2.1.2 配置 Reporter**

选取 PrometheusPushGatewayReporter 方式，通过在官网查询 Flink 1.13.2 Metrics 的配置后，在 flink-conf.yaml 设置，配置如下：

metrics.reporter.promgateway.class: org.apache.flink.metrics.prometheus.PrometheusPushGatewayReporter metrics.reporter.promgateway.host: 192.168.244.129 metrics.reporter.promgateway.port: 9091 metrics.reporter.promgateway.jobName: myJob metrics.reporter.promgateway.randomJobNameSuffix: true metrics.reporter.promgateway.deleteOnShutdown: false metrics.reporter.promgateway.groupingKey: k1=v1;k2=v2
    metrics.reporter.promgateway.interval: 60 SECONDS

### 2.2 部署 pushgateway ###

Pushgateway 是一个独立的服务，Pushgateway 位于应用程序发送指标和 Prometheus 服务器之间。

Pushgateway 接收指标，然后将其作为目标被 Prometheus 服务器拉取。可以将其看作代理服务，或者与 blackbox exporter 的行为相反，它接收度量，而不是探测它们。

**2.2.1 解压 pushgateway**

![4dbe5709-903f-4da5-88b6-e9928694c8bd.png][]

**2.2.2. 启动 pushgateway**

进入到 pushgateway-1.4.1 目录下

./pushgateway &

查看是否在后台启动成功

ps aux|grep pushgateway

**2.2.3. 登录 pushgateway webui**

![f0e0327f-1810-43b2-8a79-40a34c815071.png][]

### 2.3 部署 prometheus ###

Prometheus（普罗米修斯）是一个最初在 SoundCloud 上构建的监控系统。自 2012 年成为社区开源项目，拥有非常活跃的开发人员和用户社区。为强调开源及独立维护，Prometheus 于 2016 年加入云原生云计算基金会（CNCF），成为继Kubernetes 之后的第二个托管项目。

**2.3.1 解压prometheus-2.30.0**

![f602f8ed-aa54-4d56-a735-430a86e2f695.png][]

**2.3.2 编写配置文件**

scrape_configs:
      - job_name: 'prometheus'
        static_configs:
          - targets: ['192.168.244.129:9090']
            labels:
              instance: 'prometheus'
      - job_name: 'linux'
        static_configs:
          - targets: ['192.168.244.129:9100']
            labels:
              instance: 'localhost'
      - job_name: 'pushgateway'
        static_configs:
          - targets: ['192.168.244.129:9091']
            labels:
              instance: 'pushgateway'

**2.3.3 启动prometheus**

./prometheus --config.file=prometheus.yml &

启动完后，可以通过 ps 查看一下端口

ps aux|grep prometheus

**2.3.4 登录prometheus webui**

![df6c435b-950e-4244-9ace-36232162861c.png][]

### 2.4 部署 grafana ###

Grafana 是一个跨平台的开源的度量分析和可视化工具，可以通过将采集的数据查询然后可视化的展示，并及时通知。它主要有以下六大特点：

1、展示方式：快速灵活的客户端图表，面板插件有许多不同方式的可视化指标和日志，官方库中具有丰富的仪表盘插件，比如热图、折线图、图表等多种展示方式；

2、数据源：Graphite，InfluxDB，OpenTSDB，Prometheus，Elasticsearch，CloudWatch 和 KairosDB 等；

3、通知提醒：以可视方式定义最重要指标的警报规则，Grafana将不断计算并发送通知，在数据达到阈值时通过 Slack、PagerDuty 等获得通知；

4、混合展示：在同一图表中混合使用不同的数据源，可以基于每个查询指定数据源，甚至自定义数据源；

5、注释：使用来自不同数据源的丰富事件注释图表，将鼠标悬停在事件上会显示完整的事件元数据和标记；

6、过滤器：Ad-hoc 过滤器允许动态创建新的键/值过滤器，这些过滤器会自动应用于使用该数据源的所有查询。

**2.4.1 解压grafana-8.1.5**

![20018026-df24-4cd2-ad55-33b2a30d9e81.png][]

**2.4.2 启动grafana-8.1.5**

./bin/grafana-server web &

**2.4.3 登录 grafana**

登录用户名和密码都是 admin

![ccb6f72a-19d4-4690-ac6a-a1b0c1f66aae.png][]

**grafana 配置中文教程**：

https://grafana.com/docs/grafana/latest/datasources/prometheus/

**2.4.4 配置数据源、创建系统负载监控**

要访问 Prometheus 设置，请将鼠标悬停在配置（齿轮）图标上，然后单击数据源，然后单击 Prometheus 数据源，根据下图进行操作。

![9a496110-09db-48ea-8b48-20df9009936f.png][]

操作完成后，点击进行验证。

![d595bc7f-d839-4951-bb76-889b1e6ca219.png][]

**2.4.5 添加仪表盘**

点击最左侧 + 号，选择 DashBoard,选择新建一个 pannel

![53f8135e-5b98-4bab-abed-b03aeccf65da.png][]![6722d14d-8ec3-4365-945d-3d5fc39d7b22.png][]

至此，Flink 的 metrics 的指标展示在 Grafana 中了

flink 指标对应的指标名比较长，可以在 Legend 中配置显示内容，在\{ \{key\}\} 将 key 换成对应需要展示的字段即可，如： \{ \{job\_name\}\},\{ \{operator\_name\}\}

# 3 指标性能测试 #

上述监控系统搭建好了之后，我们可以进行性能指标监控了。现在以一个实战案例进行介绍：

### 3.1 业务场景介绍 ###

金融风控场景

**3.1.1 业务需求**:

Flink Source 从 data kafka topic 中读取推理数据，通过 sql 预处理成模型推理要求的数据格式，在进行 keyBy 分组后流入下游 connect 算子，与模型 connect 后进入 Co-FlatMap 算子再进行推理，原理图如下：

![a61d9b91-88fe-4e8e-8fe2-d4f6170ae23a.png][]

**3.1.2 业务要求**：

根据模型的复杂程度，要求推理时延到达 20ms 以内，全链路耗时 50ms 以内， 吞吐量达到每秒 1.2w 条以上。

**3.1.3 业务数据**：

推理数据：3000w，推理字段 495 个，机器学习 Xgboost 模型字段：495，

### 3.2 指标解析 ###

由于性能测试要求全链路耗时 50ms 以内，应该使用 Flink Metrics 的 Latency Marker 进行计算

**3.2.1 全链路时延计算方式** **：**

全链路时延指的是`一条推理数据`进入 source 算子到数据预处理算子直到最后一个算子输出结果的耗时，即处理一条数据需要多长时间，**包含算子内处理逻辑时间，算子间数据传递时间，缓冲区内等待时间**。

全链路时延要使用 latency metric 计算。latency metric 是由 source 算子根据当前本地时间生成的一个 marker ，并不参与各个算子的逻辑计算，仅仅跟着数据往下游算子流动，每到达一个算子则算出当前本地时间戳并与 source 生成的时间戳相减，得到 source 算子到当前算子的耗时，当到达 sink 算子或者说最后一个算子时，算出当前本地时间戳与 source 算子生成的时间戳相减，即得到全链路时延。原理图如下：

![1acac743-f03f-4b30-844e-d15f2ddb665c.png][]

由于使用到 Lateny marker,所有需要在 flink-conf.yaml 配置参数

latency.metrics.interval

系统配置截图如下：

![08ed7f27-6ada-4af4-901f-f3a99b38aad6.png][]

**3.2.2 全链路吞吐计算方式** **：**

全链路吞吐 = 单位时间处理数据数量 / 单位时间

### 3.3 提交任务到Flink on Yarn集群 ###

\*\*3.3.1 直接提交 Job \*\*

# -m jobmanager 的地址
    # -yjm 1024 指定 jobmanager 的内存信息
    # -ytm 1024 指定 taskmanager 的内存信息
    bin/flink run \
    -t yarn-per-job -yjm 4096 -ytm  8800 -s 96  \
    --detached  -c com.threeknowbigdata.datastream.XgboostModelPrediction \
    examples/batch/WordCount.jar  \

提交完成后，我们通过 Flink WEBUI 可以看到 job 运行的任务结果如下：

![17e6b25b-32f5-4377-9fc0-6ec7ec721d6d.png][]

因为推理模型只是一个 model，存在状态中，所以`全链路吞吐`考虑的是每秒`有多少条推理数据进入 source 算子到倒数第二个算子`（最后一个算子只是指标汇总）流出，这个条数就是全链路吞吐。

可以看到在处理 2000W 条数据时，代码直接统计输出的数值和 flink webUI 的统计数值基本一致，所以统计数值是可信的

![da617f70-3fa7-4f9e-8b40-44d7dc427e44.png][]

Flink WEBUI 跑的结果数据

![7533dbfc-e925-4a29-97d6-722020f66926.png][]

打开 Prometheus 在对话框输入全链路时延计算公式

计算公式：
    avg(flink_taskmanager_job_latency_source_id_
    operator_id _operator_subtask_index_latency{
        
    source_id="cbc357ccb763df2852fee8c4fc7d55f2",
    operator_id="c9c0ca46716e76f6b700eddf4366d243",quantile="0.999"})

![3a02b551-9fc0-4e57-9adc-9d7449c6af28.png][]

### 3.4 优化前性能分析 ###

在将任务提交到集群后，经过全链路时延计算公式、吞吐时延计算公式，最后得到优化前的结果 时延指标统计图如下：

![e5acb501-e74a-438a-86a3-b6e0c9c326f1.png][]

吞吐指标统计图如下：

![07c612e6-234c-4d87-b7f3-eb90a6cf34e0.png][]

通过本次测试完后，从图中可以发现：

**时延指标：加并行度，吞吐量也跟随高，但是全链路时延大幅增长（ 1并行至32并行，时延从 110ms 增加至 3287ms ）**

这远远没有达到要求的结果。

### 3.5 问题分析 ###

通过 Prometheus分析后，结果如下

![d9773898-0391-4338-a2b1-a83047015313.png][]

**3.5.1 并行度问题** **：**

1.  反压现象：在 Flink WEB-UI 上，可以看到应用存在着 **非常严重的反压**，这说明链路中存在较为耗时的算子，阻塞了整个链路；
2.  数据处理慢于拉取数据：数据源 **消费**数据的速度， **大于下游**数据处理速度；
3.  增加计算并行度：所以在接下来的测试中会 **调大推理算子并行度**，相当于提高下游数据处理能力；

**3.5.2 Buffer 超时问题** **：**

1.  Flink 虽是纯流式框架，但 **默认开启了缓存机制**（上游累积部分数据再发送到下游）；
2.  缓存机制可以提高应用的吞吐量，但是也 **增大了时延**；
3.  推理场景：为获取最好的时延指标，第二轮测试 **超时时间置 0**，记录吞吐量；

**3.5.3 Buffer 数量问题** **：**

同上，Flink 中的 Buffer 数量是可以配置的；

1.  Buffer 数量越多，能 **缓存的数据**也就 **越多**；
2.  推理场景：为获取最好的时延指标，第二轮测试： **减小 Flink 的 Buffer** 数量来优化时延指标；

**3.5.4 调优参数配置**

1.  SOURCE 与 COFLATMAP 的并行度按照 1:12 配置；
2.  Buffer 超时时间配置为 0ms (默认100ms)；

//在代码中设置
    
    senv.setBufferTimeout(0);

1.  Buffer 数量的配置如下：

修改flink-conf.yaml

memory.buffers-per-channel: 2
    memory.float-buffers-per-gate: 2
    memory.max-buffers-per-channel: 2

配置截图如下：

![55b2f6eb-e909-4408-a919-5f5cc9d69887.png][]

### 3.6 优化后性能分析 ###

经过修改配置后，将任务再次提交到集群后，经过全链路时延计算公式、吞吐时延计算公式，最后得到优化后的结果

时延指标统计图如下：

![827ee083-b73b-4a19-9be1-d481c691bc80.png][]

吞吐指标统计图如下：

![d1a913af-f4bf-4efb-8b14-ce318b1eb3c4.png][]

**优化后 LGB 推理测试总结** **：**

时延指标：并行度提升，时延也会增加，但幅度很小（可接受）。实际上，在测试过程中存在一定反压，若调大 SOURCE 与 COFLATMAP 的并行度比例，全链路时延可进一步降低； 吞吐量指标：随着并行度的增加，吞吐量也随着提高，当并行度提高至 96 时，吞吐量可以达到 1.3W，此时的时延维持在 50ms 左右（比较稳定）；

### 3.7 优化前后 LGB 分析总结 ###

如下图所示：

![a0ae0f85-9d57-4a73-94de-691303c8c72b.png][]

**3.7.1吞吐量---影响因素**：

**内存：对吞吐和时延没什么影响**， **并行度与吞吐成正相关**

1.  增大 kafka 分区,吞吐增加
2.  增大 source、维表 source 并行度
3.  增大 flatmap 推理并行度

**3.7.2全链路时延---影响因素**：

1.  **Buffer 超时越短、个数越少、时延越低**。
2.  整个链路是否有算子 **堵塞**（车道排队模型）
3.  调大推理算子并行度，时延降低，吞吐升高(即增加了推理的处理能力)

本文原创作者：土哥、一名大数据算法工程师。 文章首发平台：微信公众号【3分钟秒懂大数据】

土哥花费好几天写的文章，原创不易，各位觉得文章不错的话，不妨**点赞（在看）、留言、转发**三连走起！谢谢大家！

[a2080ab3-df56-4be7-b27e-6acef34ce893.png]: /images/20220828/ee9d002a60a648fbbe5baccb4b711b14.png
[6a40341b-4ce8-4c83-972b-cd87909e5854.png]: /images/20220828/857d17f8cdc346a69e6a82c556707229.png
[521340d0-be35-49a6-9fc5-d2e2ea6f6d2e.png]: /images/20220828/85ea9ef7c08140af88e2097e42e35e25.png
[1e051497-f250-41fd-b621-b6e31e7a0f29.png]: /images/20220828/985763449f4b4587adb65ba1d84a21de.png
[e1b4ed10-4c14-4880-94fc-1b3dd056bd61.png]: /images/20220828/f0aa8d5c6faf427faeee465c41b833da.png
[c63ef651-7812-4d75-a977-7df485f1824f.png]: /images/20220828/2b9610b8eb6d405cb632e7dbd87ecd89.png
[feaec9c8-0882-49d0-9677-ec544eb98602.png]: /images/20220828/9c2a87ccfe6648f59054d40795ab387d.png
[61ad6f82-5a2b-463f-8376-b0386f8490b2.png]: /images/20220828/ef5d440ced1a49f0be0c05f3fce8fbcf.png
[4dbe5709-903f-4da5-88b6-e9928694c8bd.png]: /images/20220828/f78a21dd13f84c5ab6248b7f454fda09.png
[f0e0327f-1810-43b2-8a79-40a34c815071.png]: /images/20220828/f41a87b22aaa4f85a5f76b354f57289e.png
[f602f8ed-aa54-4d56-a735-430a86e2f695.png]: /images/20220828/2ee31ccca09442d889ca020156a26d1e.png
[df6c435b-950e-4244-9ace-36232162861c.png]: /images/20220828/142c3fd780604410a45d446d9730cdc8.png
[20018026-df24-4cd2-ad55-33b2a30d9e81.png]: /images/20220828/c625ccad9728434a8a5cee73a8a912ee.png
[ccb6f72a-19d4-4690-ac6a-a1b0c1f66aae.png]: /images/20220828/60b8b130720a4f7e8d63928a02326bb7.png
[9a496110-09db-48ea-8b48-20df9009936f.png]: /images/20220828/afc44aab3b284965ac2a0f088c0f1faf.png
[d595bc7f-d839-4951-bb76-889b1e6ca219.png]: /images/20220828/4db643feacf74fe7bb7bc3db238bb0b1.png
[53f8135e-5b98-4bab-abed-b03aeccf65da.png]: https://files.mdnice.com/user/19005/53f8135e-5b98-4bab-abed-b03aeccf65da.png
[6722d14d-8ec3-4365-945d-3d5fc39d7b22.png]: /images/20220828/2f5988fb05b048f481aa438b903f6e06.png
[a61d9b91-88fe-4e8e-8fe2-d4f6170ae23a.png]: /images/20220828/03950a40e84f47b18a7bbc1ca7351aaf.png
[1acac743-f03f-4b30-844e-d15f2ddb665c.png]: /images/20220828/8504cc50dc8a4dbbad832526617106bd.png
[08ed7f27-6ada-4af4-901f-f3a99b38aad6.png]: /images/20220828/0de6e20ff7c245baa5fcc9f9c6625be8.png
[17e6b25b-32f5-4377-9fc0-6ec7ec721d6d.png]: /images/20220828/6c7a66f5fadd45bcba903cf8b79a3531.png
[da617f70-3fa7-4f9e-8b40-44d7dc427e44.png]: /images/20220828/be39b2a024e240b0bc0fb5d7e2bbc831.png
[7533dbfc-e925-4a29-97d6-722020f66926.png]: /images/20220828/5c471a2df63a4062910d0de408648660.png
[3a02b551-9fc0-4e57-9adc-9d7449c6af28.png]: /images/20220828/4925330e8b534f12abff8447ecf8ee39.png
[e5acb501-e74a-438a-86a3-b6e0c9c326f1.png]: /images/20220828/ee6aca343b7f4ffc86dc31aed1566930.png
[07c612e6-234c-4d87-b7f3-eb90a6cf34e0.png]: /images/20220828/7d563bf2fc33436c91eb84aedfadb5b1.png
[d9773898-0391-4338-a2b1-a83047015313.png]: /images/20220828/8f5661d8768a4cf29a01a248ce9aa115.png
[55b2f6eb-e909-4408-a919-5f5cc9d69887.png]: /images/20220828/a848320437da4fafa84007756cef962c.png
[827ee083-b73b-4a19-9be1-d481c691bc80.png]: /images/20220828/782321fcffbf4e21917d5267b94df7d5.png
[d1a913af-f4bf-4efb-8b14-ce318b1eb3c4.png]: /images/20220828/fd5aba72ec8a42918c69026d1c26a0ff.png
[a0ae0f85-9d57-4a73-94de-691303c8c72b.png]: /images/20220828/486889b88622445a9484fc0d8aa36d90.png