Flink Metrics

Posted 2023-02-20

参考技术A Flink 提供了 Metric 系统，允许收集 Metric 并暴露给外部系统。

可以通过任何继承了 RichFunction 的函数访问 Metric 系统。调用 getRuntionContext().getMetricGroup() 方法，该方法返回一个 MetricGroup 对象，可以创建并注册 Metric。

Counter 用来计数。当前值可以使用 inc() / inc(long n) 或 dec() / dec(long n) 进行增减。

Gauge 根据需要提供任何类型的值。需要先创建一个实现 org.apache.flink.metrics.Gauge 的类，返回值的类形没有限制。

Report 程序在暴露数据给外部系统时，会把对象转换为字符串，这意味着需要一个有意义的 toString() 实现。

Histogram 统计值的分布。

Flink 没有提供 Histogram 的默认实现，可以添加依赖使用 DropwizardHistogramWrapper 实现

Meter 用来统计平均吞吐量。

同样添加 flink-metrics-dropwizard 依赖，可以使用 DropwizardMeterWrapper 实现

每个 Metric 都会分配一个标识符和一组键值对，用来报告 Metric。

标识符基于3个组成部分：注册时的用户定义名称、可选的用户定义 Scope 和系统提供的 Scope。例如，如果 A.B 是系统 Scope，C.D 是用户 Scope，E 是名称，那么标识符将是 A.B.C.D.E。

可以通过在 conf/flink-conf.yaml 中设置 metrics.scope.delimiter 键来配置用于标识符的分隔符（默认值：.）。

定义 User Scope 的方法： 调用 MetricGroup#addGroup(String name) ， MetricGroup#addGroup(int name) ， MetricGroup#addGroup(String key, String value) 。这些方法会影响 MetricGroup#getMetricIdentifier 和 MetricGroup#getScopeComponents 的返回值。

System Scope 包含 Metric 的上下文信息，例如注册在哪个 Task（<task_name>）或属于哪个 Job（<job_name>）。

应该包含哪些上下文信息可以通过 conf/flink-conf.yaml 配置。

<host> | <job_name> | <tm_id> | <task_name> | <operator_name> | <subtask_index> 可以作为变量使用。变量的数量或顺序没有限制，区分大小写。

例如：Operator Metric 的默认 Scope 格式为 <host>.taskmanager.<tm_id><job_name><operator_name><subtask_index> ，生成的标识符类似 localhost.taskmanager.1234.MyJob.MyOperator.0.MyMetric 的形式；如果希望包含 Task 名称，并且忽略 TaskManager 信息，可以设置 metrics.scope.operator: <host>.<job_name>.<task_name>.<operator_name>.<subtask_index> ，生成的标识符会变成 localhost.MyJob.MySource_->_MyOperator.MyOperator.0.MyMetric 。

建议添加带有 ID 的变量（如：<job_id>）保证唯一性，避免出现命名冲突的问题。所有可以使用的变量：

Flink 允许向外部系统报告 Metric。

通过在 conf/flink-conf.yaml 中配置一个或多个 Reporter，可以将 Metric 暴露给外部系统。这些 Reporter 在启动时实例化。

Reporter 必须至少配置 class 或 factory.class 属性（使用哪个取决于 Reporter 的实现）。

配置 Reporter 示例

自定义 Reporter：

下面列出了一些支持的 Reporter

org.apache.flink.metrics.jmx.JMXReporter

参数：

通过 JMX 公开的 Metric 由一个 domain 和一组 key 属性组成标识。domain 总是以 org.apache.flink 开始，接一个通用 metric 标识（与一般的 metric 标识不同，不受 scope 格式的影响，不包含任何变量），例如：org.apache.flink.job.task.numBytesOut。

key 属性列表包含与给定 Metric 关联的所有变量的值（不受 scope 格式影响）。例如： host=localhost,job_name=MyJob,task_name=MyTask 。

org.apache.flink.metrics.prometheus.PrometheusReporter

参数：

Flink Metric 类型和 Prometheus Metric 类型映射

org.apache.flink.metrics.prometheus.PrometheusPushGatewayReporter

参数

PrometheusPushGatewayReporter 将 Metric 推到 Pushgateway

默认情况下，Flink 收集的指标

代替 Network/IO 部分 Metrics

如果启用了 Reactive Mode （1.13 MVP 特性），这些 Metric（除 numRestarts）不能正常工作。

如果启用了 Reactive Mode （1.13 MVP 特性），Job Scope 的 Metric 不能正常工作。

Flink 允许跟踪在系统中传输的记录的延迟。默认情况下禁用此功能。要启用延迟跟踪，必须在 Flink 配置（ conf/flink-conf.yaml ）或 ExecutionConfig 中将 latencyTrackingInterval 设置为正数。

Source 会定期（latencyTrackingInterval）发出一个特殊的记录，称为 LatencyMarker。记录包含一个时间戳，该时间戳从记录在源处发出时算起。LatencyMarker 不能超过（overtake）正常记录，因此如果正常记录在 Operator 前排队，将增加标记跟踪的延迟。

延迟监控的粒度，分为以下3档：

需要注意：

Metrics 可以通过 REST API 查询。下面列出一些可用的 Endpoint 和 JSON 返回格式。

Base URL： http://hostname:8081/jobmanager/metrics

查询 Metric 未聚合值

查询 Metric 聚合值

查询 Metric 部分值的聚合值

特殊字符需要转义（符合 URL 标准）

查看 Metric 列表

GET /jobmanager/metrics

请求特定 Metric 的值（未聚合）

GET taskmanagers/<taskmanagerid>/metrics?get=metric1,metric2

请求特定 Metric 的聚合值

GET /taskmanagers/metrics?get=metric1,metric2

请求特定 Metric 的特定值的聚合值

GET /taskmanagers/metrics?get=metric1,metric2&agg=min,max

为 Task 或 Operator 收集的 Metric 也可以在仪表板中可视化。在作业的主页面上，选择 Metrics 选项卡。在 Graph 中选择一个任务后，可以使用 Add Metric 下拉菜单选择要显示的 Metric。

每个 Metric 可以被可视化为一个单独的图形，x轴表示时间，y轴表示测量值。图表每10秒自动更新一次。