Spark的度量系統(tǒng)有以下幾部分掉瞳，也可以參照MetricsSystem類的注釋部分

• Instance: 數(shù)據(jù)實(shí)例欧引。Spark的Instance有Master契耿、Worker定嗓、ApplicationInfo症昏、StreamingContext等课蔬，主要用來提供Source數(shù)據(jù)何暇、啟停MetricsSystem
• Source: 度量數(shù)據(jù)輸入源句柠。Source采集的數(shù)據(jù)來源于Instance實(shí)例屬性
• Sink: 度量數(shù)據(jù)輸出源咸包。Spark使用MetricsServlet作為默認(rèn)Sink
• MetricsConfig: 度量需要的配置信息桃序。initialize()方法初始化properties
• MetricsSystem: instance粒度的Source、Sink控制中心

Source

Spark將度量數(shù)據(jù)來源抽象為Source接口诉儒。提供了ApplicationSource葡缰、MasterSource、WorkerSource忱反、DAGSchedulerSource泛释、StreamingSource、JvmSource等實(shí)現(xiàn)

private[spark] trait Source {
  def sourceName: String
  def metricRegistry: MetricRegistry
}

• sourceName: 度量源名稱
• metricRegistry: 度量源注冊對象

具體分析下MasterSource温算、WorkerSource怜校、JvmSource輸入源

private[spark] class MasterSource(val master: Master) extends Source {
  override val metricRegistry = new MetricRegistry()
  override val sourceName = "master"

  // Gauge for worker numbers in cluster
  metricRegistry.register(MetricRegistry.name("workers"), new Gauge[Int] {
    override def getValue: Int = master.workers.size
  })

  // Gauge for alive worker numbers in cluster
  metricRegistry.register(MetricRegistry.name("aliveWorkers"), new Gauge[Int]{
    override def getValue: Int = master.workers.count(_.state == WorkerState.ALIVE)
  })

  // Gauge for application numbers in cluster
  metricRegistry.register(MetricRegistry.name("apps"), new Gauge[Int] {
    override def getValue: Int = master.apps.size
  })

  // Gauge for waiting application numbers in cluster
  metricRegistry.register(MetricRegistry.name("waitingApps"), new Gauge[Int] {
    override def getValue: Int = master.apps.count(_.state == ApplicationState.WAITING)
  })
}

private[worker] class WorkerSource(val worker: Worker) extends Source {
  override val sourceName = "worker"
  override val metricRegistry = new MetricRegistry()

  metricRegistry.register(MetricRegistry.name("executors"), new Gauge[Int] {
    override def getValue: Int = worker.executors.size
  })

  // Gauge for cores used of this worker
  metricRegistry.register(MetricRegistry.name("coresUsed"), new Gauge[Int] {
    override def getValue: Int = worker.coresUsed
  })

  // Gauge for memory used of this worker
  metricRegistry.register(MetricRegistry.name("memUsed_MB"), new Gauge[Int] {
    override def getValue: Int = worker.memoryUsed
  })

  // Gauge for cores free of this worker
  metricRegistry.register(MetricRegistry.name("coresFree"), new Gauge[Int] {
    override def getValue: Int = worker.coresFree
  })

  // Gauge for memory free of this worker
  metricRegistry.register(MetricRegistry.name("memFree_MB"), new Gauge[Int] {
    override def getValue: Int = worker.memoryFree
  })
}

MetricRegistry的Gauge統(tǒng)計(jì)數(shù)據(jù)來源于Master、Worker對象的字段屬性

JvmSource的MetricSet來源于metrics-jvm包的實(shí)現(xiàn)

private[spark] class JvmSource extends Source {
  override val sourceName = "jvm"
  override val metricRegistry = new MetricRegistry()

  metricRegistry.registerAll(new GarbageCollectorMetricSet)
  metricRegistry.registerAll(new MemoryUsageGaugeSet)
  metricRegistry.registerAll(
    new BufferPoolMetricSet(ManagementFactory.getPlatformMBeanServer))
}

Source用來創(chuàng)建MetricRegistry對象注竿，并register需要統(tǒng)計(jì)的數(shù)據(jù)指標(biāo)茄茁，指標(biāo)來源于Instance實(shí)例對象屬性

Sink

