add in metrics now available

jeanlyn · Mar 10, 2015 · 017c755 · 017c755
1 parent 1b78cb7
commit 017c755
Showing 1 changed file with 9 additions and 9 deletions.
diff --git a/core/src/main/scala/org/apache/spark/status/api/v1/AllStagesResource.scala b/core/src/main/scala/org/apache/spark/status/api/v1/AllStagesResource.scala
@@ -95,13 +95,13 @@ object AllStagesResource {
       numFailedTasks = stageUiData.numFailedTasks,
       executorRunTime = stageUiData.executorRunTime,
       inputBytes = stageUiData.inputBytes,
-      inputRecords = 0, //fix after SPARK-4874
+      inputRecords = stageUiData.inputRecords,
       outputBytes = stageUiData.outputBytes,
-      outputRecords = 0, //fix after SPARK-4874
+      outputRecords = stageUiData.outputRecords,
       shuffleReadBytes = stageUiData.shuffleReadTotalBytes,
-      shuffleReadRecords = 0, //fix after SPARK-4874
+      shuffleReadRecords = stageUiData.shuffleReadRecords,
       shuffleWriteBytes = stageUiData.shuffleWriteBytes,
-      shuffleWriteRecords = 0, //fix after SPARK-4874
+      shuffleWriteRecords = stageUiData.shuffleWriteRecords,
       memoryBytesSpilled = stageUiData.memoryBytesSpilled,
       diskBytesSpilled = stageUiData.diskBytesSpilled,
       schedulingPool = stageUiData.schedulingPool,
@@ -159,14 +159,14 @@ object AllStagesResource {
   def convertInputMetrics(internal: InternalInputMetrics): InputMetrics = {
     InputMetrics(
       bytesRead = internal.bytesRead,
-      recordsRead = 0 //fix after SPARK-4874
+      recordsRead = internal.recordsRead
     )
   }
 
   def convertOutputMetrics(internal: InternalOutputMetrics): OutputMetrics = {
     OutputMetrics(
       bytesWritten = internal.bytesWritten,
-      recordsWritten = 0  //fix after SPARK-4874
+      recordsWritten = internal.recordsWritten
     )
   }
 
@@ -177,16 +177,16 @@ object AllStagesResource {
       fetchWaitTime = internal.fetchWaitTime,
       remoteBytesRead = internal.remoteBytesRead,
       totalBlocksFetched = internal.totalBlocksFetched,
-      recordsRead = 0 //fix after SPARK-4874
+      recordsRead = internal.recordsRead
     )
   }
 
   def convertShuffleWriteMetrics(internal: InternalShuffleWriteMetrics): ShuffleWriteMetrics = {
     ShuffleWriteMetrics(
       bytesWritten = internal.shuffleBytesWritten,
       writeTime = internal.shuffleWriteTime,
-      recordsWritten = 0  //fix after SPARK-4874
+      recordsWritten = internal.shuffleRecordsWritten
     )
   }
 
-}
+}