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。

Flink Metrics监控指标和性能优化 - 图1

1. Count 计数器

统计一个 指标的总量。写过 MapReduce 的开发人员就应该很熟悉 Counter，其实含义都是一样的，就是对一个计数器进行累加，即对于多条数据和多兆数据一直往上加的过程。其中 Flink 算子的接收记录总数 (numRecordsIn) 和发送记录总数 (numRecordsOut) 属于 Counter 类型。 使用方式：可以通过调用 counter(String name)来创建和注册 MetricGroup

Flink Metrics监控指标和性能优化 - 图2

2. Gauge 指标瞬时值

Gauge 是最简单的 Metrics ，它反映一个指标的瞬时值。比如要看现在 TaskManager 的 JVM heap 内存用了多少，就可以每次实时的暴露一个 Gauge，Gauge 当前的值就是 heap 使用的量。 使用前首先创建一个实现 org.apache.flink.metrics.Gauge 接口的类。返回值的类型没有限制。您可以通过在 MetricGroup 上调用 gauge

Flink Metrics监控指标和性能优化 - 图3

3. Meter 平均值

用来记录一个指标在某个时间段内的平均值。Flink 中的指标有 Task 算子中的 numRecordsInPerSecond,记录此 Task 或者算子每秒接收的记录数。 使用方式：通过 markEvent() 方法注册事件的发生。通过markEvent(long n) 方法注册同时发生的多个事件。

Flink Metrics监控指标和性能优化 - 图4

4. Histogram 直方图

Histogram 用于统计一些数据的分布，比如说 Quantile、Mean、StdDev、Max、Min 等，其中最重要一个是统计算子的延迟。此项指标会记录数据处理的延迟信息，对任务监控起到很重要的作用。 使用方式：通过调用 histogram(String name, Histogram histogram) 来注册一个 MetricGroup

Flink Metrics监控指标和性能优化 - 图5

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 参数来配置标识符的分隔符(默认“.”) 举例说明：以算子的指标组结构为例，其默认为： 
  1.  <host>.taskmanager.<tm_id>.<job_name>.<operator_name>.<subtask_index>
算子的输入记录数指标为： 
  1. 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可视化工具展示出来。原理图如下：

Flink Metrics监控指标和性能优化 - 图6

3 指标性能测试

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

3.1 业务场景介绍

金融风控场景

3.1.1 业务需求:

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

Flink Metrics监控指标和性能优化 - 图7

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 算子生成的时间戳相减，即得到全链路时延。原理图如下：

Flink Metrics监控指标和性能优化 - 图8

由于使用到 Lateny marker,所有需要在 flink-conf.yaml 配置参数 
  1. latency.metrics.interval
系统配置截图如下：

Flink Metrics监控指标和性能优化 - 图9

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

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

3.3 提交任务到Flink on Yarn集群

3.3.1 直接提交 Job 

  1. # -m jobmanager 的地址
  2. # -yjm 1024 指定 jobmanager 的内存信息
  3. # -ytm 1024 指定 taskmanager 的内存信息
  4. bin/flink run \
  5. -t yarn-per-job -yjm 4096 -ytm  8800 -s 96  \
  6. --detached  -c com.threeknowbigdata.datastream.XgboostModelPrediction \
  7. examples/batch/WordCount.jar  \
提交完成后，我们通过 Flink WEBUI 可以看到 job 运行的任务结果如下：

Flink Metrics监控指标和性能优化 - 图10

因为推理模型只是一个 model，存在状态中，所以全链路吞吐考虑的是每秒有多少条推理数据进入 source 算子到倒数第二个算子（最后一个算子只是指标汇总）流出，这个条数就是全链路吞吐。 可以看到在处理 2000W 条数据时，代码直接统计输出的数值和 flink webUI 的统计数值基本一致，所以统计数值是可信的

Flink Metrics监控指标和性能优化 - 图11

Flink WEBUI 跑的结果数据

Flink Metrics监控指标和性能优化 - 图12

打开 Prometheus 在对话框输入全链路时延计算公式 
  1. 计算公式：
  2. avg(flink_taskmanager_job_latency_source_id_
  3. operator_id _operator_subtask_index_latency{
  4. source_id="cbc357ccb763df2852fee8c4fc7d55f2",
  5. operator_id="c9c0ca46716e76f6b700eddf4366d243",quantile="0.999"})

Flink Metrics监控指标和性能优化 - 图13

3.4 优化前性能分析

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

Flink Metrics监控指标和性能优化 - 图14

吞吐指标统计图如下：

Flink Metrics监控指标和性能优化 - 图15

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

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

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

3.5 问题分析

通过 Prometheus分析后，结果如下

Flink Metrics监控指标和性能优化 - 图16

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)； 
  1. //在代码中设置
  3. senv.setBufferTimeout(0);
  1. Buffer 数量的配置如下：
修改flink-conf.yaml 
  1. memory.buffers-per-channel: 2
  2. memory.float-buffers-per-gate: 2
  3. memory.max-buffers-per-channel: 2
配置截图如下：

Flink Metrics监控指标和性能优化 - 图17

3.6 优化后性能分析

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

Flink Metrics监控指标和性能优化 - 图18

吞吐指标统计图如下：

Flink Metrics监控指标和性能优化 - 图19

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

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

3.7 优化前后 LGB 分析总结

如下图所示：

Flink Metrics监控指标和性能优化 - 图20

3.7.1吞吐量—-影响因素：

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

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

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

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