Spark將度量數(shù)據(jù)統(tǒng)計(jì)輸出源抽象為Sink接口。提供了ConsoleSink巩割、CsvSink裙顽、MetricsServlet、GraphiteSink宣谈、JmxSink愈犹、Slf4jSink等實(shí)現(xiàn)

private[spark] trait Sink {
  def start(): Unit
  def stop(): Unit
  def report(): Unit
}

• MetricsServlet: 在Spark UI的jetty服務(wù)中創(chuàng)建ServletContextHandler，將度量數(shù)據(jù)統(tǒng)計(jì)展示在瀏覽器

具體分析下Slf4jSink實(shí)現(xiàn)

private[spark] class Slf4jSink(
    val property: Properties,
    val registry: MetricRegistry,
    securityMgr: SecurityManager)
  extends Sink {
  val SLF4J_DEFAULT_PERIOD = 10
  val SLF4J_DEFAULT_UNIT = "SECONDS"

  val SLF4J_KEY_PERIOD = "period"
  val SLF4J_KEY_UNIT = "unit"

  val pollPeriod = Option(property.getProperty(SLF4J_KEY_PERIOD)) match {
    case Some(s) => s.toInt
    case None => SLF4J_DEFAULT_PERIOD
  }

  val pollUnit: TimeUnit = Option(property.getProperty(SLF4J_KEY_UNIT)) match {
    case Some(s) => TimeUnit.valueOf(s.toUpperCase(Locale.ROOT))
    case None => TimeUnit.valueOf(SLF4J_DEFAULT_UNIT)
  }
  // 檢查scheduleAtFixedRate周期時(shí)間最短1s
  MetricsSystem.checkMinimalPollingPeriod(pollUnit, pollPeriod)

  val reporter: Slf4jReporter = Slf4jReporter.forRegistry(registry)
    .convertDurationsTo(TimeUnit.MILLISECONDS)
    .convertRatesTo(TimeUnit.SECONDS)
    .build()

  override def start() {
    reporter.start(pollPeriod, pollUnit)
  }

  override def stop() {
    reporter.stop()
  }

  override def report() {
    reporter.report()
  }
}

主要看start()方法，需要一個(gè)reporter對象漩怎，以及數(shù)據(jù)產(chǎn)生的周期時(shí)間pollPeriod勋颖、pollUnit。start再調(diào)用ScheduledReporter.start()

public void start(long period, TimeUnit unit) {
    // executor對象是Executors.newSingleThreadScheduledExecutor實(shí)現(xiàn)
    executor.scheduleAtFixedRate(new Runnable() {
        @Override
        public void run() {
            try {
                report();
            } catch (Exception ex) {
                LOG.error("Exception thrown from {}#report. Exception was suppressed.", ScheduledReporter.this.getClass().getSimpleName(), ex);
            }
        }
    }, period, period, unit);
}

Sink需要?jiǎng)?chuàng)建reporter勋锤、pollPeriod饭玲、pollUnit。周期性獲取Source數(shù)據(jù)并reporter

MetricsConfig

讀取Metrics相關(guān)的配置信息

