把生命浪费在美好事物上

产品产品产品

358篇博客

Flink系列之Metrics

 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

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