Flink 指标（二）

通过 conf/flink-conf.yaml 文件配置一个或多个 Reporters 来暴露度量值给外部系统，这些 Reporter 将在作业和任务启动的时候实例化。

所有的 Reporter 配置至少需要配置 class 属性，还有一些允许配置记录间隔。下面是一些 Reporter 的配置实例：

包含 Reporter 的 jar 必须放到 /lib 文件夹，这样 Flink 就可以访问到这些 jar。

可以通过继承 org.apache.flink.metrics.reporter.MetricReporter 接口来实现自己的 Reporter，如果需要定期发送记录，需要继承 Scheduled 接口。

下面是一些支持的 Reporter：

不需要添加额外的依赖就可以支持 JMX Reporter，默认是不激活的。

参数：

配置示例：

通过 JMX 公开的度量由域（domain）和键属性列表（key-properties）标识，这些属性一起构成对象名。

域始终以 org.apache.flink 开头，后跟一个通用的度量标识符。与通常的标识符不同，它不受作用域格式的影响，不包含任何变量，并且在跨作业时也是常量。例子： org.apache.flink.job.task.numbytesout 。

键属性列表包含与给定指标关联的所有变量的值，无论配置的作用域格式如何。例子： host=localhost,job_name=myjob,task_name=mytask 。

因此，域标识一个度量类，键属性列表标识该度量的一个（或多个）实例。

要使用此 Reporter，必须复制 /opt/flink-metrics-ganglia-1.6.1-SNAPSHOT.jar 到 Flink 的 /lib 文件夹下。

参数：

配置示例：

要使用此 Reporter，必须复制 /opt/flink-metrics-graphite-1.6.1-SNAPSHOT.jar 到 Flink 的 /lib 文件夹下。

参数：

配置示例：

要使用此 Reporter，必须复制 /opt/flink-metrics-prometheus-1.6.1-SNAPSHOT.jar 到 Flink 的 /lib 文件夹下。

参数：

配置示例：

Flink 度量类型映射到 Prometheus 度量类型，如下所示：

要使用此 Reporter，必须复制 /opt/flink-metrics-prometheus-1.6.1-SNAPSHOT.jar 到 Flink 的 /lib 文件夹下。

参数：

配置示例：

PrometheusPushGatewayReporter 将指标推送到 Pushgateway，可由 Prometheus 抓取。

要使用此 Reporter，必须复制 /opt/flink-metrics-statsd-1.6.1-SNAPSHOT.jar 到 Flink 的 /lib 文件夹下。

参数：

配置示例：

要使用此 Reporter，必须复制 /opt/flink-metrics-datadog-1.6.1-SNAPSHOT.jar 到 Flink 的 /lib 文件夹下。

Flink 指标，如任何变量 <host>， <job_name>， <tm_id>， <subtask_index>， <task_name> 和 <operator_name>，将被发送到 Datadog 作为标签。标签看起来像 host:localhost 和 job_name:myjobname 。

参数：

配置示例：

要使用此 Reporter，必须复制 /opt/flink-metrics-slf4j-1.6.1-SNAPSHOT.jar 到 Flink 的 /lib 文件夹下。

配置示例：

Flink 默认会收集当前状态的指标，下文的表格中包括以下5列：

请注意，“infix” 和 “Metrics” 列中所有的点根据 “metrics.delimiter” 设置变化。

因此，为了推断指标的标识符：

Reference:

https://ci.apache.org/projects/flink/flink-docs-release-1.6/monitoring/metrics.html

flink metric用来对外暴露系统内部的一些运行指标，比如flink框架运行时的JVM相关配置，或者基于flink开发的项目。

flink提供了Counter, Gauge, Histogram and Meter四种类型的指标。我们通过继承RichFunction拿到MetricGroup，并向其中填充指标。

Counter：

用与存储数值类型，比如统计数据输入、输出总数量。

Gauge:

可以用来存储任何类型，前提要实现org.apache.flink.metrics.Gauge接口，重写getValue方法，如果返回类型为Object则该类需要重写toString方法。

有些场景下，需要根据业务计算出指标，则Gauge使用起来更灵活。

Meter：

用来计算平均速率，直接使用其子类MeterView更方便一些。

以flink1.5的Kafka读取以及写入为例，添加rps、dirtyData等相关指标信息。�kafka读取和写入重点是先拿到RuntimeContex初始化指标，并传递给要使用的序列类，通过重写序列化和反序列化方法，来更新指标信息。

这样就可以在监控框架里面看到采集的指标信息了，比如flink_taskmanager_job_task_operator_dtDirtyData指标，dtDirtyData是自己添加的指标，前面的字符串是operator默认使用的metricGroup。

---