private[spark] class MetricsConfig(conf: SparkConf) extends Logging {

  private val DEFAULT_PREFIX = "*"
  private val INSTANCE_REGEX = "^(\\*|[a-zA-Z]+)\\.(.+)".r
  private val DEFAULT_METRICS_CONF_FILENAME = "metrics.properties"

  private[metrics] val properties = new Properties()
  private[metrics] var perInstanceSubProperties: mutable.HashMap[String, Properties] = null
  
  // 設(shè)置default Properties屬性
  private def setDefaultProperties(prop: Properties) {
    prop.setProperty("*.sink.servlet.class", "org.apache.spark.metrics.sink.MetricsServlet")
    prop.setProperty("*.sink.servlet.path", "/metrics/json")
    prop.setProperty("master.sink.servlet.path", "/metrics/master/json")
    prop.setProperty("applications.sink.servlet.path", "/metrics/applications/json")
  }

  /**
   * Load properties from various places, based on precedence
   * If the same property is set again latter on in the method, it overwrites the previous value
   */
  // 入口方法叁执，加載配置信息
  def initialize() {
    // Add default properties in case there's no properties file
    setDefaultProperties(properties)

    loadPropertiesFromFile(conf.getOption("spark.metrics.conf"))

    // Also look for the properties in provided Spark configuration
    val prefix = "spark.metrics.conf."
    conf.getAll.foreach {
      case (k, v) if k.startsWith(prefix) =>
        properties.setProperty(k.substring(prefix.length()), v)
      case _ =>
    }

    // Now, let's populate a list of sub-properties per instance, instance being the prefix that
    // appears before the first dot in the property name.
    // Add to the sub-properties per instance, the default properties (those with prefix "*"), if
    // they don't have that exact same sub-property already defined.
    //
    // For example, if properties has ("*.class"->"default_class", "*.path"->"default_path",
    // "driver.path"->"driver_path"), for driver specific sub-properties, we'd like the output to be
    // ("driver"->Map("path"->"driver_path", "class"->"default_class")
    // Note how class got added to based on the default property, but path remained the same
    // since "driver.path" already existed and took precedence over "*.path"
    perInstanceSubProperties = subProperties(properties, INSTANCE_REGEX)
    if (perInstanceSubProperties.contains(DEFAULT_PREFIX)) {
      val defaultSubProperties = perInstanceSubProperties(DEFAULT_PREFIX).asScala
      for ((instance, prop) <- perInstanceSubProperties if (instance != DEFAULT_PREFIX);
           (k, v) <- defaultSubProperties if (prop.get(k) == null)) {
        prop.put(k, v)
      }
    }
  }

  /**
   * Take a simple set of properties and a regex that the instance names (part before the first dot)
   * have to conform to. And, return a map of the first order prefix (before the first dot) to the
   * sub-properties under that prefix.
   *
   * For example, if the properties sent were Properties("*.sink.servlet.class"->"class1",
   * "*.sink.servlet.path"->"path1"), the returned map would be
   * Map("*" -> Properties("sink.servlet.class" -> "class1", "sink.servlet.path" -> "path1"))
   * Note in the subProperties (value of the returned Map), only the suffixes are used as property
   * keys.
   * If, in the passed properties, there is only one property with a given prefix, it is still
   * "unflattened". For example, if the input was Properties("*.sink.servlet.class" -> "class1"
   * the returned Map would contain one key-value pair
   * Map("*" -> Properties("sink.servlet.class" -> "class1"))
   * Any passed in properties, not complying with the regex are ignored.
   *
   * @param prop the flat list of properties to "unflatten" based on prefixes
   * @param regex the regex that the prefix has to comply with
   * @return an unflatted map, mapping prefix with sub-properties under that prefix
   */
   // 參考下面圖片示例
  def subProperties(prop: Properties, regex: Regex): mutable.HashMap[String, Properties] = {
    val subProperties = new mutable.HashMap[String, Properties]
    prop.asScala.foreach { kv =>
      if (regex.findPrefixOf(kv._1.toString).isDefined) {
        val regex(prefix, suffix) = kv._1.toString
        subProperties.getOrElseUpdate(prefix, new Properties).setProperty(suffix, kv._2.toString)
      }
    }
    subProperties
  }

  // 當(dāng)key不存在時(shí)茄厘，獲取*對應(yīng)的properties屬性
  def getInstance(inst: String): Properties = {
    perInstanceSubProperties.get(inst) match {
      case Some(s) => s
      case None => perInstanceSubProperties.getOrElse(DEFAULT_PREFIX, new Properties)
    }
  }

  /**
   * Loads configuration from a config file. If no config file is provided, try to get file
   * in class path.
   */
  private[this] def loadPropertiesFromFile(path: Option[String]): Unit = {
    var is: InputStream = null
    try {
      is = path match {
        // 標(biāo)準(zhǔn)寫法。path存在時(shí)FileInputStream讀韧搅怠蚕断；不存在讀取項(xiàng)目中的metrics.properties文件，通過classloader加載: Utils.getSparkClassLoader.getResourceAsStream
        case Some(f) => new FileInputStream(f)
        case None => Utils.getSparkClassLoader.getResourceAsStream(DEFAULT_METRICS_CONF_FILENAME)
      }

      if (is != null) {
        // load進(jìn)properties里
        properties.load(is)
      }
    } catch {
      case e: Exception =>
        val file = path.getOrElse(DEFAULT_METRICS_CONF_FILENAME)
        logError(s"Error loading configuration file $file", e)
    } finally {
      if (is != null) {
        // 切記必須close
        is.close()
      }
    }
  }
}

subprop.png

MetricsSystem

