Flink中如何实现一个自定义MetricReporter

2021-11-20 16:03:59 阅读：169 来源： 互联网

标签：flink 自定义 reporter Metric Flink metrics MetricReporter my

什么是 Metrics

在 flink 任务运行的过程中，用户通常想知道任务运行的一些基本指标，比如吞吐量、内存和 cpu 使用情况、checkpoint 稳定性等等。而通过 flink metrics 这些指标都可以轻而易举地获取到，避免任务的运行处于黑盒状态，通过分析这些指标，可以更好的调整任务的资源、定位遇到的问题、对任务进行监控。接下来本文将介绍 flink metrics 的一些基本概念与原理以及实践。

Flink 对于指标监测有一套自己的实现，同时 flink 自身系统有一些固定的 metric 数据，　包括系统的一些指标，CPU,内存, IO 　或者各个 task 运行的一些指标。指标的统计方式有四种，这些指标都实现了 Metric 这个接口，而 Metric 这个接口只是一个标识，本身并没有定义如何方法接口，部分子类的继承关系如下所示。

从图中可以看出，Metric 这个接口有四个直接子类，分别是：

Gauge —— 最简单的度量指标，只是简单的返回一个值，比如返回一个队列中当前元素的个数；
Counter —— 计数器，在一些情况下，会比 Gauge 高效，比如通过一个 AtomicLong 变量来统计一个队列的长度；
Meter —— 吞吐量的度量，也就是一系列事件发生的速率，例如 TPS；
Histogram —— 度量值的统计结果，如最大值、最小值、平均值，以及分布情况等。

Metrics 使用

下面以 Counter 为例，说明 Metric 的具体用法，Counters 通常用来计数，可以通过 inc 或 dec 方法来对计数值进行增加或减少。

public class MyMapper extends RichMapFunction<String, String> {
  private transient Counter counter;

  @Override
  public void open(Configuration config) {
    this.counter = getRuntimeContext()
      .getMetricGroup()
      .counter("myCounter");
  }

  @Override
  public String map(String value) throws Exception {
    this.counter.inc();
    return value;
  }
}

Metric Reporter

获取 Metrics 有三种方法，首先可以在 WebUI 上看到；其次可以通过 RESTful API 获取，RESTful API 对程序比较友好，比如写自动化脚本或程序，自动化运维和测试，通过 RESTful API 解析返回的 Json 格式对程序比较友好；最后，还可以通过 Metric Reporter 获取，监控主要使用 Metric Reporter 功能。

flink 提供了很多外部监控系统的支持：JMX(java 自带的技术，不严格属于第三方)、Graphite、InfluxDB、Prometheus、StatsD、Datadog、Slf4j（直接打 log 里）等，也可以通过实现 org.apache.flink.metrics.reporter.MetricReporter 接口来编写自己的 Reporter。如果想要定期发送报告，可以实现 Scheduled 接口。

Metric Reporter 是如何配置的？首先 Metrics Reporters 的名字用逗号分隔，然后通过 metrics.reporter.jmx.class 的 classname 反射找 reporter，还需要拿到 metrics.reporter.jmx.port 的配置，比如向第三方系统通过网络发送的比较多，但要知道往哪里发，ip 地址、port 信息是比较常见的。

vim flink/conf/flink-conf.yaml

metrics.reporters: my_jmx_reporter,my_other_reporter

metrics.reporter.my_jmx_reporter.factory.class: org.apache.flink.metrics.jmx.JMXReporterFactory
metrics.reporter.my_jmx_reporter.port: 9020-9040
metrics.reporter.my_jmx_reporter.scope.variables.excludes:job_id;task_attempt_num

metrics.reporter.my_other_reporter.class: org.apache.flink.metrics.graphite.GraphiteReporter
metrics.reporter.my_other_reporter.host: 192.168.1.1
metrics.reporter.my_other_reporter.port: 10000
metrics.reporter.my_other_reporter.interval: 60 SECONDS

开发者可以实现自己的 reporter，将 metrics 数据导出到不同的系统。

实现 MetricReporter 类中的 open,close, notifyOfAddedMetric, notifyOfRemovedMetric 方法。
实现 Scheduled 的 report 方法，表示其需要被定期调度执行，在该方法中实现写入到其他系统的逻辑。

自定义 Metric Reporter

MetricReporter 是用来向外暴露 Metric 的监测结果的接口。由于 MetricReporter 的子类在实例化时，都是通过反射机制，所以对于其实现子类，需要有一个公共、无参的构造函数，这个接口的定义如下：

public interface MetricReporter {
   void open(MetricConfig config);
   void close();
   void notifyOfAddedMetric(Metric metric, String metricName, MetricGroup group);
   void notifyOfRemovedMetric(Metric metric, String metricName, MetricGroup group);
}

open —— 由于子类都是用无参构造函数，通过反射进行实例化，所以相关初始化的工作都是放在这里进行的，并且这个方法需要在实例化后，就需要调用该方法进行相关初始化的工作；
close —— 这里就是在关闭时，进行资源回收等相关操作的；
notifyOfAddedMetric —— 当有一个新的 Metric 注册时，会调用该方法来通知 MetricReporter；
notifyOfRemovedMetric —— 当有一个 Metric 被移除时，通过这个方法来通知 MetricReporter；

关注 gzh “HEY DATA” 后台回复关键字 MetricReporter 可获得自定义 MetricReporter 实现例子文件。

标签：flink,自定义,reporter,Metric,Flink,metrics,MetricReporter,my
来源： https://blog.csdn.net/weixin_40278610/article/details/121440853

本站声明： 1. iCode9 技术分享网（下文简称本站）提供的所有内容，仅供技术学习、探讨和分享；
2. 关于本站的所有留言、评论、转载及引用，纯属内容发起人的个人观点，与本站观点和立场无关；
3. 关于本站的所有言论和文字，纯属内容发起人的个人观点，与本站观点和立场无关；
4. 本站文章均是网友提供，不完全保证技术分享内容的完整性、准确性、时效性、风险性和版权归属；如您发现该文章侵犯了您的权益，可联系我们第一时间进行删除；
5. 本站为非盈利性的个人网站，所有内容不会用来进行牟利，也不会利用任何形式的广告来间接获益，纯粹是为了广大技术爱好者提供技术内容和技术思想的分享性交流网站。

ICode9

Flink中如何实现一个自定义MetricReporter

什么是 Metrics

Metrics 使用

Metric Reporter

自定义 Metric Reporter