如何进行生产环境作业监控

2024-01-03 22:40:52

声明：本系列博客为原创，最先发表在拉勾教育，其中一部分为免费阅读部分。被读者各种搬运至各大网站。所有其他的来源均为抄袭。

《2021年最新版大数据面试题全面开启更新》

     之前我们介绍如何发现Flink任务是否出现反压，Flink后台页面是发现问题的第一选择，后台页面可以直观、清晰地看到当前作业的运行状态。
     在实际生产中，Flink的后台页面可以方便对Flink JobManager、TaskManager、执行计划、Slot分配、是否反压等参数进行定位，对单个任务来讲可以方便地进行问题排查。
     但是对很多中大型企业来讲，对集群的作业进行管理时，更多的是关心作业精细化实施运行状态。例如，实时吞吐的同比环比、整个集群的任务运行概览、集群水位，或者监控利用Flink实现的ETL框架的运行情况等，这时候需要设计专门的监控系统来监控集群的任务作业情况。

Flink Metrics

     针对上面的情况，可以使用Flink提供的另一个强大的功能：Flink Metrics。
     Flink Metrics是Flink实现的一套运行信息收集库，不但可以手机Flink本身提供的系统指标，比如CPU、内存、线程使用情况、JVM垃圾收集情况、网络和IO等，还可以通过继承和实现指定类或者接口打点手机用户自定义的指标。
     使用Flink Metrics可以做到：

实时采集Flink中的Metrics信息或者自定义用户需要的指标信息并进行展示；
通过Flink提供的Rest API收集这些信息，并且接入第三方系统进行展示。

Flink Metrics分类

Flink提供了四种类型的监控指标，分别是：Counter、Gauge、Histogram、Meter。

Counter

Counter称为计数器，一般用来统计其中一个指标的总量，比如统计数据的输入、输出总量。

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;
  }
}

Gauge

Gauge被用来统计某一个指标的瞬时值。比如，在监控Flink中某一个节点的内存使用情况或者某个map算子的输出值数量。

public class MyMapper extends RichMapFunction<String, String> {
  private transient int valueNumber = 0L;
  @Override
  public void open(Configuration config) {
    getRuntimeContext()
      .getMetricGroup()
      .gauge("MyGauge", new Gauge<Long>() {
        @Override
        public Long getValue() {
          return valueNumber;
        }
      });
  }

  @Override
  public String map(String value) throws Exception {
    valueNumber++;
    return value;
  }
}

Meter

Meter被用来计算一个指标的平均值。

public class MyMapper extends RichMapFunction<Long, Integer> {
  private Meter meter;
  @Override
  public void open(Configuration config) {
    this.meter = getRuntimeContext()
      .getMetricGroup()
      .meter("myMeter", new MyMeter());
  }

  @public Integer map(Long value) throws Exception {
    this.meter.markEvent();
  }
}

Histogram

Histogram是直方图，Flink中属于直方图的指标非常少，通常被用来计算指标的最大值、最小值、中位数等。

public class MyMapper extends RichMapFunction<Long, Integer> {
  private Histogram histogram;
  @Override
  public void open(Configuration config) {
    this.histogram = getRuntimeContext()
      .getMetricGroup()
      .histogram("myHistogram", new MyHistogram());
  }

  @public Integer map(Long value) throws Exception {
    this.histogram.update(value);
  }
}

Flink中的Metrics是一个多层的结构，以Group的方式存在，用来定位唯一的一个Metrics是通过Metric Group + Metric Name的方式。

源码分析

Flink Metrics相关的实现都是通过org.apache.flink.metrics.Metric这个类实现的，整体的类图如下所示：

此外，Flink还提供了向外披露Metric的监控结果的接口，该接口是org.apache.flink.metrics.reporter.MetricReporter。这个接口的实现类通过Metrics的类型进行注册和移除。

public abstract class AbstractReporter implements MetricReporter, CharacterFilter {
    protected final Logger log = LoggerFactory.getLogger(this.getClass());
    protected final Map<Gauge<?>, String> gauges = new HashMap();
    protected final Map<Counter, String> counters = new HashMap();
    protected final Map<Histogram, String> histograms = new HashMap();
    protected final Map<Meter, String> meters = new HashMap();
    public AbstractReporter() {
    }

    public void notifyOfAddedMetric(Metric metric, String metricName, MetricGroup group) {
        String name = group.getMetricIdentifier(metricName, this);
        synchronized(this) {
            if(metric instanceof Counter) {
                this.counters.put((Counter)metric, name);
            } else if(metric instanceof Gauge) {
                this.gauges.put((Gauge)metric, name);
            } else if(metric instanceof Histogram) {
                this.histograms.put((Histogram)metric, name);
            } else if(metric instanceof Meter) {
                this.meters.put((Meter)metric, name);
            } else {
                this.log.warn("Cannot add unknown metric type {}. This indicates that the reporter does not support this metric type.", metric.getClass().getName());
            }
        }
    }

