Remove input metrics for parallel collections

Author: kayousterhout

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

@@ -67,8 +67,8 @@ class TaskMetrics extends Serializable {
   var diskBytesSpilled: Long = _
 
   /**
-   * If this task reads from a HadoopRDD, from cached data, or from a parallelized collection,
-   * metrics on how much data was read are stored here.
+   * If this task reads from a HadoopRDD or from persisted data, metrics on how much data was read
+   * are stored here.
    */
   var inputMetrics: Option[InputMetrics] = None
 

core/src/main/scala/org/apache/spark/rdd/ParallelCollectionRDD.scala

@@ -26,9 +26,8 @@ import scala.collection.mutable.ArrayBuffer
 import scala.reflect.ClassTag
 
 import org.apache.spark._
-import org.apache.spark.executor.{DataReadMethod, InputMetrics}
 import org.apache.spark.serializer.JavaSerializer
-import org.apache.spark.util.{SizeEstimator, Utils}
+import org.apache.spark.util.Utils
 
 private[spark] class ParallelCollectionPartition[T: ClassTag](
     var rddId: Long,
@@ -100,14 +99,7 @@ private[spark] class ParallelCollectionRDD[T: ClassTag](
   }
 
   override def compute(s: Partition, context: TaskContext) = {
-    val parallelCollectionsPartition = s.asInstanceOf[ParallelCollectionPartition[T]]
-
-    // Set the input metrics for the task.
-    val inputMetrics = new InputMetrics(DataReadMethod.Memory)
-    inputMetrics.bytesRead = SizeEstimator.estimate(parallelCollectionsPartition.values)
-    context.taskMetrics.inputMetrics = Some(inputMetrics)
-
-    new InterruptibleIterator(context, parallelCollectionsPartition.iterator)
+    new InterruptibleIterator(context, s.asInstanceOf[ParallelCollectionPartition[T]].iterator)
   }
 
   override def getPreferredLocations(s: Partition): Seq[String] = {