SPARK中metrics是怎么传递的

case class FilterExec(condition: Expression, child: SparkPlan)
  extends UnaryExecNode with CodegenSupport with GeneratePredicateHelper {
  ...
  override lazy val metrics = Map(
    "numOutputRows" -> SQLMetrics.createMetric(sparkContext, "number of output rows"))
  ...
  protected override def doExecute(): RDD[InternalRow] = {
    val numOutputRows = longMetric("numOutputRows")
    child.execute().mapPartitionsWithIndexInternal { (index, iter) =>
      val predicate = Predicate.create(condition, child.output)
      predicate.initialize(0)
      iter.filter { row =>
        val r = predicate.eval(row)
        if (r) numOutputRows += 1
        r
      }
    }
  }

 def createMetric(sc: SparkContext, name: String): SQLMetric = {
    val acc = new SQLMetric(SUM_METRIC)
    acc.register(sc, name = metricsCache.get(name), countFailedValues = false)
    acc
  }
 ...
 abstract class AccumulatorV2[IN, OUT] extends Serializable {

 private[spark] def register(
    sc: SparkContext,
    name: Option[String] = None,
    countFailedValues: Boolean = false): Unit = {
  if (this.metadata != null) {
    throw new IllegalStateException("Cannot register an Accumulator twice.")
  }
  this.metadata = AccumulatorMetadata(AccumulatorContext.newId(), name, countFailedValues)
  AccumulatorContext.register(this)
  sc.cleaner.foreach(_.registerAccumulatorForCleanup(this))
}

 private def readObject(in: ObjectInputStream): Unit = Utils.tryOrIOException {
  in.defaultReadObject()
  if (atDriverSide) {
    atDriverSide = false
    // Automatically register the accumulator when it is deserialized with the task closure.
    // This is for external accumulators and internal ones that do not represent task level
    // metrics, e.g. internal SQL metrics, which are per-operator.
    val taskContext = TaskContext.get()
    if (taskContext != null) {
      taskContext.registerAccumulator(this)
    }
  } else {
    atDriverSide = true
  }
}

private def updateAccumulators(event: CompletionEvent): Unit = {
val task = event.task
val stage = stageIdToStage(task.stageId)
event.accumUpdates.foreach { updates =>
  val id = updates.id
  try {
  // Find the corresponding accumulator on the driver and update it
  val acc: AccumulatorV2[Any, Any] = AccumulatorContext.get(id) match {
    case Some(accum) => accum.asInstanceOf[AccumulatorV2[Any, Any]]
    case None =>
      throw SparkCoreErrors.accessNonExistentAccumulatorError(id)
  }
  acc.merge(updates.asInstanceOf[AccumulatorV2[Any, Any]])
  // To avoid UI cruft, ignore cases where value wasn't updated
  if (acc.name.isDefined && !updates.isZero) {
    stage.latestInfo.accumulables(id) = acc.toInfo(None, Some(acc.value))
    event.taskInfo.setAccumulables(
      acc.toInfo(Some(updates.value), Some(acc.value)) +: event.taskInfo.accumulables)
  }

   private def reportHeartBeat(): Unit = {
    ...
    val accumulatorsToReport =
      if (HEARTBEAT_DROP_ZEROES) {
        taskRunner.task.metrics.accumulators().filterNot(_.isZero)
      } else {
        taskRunner.task.metrics.accumulators()
      }
    accumUpdates += ((taskRunner.taskId, accumulatorsToReport))
    }
    ... 
     val message = Heartbeat(executorId, accumUpdates.toArray, env.blockManager.blockManagerId,
  executorUpdates)
  try {
  val response = heartbeatReceiverRef.askSync[HeartbeatResponse](
    message, new RpcTimeout(HEARTBEAT_INTERVAL_MS.millis, EXECUTOR_HEARTBEAT_INTERVAL.key))

  override def onExecutorMetricsUpdate(event: SparkListenerExecutorMetricsUpdate): Unit = {
  val now = System.nanoTime()
  event.accumUpdates.foreach { case (taskId, sid, sAttempt, accumUpdates) =>
    liveTasks.get(taskId).foreach { task =>
      val metrics = TaskMetrics.fromAccumulatorInfos(accumUpdates)
      val delta = task.updateMetrics(metrics)
      maybeUpdate(task, now)
      Option(liveStages.get((sid, sAttempt))).foreach { stage =>
       stage.metrics = LiveEntityHelpers.addMetrics(stage.metrics, delta)
       maybeUpdate(stage, now)
       val esummary = stage.executorSummary(event.execId)
       esummary.metrics = LiveEntityHelpers.addMetrics(esummary.metrics, delta)
       maybeUpdate(esummary, now)
     }
   }
}