    public void notifyOfRemovedMetric(Metric metric, String metricName, MetricGroup group) {
        synchronized(this) {
            if(metric instanceof Counter) {
                this.counters.remove(metric);
            } else if(metric instanceof Gauge) {
                this.gauges.remove(metric);
            } else if(metric instanceof Histogram) {
                this.histograms.remove(metric);
            } else if(metric instanceof Meter) {
                this.meters.remove(metric);
            } else {
                this.log.warn("Cannot remove unknown metric type {}. This indicates that the reporter does not support this metric type.", metric.getClass().getName());
            }
        }
    }
}

获取Metrics

获取Metrics的方法有多种，首先可以通过Flink的后台管理页面看到部分指标；其次可以通过Flink提供的Http接口查询Flink任务的状态信息，因为Flink Http接口返回的都是Json信息，可以很方便地将Json解析；最后一种方法是可以通过Metric Reporter获取。

Flink HTTP接口

Flink 提供了丰富的接口来协助查询Flink中任务运行的状态，所有的请求都可以通过访问http://localhost:8081/ 加指定的URI方式查询，Flink支持的所有HTTP接口可以在这里进行查询到。
Flink支持的接口包括：

/config
/overview
/jobs
/joboverview/running
/joboverview/completed
/jobs/<jobid>
/jobs/<jobid>/vertices
/jobs/<jobid>/config
/jobs/<jobid>/exceptions
/jobs/<jobid>/accumulators
/jobs/<jobid>/vertices/<vertexid>
/jobs/<jobid>/vertices/<vertexid>/subtasktimes
/jobs/<jobid>/vertices/<vertexid>/taskmanagers
/jobs/<jobid>/vertices/<vertexid>/accumulators
/jobs/<jobid>/vertices/<vertexid>/subtasks/accumulators
/jobs/<jobid>/vertices/<vertexid>/subtasks/<subtasknum>
/jobs/<jobid>/vertices/<vertexid>/subtasks/<subtasknum>/attempts/<attempt>
/jobs/<jobid>/vertices/<vertexid>/subtasks/<subtasknum>/attempts/<attempt>/accumulators
/jobs/<jobid>/plan
/jars/upload
/jars
/jars/:jarid
/jars/:jarid/plan
/jars/:jarid/run

比如，可以通过查询/joboverview访问集群中所有任务的概率，结果类似如下：

{
  "running":[],
  "finished":[
    {
      "jid": "7684be6004e4e955c2a558a9bc463f65",
      "name": "Flink Java Job at Wed Sep 16 18:08:21 CEST 2015",
      "state": "FINISHED",
      "start-time": 1442419702857,
      "end-time": 1442419975312,
      "duration":272455,
      "last-modification": 1442419975312,
      "tasks": {
         "total": 6,
         "pending": 0,
         "running": 0,
         "finished": 6,
         "canceling": 0,
         "canceled": 0,
         "failed": 0
      }
    },
    {
      "jid": "49306f94d0920216b636e8dd503a6409",
      "name": "Flink Java Job at Wed Sep 16 18:16:39 CEST 2015",
      ...
    }]
}

Flink Reporter

     Flink还提供了很多内置的Reporter，这些Reporter在Flink的官网中可以查询到。
     比如Flink提供了Graphite、InfluxDB、Prometheus等内置的Reporter，可以方便地对这些外部系统进行集成。
     比如Flink和InfluxDB、Grafana集成进行Flink集群任务监控的案例。InfluxDB扮演监控数据存储的角色，Grafana扮演数据展示者的角色。

InfluxDB的安装
修改InfluxDB的配置/etc/influxdb/influxdb.conf

     [admin]
     enabled = true
     bind-address = ":8083"

Grafana的安装
安装文档，Grafana默认的账号密码是admin、admin，通过3000端口进行访问。
修改flink-conf.yaml
在flink的配置文件中新增以下配置：

metrics.reporter.influxdb.class: org.apache.flink.metrics.influxdb.InfluxdbReporter
metrics.reporter.influxdb.host: xxx.xxx.xxx.xxx
metrics.reporter.influxdb.port: 8086
metrics.reporter.influxdb.db: flink

同时将flink-metrics-influxdb-1.10.1.jar这个包复制到flink的/lib目录下，然后启动flink，就可以在Grafana中看到Metrics信息来。

码农公寓

《2021年最新版大数据面试题全面开启更新》

Flink Metrics

相关文章