address comments

Author: cloud-fan

core/src/main/scala/org/apache/spark/InternalAccumulator.scala

@@ -17,17 +17,11 @@
 
 package org.apache.spark
 
-import org.apache.spark.storage.{BlockId, BlockStatus}
-
-
 /**
  * A collection of fields and methods concerned with internal accumulators that represent
  * task level metrics.
  */
 private[spark] object InternalAccumulator {
-
-  import AccumulatorParam._
-
   // Prefixes used in names of internal task level metrics
   val METRICS_PREFIX = "internal.metrics."
   val SHUFFLE_READ_METRICS_PREFIX = METRICS_PREFIX + "shuffle.read."
@@ -79,125 +73,4 @@ private[spark] object InternalAccumulator {
   }
 
   // scalastyle:on
-
-  /**
-   * Create an internal [[Accumulator]] by name, which must begin with [[METRICS_PREFIX]].
-   */
-  def create(name: String): Accumulator[_] = {
-    require(name.startsWith(METRICS_PREFIX),
-      s"internal accumulator name must start with '$METRICS_PREFIX': $name")
-    getParam(name) match {
-      case p @ LongAccumulatorParam => newMetric[Long](0L, name, p)
-      case p @ IntAccumulatorParam => newMetric[Int](0, name, p)
-      case p @ StringAccumulatorParam => newMetric[String]("", name, p)
-      case p @ UpdatedBlockStatusesAccumulatorParam =>
-        newMetric[Seq[(BlockId, BlockStatus)]](Seq(), name, p)
-      case p => throw new IllegalArgumentException(
-        s"unsupported accumulator param '${p.getClass.getSimpleName}' for metric '$name'.")
-    }
-  }
-
-  /**
-   * Get the [[AccumulatorParam]] associated with the internal metric name,
-   * which must begin with [[METRICS_PREFIX]].
-   */
-  def getParam(name: String): AccumulatorParam[_] = {
-    require(name.startsWith(METRICS_PREFIX),
-      s"internal accumulator name must start with '$METRICS_PREFIX': $name")
-    name match {
-      case UPDATED_BLOCK_STATUSES => UpdatedBlockStatusesAccumulatorParam
-      case shuffleRead.LOCAL_BLOCKS_FETCHED => IntAccumulatorParam
-      case shuffleRead.REMOTE_BLOCKS_FETCHED => IntAccumulatorParam
-      case _ => LongAccumulatorParam
-    }
-  }
-
-  /**
-   * Accumulators for tracking internal metrics.
-   */
-  def createAll(): Seq[Accumulator[_]] = {
-    Seq[String](
-      EXECUTOR_DESERIALIZE_TIME,
-      EXECUTOR_RUN_TIME,
-      RESULT_SIZE,
-      JVM_GC_TIME,
-      RESULT_SERIALIZATION_TIME,
-      MEMORY_BYTES_SPILLED,
-      DISK_BYTES_SPILLED,
-      PEAK_EXECUTION_MEMORY,
-      UPDATED_BLOCK_STATUSES).map(create) ++
-      createShuffleReadAccums() ++
-      createShuffleWriteAccums() ++
-      createInputAccums() ++
-      createOutputAccums() ++
-      sys.props.get("spark.testing").map(_ => create(TEST_ACCUM)).toSeq
-  }
-
-  /**
-   * Accumulators for tracking shuffle read metrics.
-   */
-  def createShuffleReadAccums(): Seq[Accumulator[_]] = {
-    Seq[String](
-      shuffleRead.REMOTE_BLOCKS_FETCHED,
-      shuffleRead.LOCAL_BLOCKS_FETCHED,
-      shuffleRead.REMOTE_BYTES_READ,
-      shuffleRead.LOCAL_BYTES_READ,
-      shuffleRead.FETCH_WAIT_TIME,
-      shuffleRead.RECORDS_READ).map(create)
-  }
-
-  /**
-   * Accumulators for tracking shuffle write metrics.
-   */
-  def createShuffleWriteAccums(): Seq[Accumulator[_]] = {
-    Seq[String](
-      shuffleWrite.BYTES_WRITTEN,
-      shuffleWrite.RECORDS_WRITTEN,
-      shuffleWrite.WRITE_TIME).map(create)
-  }
-
-  /**
-   * Accumulators for tracking input metrics.
-   */
-  def createInputAccums(): Seq[Accumulator[_]] = {
-    Seq[String](
-      input.BYTES_READ,
-      input.RECORDS_READ).map(create)
-  }
-
-  /**
-   * Accumulators for tracking output metrics.
-   */
-  def createOutputAccums(): Seq[Accumulator[_]] = {
-    Seq[String](
-      output.BYTES_WRITTEN,
-      output.RECORDS_WRITTEN).map(create)
-  }
-
-  /**
-   * Accumulators for tracking internal metrics.
-   *
-   * These accumulators are created with the stage such that all tasks in the stage will
-   * add to the same set of accumulators. We do this to report the distribution of accumulator
-   * values across all tasks within each stage.
-   */
-  def createAll(sc: SparkContext): Seq[Accumulator[_]] = {
-    val accums = createAll()
-    accums.foreach { accum =>
-      Accumulators.register(accum)
-      sc.cleaner.foreach(_.registerAccumulatorForCleanup(accum))
-    }
-    accums
-  }
-
-  /**
-   * Create a new accumulator representing an internal task metric.
-   */
-  private def newMetric[T](
-      initialValue: T,
-      name: String,
-      param: AccumulatorParam[T]): Accumulator[T] = {
-    new Accumulator[T](initialValue, param, Some(name), internal = true, countFailedValues = true)
-  }
-
 }