負(fù)責(zé)register Sources入挣、Sinks，并start sinks硝拧。MetricsSystem不是系統(tǒng)的控制中心径筏，而是每個(gè)instance一個(gè)MetricsSystem對象，負(fù)責(zé)instance粒度的控制

MetricsSystem類三個(gè)核心方法: registerSources()障陶、registerSinks()滋恬、sinks.foreach(_.start)

private[spark] class MetricsSystem private (
    val instance: String,
    conf: SparkConf,
    securityMgr: SecurityManager)
  extends Logging {
  // 構(gòu)造MetricsConfig對象，用于讀取配置信息
  private[this] val metricsConfig = new MetricsConfig(conf)

  private val sinks = new mutable.ArrayBuffer[Sink]
  private val sources = new mutable.ArrayBuffer[Source]
  private val registry = new MetricRegistry()

  private var running: Boolean = false

  // Treat MetricsServlet as a special sink as it should be exposed to add handlers to web ui
  private var metricsServlet: Option[MetricsServlet] = None

  /**
   * Get any UI handlers used by this metrics system; can only be called after start().
   */
  def getServletHandlers: Array[ServletContextHandler] = {
    require(running, "Can only call getServletHandlers on a running MetricsSystem")
    metricsServlet.map(_.getHandlers(conf)).getOrElse(Array())
  }
  // MetricsConfig對象初始化
  metricsConfig.initialize()

  def start() {
    require(!running, "Attempting to start a MetricsSystem that is already running")
    running = true
    // 注冊StaticSources抱究，也就是CodegenMetrics恢氯、HiveCatalogMetrics
    StaticSources.allSources.foreach(registerSource)
    // 注冊Sources
    registerSources()
    // 獲取Sinks
    registerSinks()
    // 啟動(dòng)Sinks
    sinks.foreach(_.start)
  }

  def stop() {
    if (running) {
      // foreach調(diào)用Sinks的stop方法
      sinks.foreach(_.stop)
    } else {
      logWarning("Stopping a MetricsSystem that is not running")
    }
    running = false
  }

  def report() {
    // foreach調(diào)用Sinks的report方法
    sinks.foreach(_.report())
  }

  /**
   * Build a name that uniquely identifies each metric source.
   * The name is structured as follows: <app ID>.<executor ID (or "driver")>.<source name>.
   * If either ID is not available, this defaults to just using <source name>.
   *
   * @param source Metric source to be named by this method.
   * @return An unique metric name for each combination of
   *         application, executor/driver and metric source.
   */
  // 構(gòu)建registry name
  private[spark] def buildRegistryName(source: Source): String = {
    val metricsNamespace = conf.get(METRICS_NAMESPACE).orElse(conf.getOption("spark.app.id"))

    val executorId = conf.getOption("spark.executor.id")
    val defaultName = MetricRegistry.name(source.sourceName)

    if (instance == "driver" || instance == "executor") {
      if (metricsNamespace.isDefined && executorId.isDefined) {
        // 當(dāng)instance是driver或executor時(shí)，name的元素構(gòu)成
        // {{conf.getOption("spark.app.id")}}.{{conf.getOption("spark.executor.id")}}.{{source.sourceName}}
        MetricRegistry.name(metricsNamespace.get, executorId.get, source.sourceName)
      } else {
        // Only Driver and Executor set spark.app.id and spark.executor.id.
        // Other instance types, e.g. Master and Worker, are not related to a specific application.
        if (metricsNamespace.isEmpty) {
          logWarning(s"Using default name $defaultName for source because neither " +
            s"${METRICS_NAMESPACE.key} nor spark.app.id is set.")
        }
        if (executorId.isEmpty) {
          logWarning(s"Using default name $defaultName for source because spark.executor.id is " +
            s"not set.")
        }
        defaultName
      }
    } else { defaultName }
  }

  def getSourcesByName(sourceName: String): Seq[Source] =
    sources.filter(_.sourceName == sourceName)

  // 注冊單個(gè)source
  def registerSource(source: Source) {
    sources += source
    try {
      val regName = buildRegistryName(source)
      registry.register(regName, source.metricRegistry)
    } catch {
      case e: IllegalArgumentException => logInfo("Metrics already registered", e)
    }
  }
  
  // 刪除source
  def removeSource(source: Source) {
    sources -= source
    val regName = buildRegistryName(source)
    registry.removeMatching(new MetricFilter {
      def matches(name: String, metric: Metric): Boolean = name.startsWith(regName)
    })
  }

  // 注冊所有以source開頭的數(shù)據(jù)源
  private def registerSources() {
    val instConfig = metricsConfig.getInstance(instance)
    // MetricsSystem.SOURCE_REGEX: "^source\\.(.+)\\.(.+)".r
    val sourceConfigs = metricsConfig.subProperties(instConfig, MetricsSystem.SOURCE_REGEX)

    // Register all the sources related to instance
    sourceConfigs.foreach { kv =>
      val classPath = kv._2.getProperty("class")
      try {
        // 反射對象鼓寺。這里只能反射無參數(shù)的Source對象勋拟，比如JvmSource
        val source = Utils.classForName(classPath).newInstance()
        registerSource(source.asInstanceOf[Source])
      } catch {
        case e: Exception => logError("Source class " + classPath + " cannot be instantiated", e)
      }
    }
  }

  // 獲取以sink開頭的Sinks
  private def registerSinks() {
    val instConfig = metricsConfig.getInstance(instance)
    // 以sink開頭的屬性配置: "^sink\\.(.+)\\.(.+)".r
    val sinkConfigs = metricsConfig.subProperties(instConfig, MetricsSystem.SINK_REGEX)

    sinkConfigs.foreach { kv =>
      val classPath = kv._2.getProperty("class")
      if (null != classPath) {
        try {
          // 傳入構(gòu)造函數(shù)參數(shù)值創(chuàng)建sink對象: kv._2, registry, securityMgr
          val sink = Utils.classForName(classPath)
            .getConstructor(classOf[Properties], classOf[MetricRegistry], classOf[SecurityManager])
            .newInstance(kv._2, registry, securityMgr)
          if (kv._1 == "servlet") {
            // key是servlet時(shí)，轉(zhuǎn)換成MetricsServlet對象
            metricsServlet = Some(sink.asInstanceOf[MetricsServlet])
          } else {
            // 否則添加到sinks列表
            sinks += sink.asInstanceOf[Sink]
          }
        } catch {
          case e: Exception =>
            logError("Sink class " + classPath + " cannot be instantiated")
            throw e
        }
      }
    }
  }
}

