把生命浪费在美好事物上

产品产品产品

358篇博客

最新博客

最新资源下载

编辑推荐

网易数帆开源API网关与容器云项目，让云原生生产落地“多快好网易汪源：统一负载与多云环境的“开放姿态”，才是云原生网易数帆如何用 Kubernetes“原语”搞定云原生中间件快手打新挤爆券商系统，网易数帆推出券商稳定性保障方案探索智慧校园新模式，网易有数在教育行业的实践分享金融行业大数据治理之路——数据模型篇

Flink系列之Metrics

把生命浪费在美好事物上2018-08-28 14:51
 Flink是一个针对流数据和批处理的分布式处理引擎，近两年才真正的频繁出现在数据处理领域 。其实Flink在2014年就已经成为ASF（Apache Software Foundation）的顶级项目之一，也许之前是被spark掩盖了光芒，spark在数据处理上的优势不可否认，但是个人经过对spark和flink的源码研读和项目实战后，更偏爱flink一些。在实时计算方面，相对于spark的微批处理（micro batch），flink的数据处理方式更真正称得上流处理，不久前release的spark 2.3目前也已经提供了类似flink流处理的方式，但是目前还没有经过大型互联网公司的实战验证，不知道其线上表现如何，我们可以拭目以待；除了流处理的方式以外，flink在内存管理，网络传输方面也很有其独特之处，另外spark SQL和flink SQL相比，在代码层面上，flink做了简单的封装后直接利用了calcite的API，让SQL变的不再那么的神秘，更便于我们自定义语义，定制我们自己的SQL语句；Flink基于其状态机制提供的CEP（Complex Event Processing）Library可以让我们在流处理过程匹配出我们定义的事件组合。这些我之后都会在flink系列里一一做其原理说明和代码解读。

回到此篇文章的标题：flink-metrics。为什么要以这篇文章作为系列的开头，开头不应该是flink的原理和组件说明此类的文章吗？是的，本应该是从浅入深的去开启flink系列，但是我的考虑有两点：其一，此文章是以试水为目的，想了解一下大家的兴趣点在哪，如果大家大部分是身经百战的流计算实践者，那flink小白入门篇的文章貌似不用再去赘述了。但如果大家对flink很感兴趣，但是对其基本原理和概念不是很了解，那我系列第二篇会补上。其二，目前我自己的项目需要，对metrics看的多了些，想记录一下，也算是做个回顾吧。


Flink  Metrics指任务在flink集群中运行过程中的各项指标，包括机器系统指标：Hostname，CPU，Memory，Thread，GC，NetWork，IO 和 任务运行组件指标：JobManager，TaskManager，Job, Task，Operater相关指标。Flink提供metrics的目的有两点：第一，实时采集metrics的数据供flink UI进行数据展示，用户可以在页面上看到自己提交任务的状态，延迟等信息。第二，对外提供metrics收集接口，用户可以将整个fllink集群的metrics通过MetricsReport上报至第三方系统进行存储，展示和监控。第二种对大型的互联网公司很有用，一般他们的集群规模比较大，不可能通过flink UI进行所有任务的展示，所以就通过metrics上报的方式进行dashboard的展示，同时存储下来的metrics可以用于监控报警，更进一步来说，可以用历史数据进行数据挖掘产生更大的价值。Flink原生的提供了几种主流的第三方上报方式：JMXReporter，GangliaReport，GraphiteReport等，用户可以直接配置使用。


Flink Metrics是通过引入com.codahale.metrics包实现的，它将收集的metrics分为四大类：Counter，Gauge，Histogram和Meter下面分别说明：
  • Counter  计数器 ，用来统计一个metrics的总量。拿flink中的指标来举例，像Task/Operator中的numRecordsIn（此task或者operator接收到的record总量）和numRecordsOut（此task或者operator发送的record总量）就属于Counter。
  • Gauge   指标值  ， 用来记录一个metrics的瞬间值。拿flink中的指标举例，像JobManager或者TaskManager中的JVM.Heap.Used就属于Gauge，记录某个时刻JobManager或者TaskManager所在机器的JVM堆使用量。
  • Histogram  直方图， 有的时候我们不满足于只拿到metrics的总量或者瞬时值，当想得到metrics的最大值，最小值，中位数等信息时，我们就能用到Histogram了。Flink中属于Histogram的指标很少，但是最重要的一个是属于operator的latency。此项指标会记录数据处理的延迟信息，对任务监控起到很重要的作用。
  • Meter  平均值， 用来记录一个metrics某个时间段内平均值。flink中类似指标有task/operator中的numRecordsInPerSecond，字面意思就可以理解，指的是此task或者operator每秒接收的记录数。