core/src/main/scala/org/apache/spark/TaskContext.scala

@@ -62,12 +62,11 @@ object TaskContext {
   protected[spark] def unset(): Unit = taskContext.remove()
 
   /**
-   * An empty task context that does not represent an actual task.
+   * An empty task context that does not represent an actual task.  This is only used in tests.
    */
   private[spark] def empty(): TaskContextImpl = {
     new TaskContextImpl(0, 0, 0, 0, null, new Properties, null, new TaskMetrics)
   }
-
 }
 
 

core/src/main/scala/org/apache/spark/TaskContextImpl.scala

@@ -36,7 +36,7 @@ private[spark] class TaskContextImpl(
     override val taskMemoryManager: TaskMemoryManager,
     localProperties: Properties,
     @transient private val metricsSystem: MetricsSystem,
-    val taskMetrics: TaskMetrics)
+    override val taskMetrics: TaskMetrics)
   extends TaskContext
   with Logging {
 

core/src/main/scala/org/apache/spark/executor/InputMetrics.scala

@@ -42,8 +42,8 @@ object DataReadMethod extends Enumeration with Serializable {
 class InputMetrics private[spark] () extends Serializable {
   import InternalAccumulator._
 
-  private[executor] val _bytesRead = TaskMetrics.createAccum[Long](input.BYTES_READ)
-  private[executor] val _recordsRead = TaskMetrics.createAccum[Long](input.RECORDS_READ)
+  private[executor] val _bytesRead = TaskMetrics.createLongAccum(input.BYTES_READ)
+  private[executor] val _recordsRead = TaskMetrics.createLongAccum(input.RECORDS_READ)
 
   /**
    * Total number of bytes read.

core/src/main/scala/org/apache/spark/executor/OutputMetrics.scala

@@ -41,8 +41,8 @@ object DataWriteMethod extends Enumeration with Serializable {
 class OutputMetrics private[spark] () extends Serializable {
   import InternalAccumulator._
 
-  private[executor] val _bytesWritten = TaskMetrics.createAccum[Long](output.BYTES_WRITTEN)
-  private[executor] val _recordsWritten = TaskMetrics.createAccum[Long](output.RECORDS_WRITTEN)
+  private[executor] val _bytesWritten = TaskMetrics.createLongAccum(output.BYTES_WRITTEN)
+  private[executor] val _recordsWritten = TaskMetrics.createLongAccum(output.RECORDS_WRITTEN)
 
   /**
    * Total number of bytes written.

core/src/main/scala/org/apache/spark/executor/ShuffleReadMetrics.scala

@@ -31,17 +31,17 @@ class ShuffleReadMetrics private[spark] () extends Serializable {
   import InternalAccumulator._
 
   private[executor] val _remoteBlocksFetched =
-    TaskMetrics.createAccum[Int](shuffleRead.REMOTE_BLOCKS_FETCHED)
+    TaskMetrics.createIntAccum(shuffleRead.REMOTE_BLOCKS_FETCHED)
   private[executor] val _localBlocksFetched =
-    TaskMetrics.createAccum[Int](shuffleRead.LOCAL_BLOCKS_FETCHED)
+    TaskMetrics.createIntAccum(shuffleRead.LOCAL_BLOCKS_FETCHED)
   private[executor] val _remoteBytesRead =
-    TaskMetrics.createAccum[Long](shuffleRead.REMOTE_BYTES_READ)
+    TaskMetrics.createLongAccum(shuffleRead.REMOTE_BYTES_READ)
   private[executor] val _localBytesRead =
-    TaskMetrics.createAccum[Long](shuffleRead.LOCAL_BYTES_READ)
+    TaskMetrics.createLongAccum(shuffleRead.LOCAL_BYTES_READ)
   private[executor] val _fetchWaitTime =
-    TaskMetrics.createAccum[Long](shuffleRead.FETCH_WAIT_TIME)
+    TaskMetrics.createLongAccum(shuffleRead.FETCH_WAIT_TIME)
   private[executor] val _recordsRead =
-    TaskMetrics.createAccum[Long](shuffleRead.RECORDS_READ)
+    TaskMetrics.createLongAccum(shuffleRead.RECORDS_READ)
 
   /**
    * Number of remote blocks fetched in this shuffle by this task.

core/src/main/scala/org/apache/spark/executor/ShuffleWriteMetrics.scala

@@ -31,11 +31,11 @@ class ShuffleWriteMetrics private[spark] () extends Serializable {
   import InternalAccumulator._
 
   private[executor] val _bytesWritten =
-    TaskMetrics.createAccum[Long](shuffleWrite.BYTES_WRITTEN)
+    TaskMetrics.createLongAccum(shuffleWrite.BYTES_WRITTEN)
   private[executor] val _recordsWritten =
-    TaskMetrics.createAccum[Long](shuffleWrite.RECORDS_WRITTEN)
+    TaskMetrics.createLongAccum(shuffleWrite.RECORDS_WRITTEN)
   private[executor] val _writeTime =
-    TaskMetrics.createAccum[Long](shuffleWrite.WRITE_TIME)
+    TaskMetrics.createLongAccum(shuffleWrite.WRITE_TIME)
 
   /**
    * Number of bytes written for the shuffle by this task.

core/src/main/scala/org/apache/spark/executor/TaskMetrics.scala

@@ -20,6 +20,7 @@ package org.apache.spark.executor
 import scala.collection.mutable.ArrayBuffer
 
 import org.apache.spark._
+import org.apache.spark.AccumulatorParam.{IntAccumulatorParam, LongAccumulatorParam, UpdatedBlockStatusesAccumulatorParam}
 import org.apache.spark.annotation.DeveloperApi
 import org.apache.spark.internal.Logging
 import org.apache.spark.scheduler.AccumulableInfo
@@ -44,16 +45,15 @@ class TaskMetrics private[spark] () extends Serializable {
   import InternalAccumulator._
 
   // Each metric is internally represented as an accumulator
-  private val _executorDeserializeTime = TaskMetrics.createAccum[Long](EXECUTOR_DESERIALIZE_TIME)
-  private val _executorRunTime = TaskMetrics.createAccum[Long](EXECUTOR_RUN_TIME)
-  private val _resultSize = TaskMetrics.createAccum[Long](RESULT_SIZE)
-  private val _jvmGCTime = TaskMetrics.createAccum[Long](JVM_GC_TIME)
-  private val _resultSerializationTime = TaskMetrics.createAccum[Long](RESULT_SERIALIZATION_TIME)
-  private val _memoryBytesSpilled = TaskMetrics.createAccum[Long](MEMORY_BYTES_SPILLED)
-  private val _diskBytesSpilled = TaskMetrics.createAccum[Long](DISK_BYTES_SPILLED)
-  private val _peakExecutionMemory = TaskMetrics.createAccum[Long](PEAK_EXECUTION_MEMORY)
-  private val _updatedBlockStatuses =
-    TaskMetrics.createAccum[Seq[(BlockId, BlockStatus)]](UPDATED_BLOCK_STATUSES)
+  private val _executorDeserializeTime = TaskMetrics.createLongAccum(EXECUTOR_DESERIALIZE_TIME)
+  private val _executorRunTime = TaskMetrics.createLongAccum(EXECUTOR_RUN_TIME)
+  private val _resultSize = TaskMetrics.createLongAccum(RESULT_SIZE)
+  private val _jvmGCTime = TaskMetrics.createLongAccum(JVM_GC_TIME)
+  private val _resultSerializationTime = TaskMetrics.createLongAccum(RESULT_SERIALIZATION_TIME)
+  private val _memoryBytesSpilled = TaskMetrics.createLongAccum(MEMORY_BYTES_SPILLED)
+  private val _diskBytesSpilled = TaskMetrics.createLongAccum(DISK_BYTES_SPILLED)
+  private val _peakExecutionMemory = TaskMetrics.createLongAccum(PEAK_EXECUTION_MEMORY)
+  private val _updatedBlockStatuses = TaskMetrics.createBlocksAccum(UPDATED_BLOCK_STATUSES)
 
   /**
    * Time taken on the executor to deserialize this task.
@@ -176,7 +176,7 @@ class TaskMetrics private[spark] () extends Serializable {
 
   // Only used for test
   private[spark] val testAccum =
-    sys.props.get("spark.testing").map(_ => TaskMetrics.createAccum[Long](TEST_ACCUM))
+    sys.props.get("spark.testing").map(_ => TaskMetrics.createLongAccum(TEST_ACCUM))
 
   @transient private[spark] lazy val internalAccums: Seq[Accumulable[_, _]] = {
     val in = inputMetrics
@@ -243,11 +243,30 @@ private[spark] class ListenerTaskMetrics(accumUpdates: Seq[AccumulableInfo]) ext
 }
 
 private[spark] object TaskMetrics extends Logging {
-  import InternalAccumulator._
 
   def empty: TaskMetrics = new TaskMetrics
 
-  def createAccum[T](name: String): Accumulator[T] = create(name).asInstanceOf[Accumulator[T]]
+  /**
+   * Create a new accumulator representing an internal task metric.
+   */
+  private def newMetric[T](
+      initialValue: T,
+      name: String,
+      param: AccumulatorParam[T]): Accumulator[T] = {
+    new Accumulator[T](initialValue, param, Some(name), internal = true, countFailedValues = true)
+  }
+
+  def createLongAccum(name: String): Accumulator[Long] = {
+    newMetric(0L, name, LongAccumulatorParam)
+  }
+
+  def createIntAccum(name: String): Accumulator[Int] = {
+    newMetric(0, name, IntAccumulatorParam)
+  }
+
+  def createBlocksAccum(name: String): Accumulator[Seq[(BlockId, BlockStatus)]] = {
+    newMetric(Nil, name, UpdatedBlockStatusesAccumulatorParam)
+  }
 
   /**
    * Construct a [[TaskMetrics]] object from a list of accumulator updates, called on driver only.

core/src/main/scala/org/apache/spark/scheduler/StageInfo.scala

@@ -36,7 +36,7 @@ class StageInfo(
     val rddInfos: Seq[RDDInfo],
     val parentIds: Seq[Int],
     val details: String,
-    val taskMetrics: TaskMetrics = null,
+    val taskMetrics: TaskMetrics = new TaskMetrics,
     private[spark] val taskLocalityPreferences: Seq[Seq[TaskLocation]] = Seq.empty) {
   /** When this stage was submitted from the DAGScheduler to a TaskScheduler. */
   var submissionTime: Option[Long] = None
@@ -81,7 +81,7 @@ private[spark] object StageInfo {
       stage: Stage,
       attemptId: Int,
       numTasks: Option[Int] = None,
-      taskMetrics: TaskMetrics = null,
+      taskMetrics: TaskMetrics = new TaskMetrics,
       taskLocalityPreferences: Seq[Seq[TaskLocation]] = Seq.empty
     ): StageInfo = {
     val ancestorRddInfos = stage.rdd.getNarrowAncestors.map(RDDInfo.fromRdd)

core/src/main/scala/org/apache/spark/scheduler/Task.scala

@@ -46,13 +46,15 @@ import org.apache.spark.util.{ByteBufferInputStream, ByteBufferOutputStream, Uti
  * @param partitionId index of the number in the RDD
  * @param metrics a [[TaskMetrics]] that is created at driver side and sent to executor side.
  * @param localProperties copy of thread-local properties set by the user on the driver side.
+ *
+ * The default values for `metrics` and `localProperties` are used by tests only.
  */
 private[spark] abstract class Task[T](
     val stageId: Int,
     val stageAttemptId: Int,
     val partitionId: Int,
-    val metrics: TaskMetrics,
-    @transient var localProperties: Properties) extends Serializable {
+    val metrics: TaskMetrics = new TaskMetrics,
+    @transient var localProperties: Properties = new Properties) extends Serializable {
 
   /**
    * Called by [[org.apache.spark.executor.Executor]] to run this task.