總結(jié)

先看下metrics.properties.template模板

*.sink.jmx.class=org.apache.spark.metrics.sink.JmxSink

*.sink.console.class=org.apache.spark.metrics.sink.ConsoleSink

*.sink.statsd.class=org.apache.spark.metrics.sink.StatsdSink
*.sink.statsd.prefix=spark

*.sink.console.period=10
*.sink.console.unit=seconds

master.sink.console.period=15
master.sink.console.unit=seconds

*.sink.csv.class=org.apache.spark.metrics.sink.CsvSink

*.sink.csv.period=1
*.sink.csv.unit=minutes

*.sink.csv.directory=/tmp/

worker.sink.csv.period=10
worker.sink.csv.unit=minutes

*.sink.slf4j.class=org.apache.spark.metrics.sink.Slf4jSink

*.sink.slf4j.period=1
*.sink.slf4j.unit=minutes

master.source.jvm.class=org.apache.spark.metrics.source.JvmSource

worker.source.jvm.class=org.apache.spark.metrics.source.JvmSource

driver.source.jvm.class=org.apache.spark.metrics.source.JvmSource

executor.source.jvm.class=org.apache.spark.metrics.source.JvmSource

1. 先讀取properties配置信息
2. 根據(jù)instance name妈候，獲取${name}開頭的敢靡，不存在時(shí)讀取*開頭的屬性值，生成instConfig對象
3. instConfig基礎(chǔ)上再分別獲取source或sink開頭的sourceConfigs對象
4. sourceConfigs獲取source class苦银、sink class反射對象啸胧。source反射時(shí)調(diào)用默認(rèn)的無參構(gòu)造函數(shù)，只能反射比如JvmSource幔虏，對于MasterSource需要在Master類里new出來纺念；sink反射時(shí)傳入了構(gòu)造函數(shù)參數(shù)值，參數(shù)也是從prop里讀取的
5. register sources想括、sinks start
6. MetricsSystem負(fù)責(zé)source陷谱、sink的啟停，而每個(gè)instance單獨(dú)啟停自身的metrics