Metrics代码解析
那Flink代码中是怎样对metrics进行收集的呢（具体代码在flink-runtime的metrics包里）。下面我们就来按步骤说明：
  1. flink中先会定义好ScopeFormat，scopeFormat定义了各类组件metrics_group的范围，然后各个组件（JobManager，TaskManager，Operator等）都会继承ScopeFormat类进行各自的format实现。
  2. 而后开始定义各个组件的metricsGroup。每个group中定义属于这个组件中所有的metrics。比如TaskIOMetricGroup类，就定义了task执行过程中有关IO的metrics。
  3. 定义好各个metricsGroup后，在初始化各个组件的时候，会将相应的metricsGroup当做参数放入构造函数中进行初始化。我们拿JobManager为例来说：                             class JobManager(protected val flinkConfiguration: Configuration,
        protected val futureExecutor: ScheduledExecutorService,
        protected val ioExecutor: Executor,
        protected val instanceManager: InstanceManager,
        protected val scheduler: FlinkScheduler,
        protected val blobServer: BlobServer,
        protected val libraryCacheManager: BlobLibraryCacheManager,
        protected val archive: ActorRef,
        protected val restartStrategyFactory: RestartStrategyFactory,
        protected val timeout: FiniteDuration,
        protected val leaderElectionService: LeaderElectionService,
        protected val submittedJobGraphs : SubmittedJobGraphStore,
        protected val checkpointRecoveryFactory : CheckpointRecoveryFactory,
        protected val jobRecoveryTimeout: FiniteDuration,
        protected val jobManagerMetricGroup: JobManagerMetricGroup,
        protected val optRestAddress: Option[String])
    初始化JobManager的时候带上了JobManagerMetricGroup，后面此类在preStart()方法中调用了instantiateMetrics(jobManagerMetricGroup)，我们再看instantiateMetrics方法内容：   
  1.  private def instantiateMetrics(jobManagerMetricGroup: MetricGroup) : Unit = {
      jobManagerMetricGroup.gauge[Long, Gauge[Long]]("taskSlotsAvailable", new Gauge[Long] {
        override def getValue: Long = JobManager.this.instanceManager.getNumberOfAvailableSlots
      })
      jobManagerMetricGroup.gauge[Long, Gauge[Long]]("taskSlotsTotal", new Gauge[Long] {
        override def getValue: Long = JobManager.this.instanceManager.getTotalNumberOfSlots
      })
      jobManagerMetricGroup.gauge[Long, Gauge[Long]]("numRegisteredTaskManagers", new Gauge[Long] {
        override def getValue: Long
        = JobManager.this.instanceManager.getNumberOfRegisteredTaskManagers
      })
      jobManagerMetricGroup.gauge[Long, Gauge[Long]]("numRunningJobs", new Gauge[Long] {
        override def getValue: Long = JobManager.this.currentJobs.size
      })
    }   在instantiateMetrics方法内，把相应的metrics都加入到了jobManagerMetricGroup中，这样就建立了metrics和metrics_group的映射关系。
  2.  随后，在各个组件中实例化MetricRegistryImpl，然后利用MetricRegistry的startQueryService方法启动metrics查询服务（本质上是启动相应的Akka Actor）
  3. 最后，利用flink的原生reporter（主要是上文介绍的三种方式）和MetricRegistry建立联系，这样report里就可以拿出所有采集到的metrics，进而将metrics发往第三方系统。
Metrics配置
当我们了解了flink metrics的具体实现步骤后，那就是上手操作了，怎样配置才能让metrics生效呢？接下来就介绍一下配置步骤：
  • flink目录下有一个conf的文件夹，conf下有一个flink-conf.yaml文件，所有的flink有关配置都在这里进行。
  • 配置metrics_scope，metrics_scope指定metrics上报时的组合方式。一共有6个scope需要配置：                                                    metrics.scope.jm      配置JobManager相关metrics，默认格式为 <host>.jobmanager                                                   metrics.scope.jm.job   配置JobManager上Job的相关metrics，默认格式为 <host>.jobmanager.<job_name>
               metrics.scope.tm         配置TaskManager上相关metrics，默认格式为  <host>.taskmanager.<tm_id>     
               metrics.scope.tm.job   配置TaskManager上Job相关metrics，默认格式为 <host>.taskmanager.<tm_id>.<job_name>
               metrics.scope.task   配置Task相关metrics，默认为 <host>.taskmanager.<tm_id>.<job_name>.<task_name>.<subtask_index>
              metrics.scope.operator  配置Operator相关metrics，默认格式 为 <host>.taskmanager.<tm_id>.<job_name>.            <operator_name>.<subtask_index>
          以上6种scope可以根据用户意愿改变组合方式，例如 metrics.scope.operator，我可以改成  <host>.<job_name>.<task_name>.<operator_name>.<subtask_index>，修改后，接收到的operator的metrics就会是如下格式：<host>.<job_name>.<task_name>.<operator_name>.<subtask_index>.xxx =  xxxx（如果全部用默认，则可以不需要在文件里配置，源码里已经指定了默认值）
  • 配置Report，Report相关配置根据其不同的实现类有所不同，我就用项目目前使用的GraphiteReport为例来说明：
          metrics.reporters: grph
          metrics.reporter.grph.class: org.apache.flink.metrics.graphite.GraphiteReporter
          metrics.reporter.grph.host: xxx
          metrics.reporter.grph.port: xxx
          metrics.reporter.grph.protocol: TCP/UDP
      metrics.reporters指定report的名称，metrics.reporter.grph.class指定具体的MetricsReport实现类，metrics.reporter.grph.host指定远端graphite主机ip，metrics.reporter.grph.port指定远端graphite监听端口，metrics.reporter.grph.protocol指定graphite利用的协议。
  • 最后保存文件，重启flink集群即可生效
如果我们不使用flink原生的MetricsReport，想自己实现定制的Report可以吗？答案是肯定的，用户可以参照GraphiteReporter类，自定义类继承 ScheduledDropwizardReporter类，重写report方法即可。我们现在除了利用GraphiteReport，也自己定义了KafkaReport上报定制的metrics来满足更多的用户需求。

总结
Flink的metrics在整个项目里是不可或缺的，社区里Jira也经常有人提出各种improvement，要加入这样那样的metrics，但最终社区接受的很少，因为这些都是对个人公司定制化的代码改动，meger进master的意义不大。合理的利用metrics可以及时的发现集群和任务的状况，从而进行相应的对策，可以保证集群的稳定性，避免不必要的损失。

网易云新用户大礼包：https://www.163yun.com/gift

本文来自网易实践者社区，经作者李涵淼授权发布。


推荐博客