diff --git a/core/src/main/scala/org/apache/spark/status/AppStatusListener.scala b/core/src/main/scala/org/apache/spark/status/AppStatusListener.scala
index b085f21f2d5cc..6f459295f2b7d 100644
--- a/core/src/main/scala/org/apache/spark/status/AppStatusListener.scala
+++ b/core/src/main/scala/org/apache/spark/status/AppStatusListener.scala
@@ -831,6 +831,21 @@ private[spark] class AppStatusListener(
       }
     }
 
+    // update stage executor metrics
+    event.executorUpdates.foreach { updates =>
+      val activeStages = liveStages.values().asScala
+        .filter(_.status == v1.StageStatus.ACTIVE)
+      activeStages.foreach { stage =>
+        if (stage.peakExecutorMetrics.compareAndUpdatePeakValues(updates)) {
+          maybeUpdate(stage, now)
+        }
+        val executorSummary = stage.executorSummary(event.execId)
+        if (executorSummary.peakExecutorMetrics.compareAndUpdatePeakValues(updates)) {
+          maybeUpdate(executorSummary, now)
+        }
+      }
+    }
+
     // check if there is a new peak value for any of the executor level memory metrics
     // for the live UI. SparkListenerExecutorMetricsUpdate events are only processed
     // for the live UI.
@@ -851,17 +866,31 @@ private[spark] class AppStatusListener(
     }
   }
 
-  override def onStageExecutorMetrics(executorMetrics: SparkListenerStageExecutorMetrics): Unit = {
-    val now = System.nanoTime()
+  override def onStageExecutorMetrics(event: SparkListenerStageExecutorMetrics): Unit = {
+    // SparkListenerStageExecutorMetrics are only processed when reading logs, and contain
+    // the peak executor metrics per executor and stage. For live applications, executor
+    // metrics are sent as part of SparkListenerExecutorMetricsUpdate.
 
-    // check if there is a new peak value for any of the executor level memory metrics,
-    // while reading from the log. SparkListenerStageExecutorMetrics are only processed
-    // when reading logs.
-    liveExecutors.get(executorMetrics.execId).orElse(
-      deadExecutors.get(executorMetrics.execId)).foreach { exec =>
-        if (exec.peakExecutorMetrics.compareAndUpdatePeakValues(executorMetrics.executorMetrics)) {
+    // Update stage level peak executor metric values
+    val now = System.nanoTime()
+    Option(liveStages.get((event.stageId, event.stageAttemptId)))
+      .foreach { stage =>
+      if (stage.peakExecutorMetrics.compareAndUpdatePeakValues(event.executorMetrics)) {
+        update(stage, now)
+      }
+      val esummary = stage.executorSummary(event.execId)
+      if (esummary.peakExecutorMetrics.compareAndUpdatePeakValues(event.executorMetrics)) {
+        update(esummary, now)
+      }
+      // check if there is a new peak value for any of the executor level memory metrics,
+      // while reading from the log. SparkListenerStageExecutorMetrics are only processed
+      // when reading logs.
+      liveExecutors.get(event.execId).orElse(
+        deadExecutors.get(event.execId)).foreach { exec =>
+        if (exec.peakExecutorMetrics.compareAndUpdatePeakValues(event.executorMetrics)) {
           update(exec, now)
         }
+      }
     }
   }
 
diff --git a/core/src/main/scala/org/apache/spark/status/AppStatusStore.scala b/core/src/main/scala/org/apache/spark/status/AppStatusStore.scala
index 0487f2f07c097..34bb227926cc9 100644
--- a/core/src/main/scala/org/apache/spark/status/AppStatusStore.scala
+++ b/core/src/main/scala/org/apache/spark/status/AppStatusStore.scala
@@ -495,7 +495,8 @@ private[spark] class AppStatusStore(
       stage.accumulatorUpdates,
       Some(tasks),
       Some(executorSummary(stage.stageId, stage.attemptId)),
-      stage.killedTasksSummary)
+      stage.killedTasksSummary,
+      stage.peakExecutorMetrics)
   }
 
   def rdd(rddId: Int): v1.RDDStorageInfo = {
diff --git a/core/src/main/scala/org/apache/spark/status/LiveEntity.scala b/core/src/main/scala/org/apache/spark/status/LiveEntity.scala
index 6d7b34ae979f2..0f99f46316667 100644
--- a/core/src/main/scala/org/apache/spark/status/LiveEntity.scala
+++ b/core/src/main/scala/org/apache/spark/status/LiveEntity.scala
@@ -328,6 +328,9 @@ private class LiveExecutorStageSummary(
 
   var metrics = createMetrics(default = 0L)
 
+  // peak values for executor level metrics
+  val peakExecutorMetrics = new ExecutorMetrics()
+
   override protected def doUpdate(): Any = {
     val info = new v1.ExecutorStageSummary(
       taskTime,
@@ -344,7 +347,8 @@ private class LiveExecutorStageSummary(
       metrics.shuffleWriteMetrics.recordsWritten,
       metrics.memoryBytesSpilled,
       metrics.diskBytesSpilled,
-      isBlacklisted)
+      isBlacklisted,
+      if (peakExecutorMetrics.isSet()) Some(peakExecutorMetrics) else None)
     new ExecutorStageSummaryWrapper(stageId, attemptId, executorId, info)
   }
 
@@ -383,6 +387,8 @@ private class LiveStage extends LiveEntity {
 
   var blackListedExecutors = new HashSet[String]()
 
+  val peakExecutorMetrics = new ExecutorMetrics()
+
   // Used for cleanup of tasks after they reach the configured limit. Not written to the store.
   @volatile var cleaning = false
   var savedTasks = new AtomicInteger(0)
@@ -432,7 +438,8 @@ private class LiveStage extends LiveEntity {
       newAccumulatorInfos(info.accumulables.values),
       None,
       None,
-      killedSummary)
+      killedSummary,
+      if (peakExecutorMetrics.isSet()) Some(peakExecutorMetrics) else None)
   }
 
   override protected def doUpdate(): Any = {
diff --git a/core/src/main/scala/org/apache/spark/status/api/v1/api.scala b/core/src/main/scala/org/apache/spark/status/api/v1/api.scala
index 60b8bf8921814..7c17105a5ee9f 100644
--- a/core/src/main/scala/org/apache/spark/status/api/v1/api.scala
+++ b/core/src/main/scala/org/apache/spark/status/api/v1/api.scala
@@ -76,7 +76,10 @@ class ExecutorStageSummary private[spark](
     val shuffleWriteRecords : Long,
     val memoryBytesSpilled : Long,
     val diskBytesSpilled : Long,
-    val isBlacklistedForStage: Boolean)
+    val isBlacklistedForStage: Boolean,
+    @JsonSerialize(using = classOf[ExecutorMetricsJsonSerializer])
+    @JsonDeserialize(using = classOf[ExecutorMetricsJsonDeserializer])
+    val peakMemoryMetrics: Option[ExecutorMetrics])
 
 class ExecutorSummary private[spark](
     val id: String,
@@ -237,7 +240,11 @@ class StageData private[spark](
     val accumulatorUpdates: Seq[AccumulableInfo],
     val tasks: Option[Map[Long, TaskData]],
     val executorSummary: Option[Map[String, ExecutorStageSummary]],
-    val killedTasksSummary: Map[String, Int])
+    val killedTasksSummary: Map[String, Int],
+
+    @JsonSerialize(using = classOf[ExecutorMetricsJsonSerializer])
+    @JsonDeserialize(using = classOf[ExecutorMetricsJsonDeserializer])
+    val peakExecutorMetrics: Option[ExecutorMetrics])
 
 class TaskData private[spark](
     val taskId: Long,
diff --git a/core/src/main/scala/org/apache/spark/ui/jobs/JobPage.scala b/core/src/main/scala/org/apache/spark/ui/jobs/JobPage.scala
index 46295e73e086b..bd837aab21789 100644
--- a/core/src/main/scala/org/apache/spark/ui/jobs/JobPage.scala
+++ b/core/src/main/scala/org/apache/spark/ui/jobs/JobPage.scala
@@ -26,6 +26,7 @@ import scala.xml.{Node, NodeSeq, Unparsed, Utility}
 import org.apache.commons.lang3.StringEscapeUtils
 
 import org.apache.spark.JobExecutionStatus
+import org.apache.spark.executor.ExecutorMetrics
 import org.apache.spark.scheduler._
 import org.apache.spark.status.AppStatusStore
 import org.apache.spark.status.api.v1
@@ -216,7 +217,9 @@ private[ui] class JobPage(parent: JobsTab, store: AppStatusStore) extends WebUIP
           Nil,
           None,
           None,
-          Map())
+          Map(),
+          None
+        )
       }
     }
 
diff --git a/core/src/test/resources/HistoryServerExpectations/stage_list_with_peak_metrics_expectation.json b/core/src/test/resources/HistoryServerExpectations/stage_list_with_peak_metrics_expectation.json
new file mode 100644
index 0000000000000..910c62baf03c4
--- /dev/null
+++ b/core/src/test/resources/HistoryServerExpectations/stage_list_with_peak_metrics_expectation.json
@@ -0,0 +1,743 @@
+[ {
+  "status" : "COMPLETE",
+  "stageId" : 13,
+  "attemptId" : 0,
+  "numTasks" : 2,
+  "numActiveTasks" : 0,
+  "numCompleteTasks" : 2,
+  "numFailedTasks" : 2,
+  "numKilledTasks" : 0,
+  "numCompletedIndices" : 2,
+  "executorRunTime" : 2914,
+  "executorCpuTime" : 1348319363,
+  "submissionTime" : "2018-10-02T00:43:46.936GMT",
+  "firstTaskLaunchedTime" : "2018-10-02T00:43:46.947GMT",
+  "completionTime" : "2018-10-02T00:44:02.335GMT",
+  "inputBytes" : 108,
+  "inputRecords" : 6,
+  "outputBytes" : 0,
+  "outputRecords" : 0,
+  "shuffleReadBytes" : 0,
+  "shuffleReadRecords" : 0,
+  "shuffleWriteBytes" : 0,
+  "shuffleWriteRecords" : 0,
+  "memoryBytesSpilled" : 0,
+  "diskBytesSpilled" : 0,
+  "name" : "sum at BisectingKMeansModel.scala:101",
+  "details" : "org.apache.spark.rdd.DoubleRDDFunctions.sum(DoubleRDDFunctions.scala:34)\norg.apache.spark.mllib.clustering.BisectingKMeansModel.computeCost(BisectingKMeansModel.scala:101)\norg.apache.spark.mllib.clustering.BisectingKMeansModel.computeCost(BisectingKMeansModel.scala:108)\nsun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)\nsun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)\nsun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)\njava.lang.reflect.Method.invoke(Method.java:498)\npy4j.reflection.MethodInvoker.invoke(MethodInvoker.java:244)\npy4j.reflection.ReflectionEngine.invoke(ReflectionEngine.java:357)\npy4j.Gateway.invoke(Gateway.java:282)\npy4j.commands.AbstractCommand.invokeMethod(AbstractCommand.java:132)\npy4j.commands.CallCommand.execute(CallCommand.java:79)\npy4j.GatewayConnection.run(GatewayConnection.java:238)\njava.lang.Thread.run(Thread.java:745)",
+  "schedulingPool" : "default",
+  "rddIds" : [ 36, 35, 34, 0, 1 ],
+  "accumulatorUpdates" : [ ],
+  "killedTasksSummary" : { },
+  "peakExecutorMetrics" : {
+    "JVMHeapMemory" : 266240264,
+    "JVMOffHeapMemory" : 104976128,
+    "OnHeapExecutionMemory" : 0,
+    "OffHeapExecutionMemory" : 0,
+    "OnHeapStorageMemory" : 1088805,
+    "OffHeapStorageMemory" : 0,
+    "OnHeapUnifiedMemory" : 1088805,
+    "OffHeapUnifiedMemory" : 0,
+    "DirectPoolMemory" : 228407,
+    "MappedPoolMemory" : 0,
+    "ProcessTreeJVMVMemory" : 5067235328,
+    "ProcessTreeJVMRSSMemory" : 710475776,
+    "ProcessTreePythonVMemory" : 625926144,
+    "ProcessTreePythonRSSMemory" : 69013504,
+    "ProcessTreeOtherVMemory" : 0,
+    "ProcessTreeOtherRSSMemory" : 0,
+    "MinorGCCount" : 0,
+    "MinorGCTime" : 0,
+    "MajorGCCount" : 0,
+    "MajorGCTime" : 0
+  }
+}, {
+  "status" : "COMPLETE",
+  "stageId" : 12,
+  "attemptId" : 0,
+  "numTasks" : 2,
+  "numActiveTasks" : 0,
+  "numCompleteTasks" : 2,
+  "numFailedTasks" : 0,
+  "numKilledTasks" : 0,
+  "numCompletedIndices" : 2,
+  "executorRunTime" : 52,
+  "executorCpuTime" : 34779652,
+  "submissionTime" : "2018-10-02T00:43:46.702GMT",
+  "firstTaskLaunchedTime" : "2018-10-02T00:43:46.714GMT",
+  "completionTime" : "2018-10-02T00:43:46.840GMT",
+  "inputBytes" : 0,
+  "inputRecords" : 0,
+  "outputBytes" : 0,
+  "outputRecords" : 0,
+  "shuffleReadBytes" : 533,
+  "shuffleReadRecords" : 3,
+  "shuffleWriteBytes" : 0,
+  "shuffleWriteRecords" : 0,
+  "memoryBytesSpilled" : 0,
+  "diskBytesSpilled" : 0,
+  "name" : "collect at BisectingKMeans.scala:304",
+  "details" : "org.apache.spark.rdd.RDD.collect(RDD.scala:944)\norg.apache.spark.mllib.clustering.BisectingKMeans$.org$apache$spark$mllib$clustering$BisectingKMeans$$summarize(BisectingKMeans.scala:304)\norg.apache.spark.mllib.clustering.BisectingKMeans$$anonfun$run$1.apply$mcVI$sp(BisectingKMeans.scala:216)\nscala.collection.immutable.Range.foreach$mVc$sp(Range.scala:160)\norg.apache.spark.mllib.clustering.BisectingKMeans.run(BisectingKMeans.scala:210)\norg.apache.spark.mllib.clustering.BisectingKMeans.run(BisectingKMeans.scala:255)\norg.apache.spark.mllib.clustering.BisectingKMeans.run(BisectingKMeans.scala:261)\norg.apache.spark.mllib.api.python.PythonMLLibAPI.trainBisectingKMeans(PythonMLLibAPI.scala:135)\nsun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)\nsun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)\nsun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)\njava.lang.reflect.Method.invoke(Method.java:498)\npy4j.reflection.MethodInvoker.invoke(MethodInvoker.java:244)\npy4j.reflection.ReflectionEngine.invoke(ReflectionEngine.java:357)\npy4j.Gateway.invoke(Gateway.java:282)\npy4j.commands.AbstractCommand.invokeMethod(AbstractCommand.java:132)\npy4j.commands.CallCommand.execute(CallCommand.java:79)\npy4j.GatewayConnection.run(GatewayConnection.java:238)\njava.lang.Thread.run(Thread.java:745)",
+  "schedulingPool" : "default",
+  "rddIds" : [ 30, 29 ],
+  "accumulatorUpdates" : [ ],
+  "killedTasksSummary" : { },
+  "peakExecutorMetrics" : {
+    "JVMHeapMemory" : 252161344,
+    "JVMOffHeapMemory" : 104055736,
+    "OnHeapExecutionMemory" : 0,
+    "OffHeapExecutionMemory" : 0,
+    "OnHeapStorageMemory" : 519458,
+    "OffHeapStorageMemory" : 0,
+    "OnHeapUnifiedMemory" : 519458,
+    "OffHeapUnifiedMemory" : 0,
+    "DirectPoolMemory" : 228407,
+    "MappedPoolMemory" : 0,
+    "ProcessTreeJVMVMemory" : 5047037952,
+    "ProcessTreeJVMRSSMemory" : 708661248,
+    "ProcessTreePythonVMemory" : 958894080,
+    "ProcessTreePythonRSSMemory" : 106696704,
+    "ProcessTreeOtherVMemory" : 0,
+    "ProcessTreeOtherRSSMemory" : 0,
+    "MinorGCCount" : 0,
+    "MinorGCTime" : 0,
+    "MajorGCCount" : 0,
+    "MajorGCTime" : 0
+  }
+}, {
+  "status" : "COMPLETE",
+  "stageId" : 11,
+  "attemptId" : 0,
+  "numTasks" : 2,
+  "numActiveTasks" : 0,
+  "numCompleteTasks" : 2,
+  "numFailedTasks" : 0,
+  "numKilledTasks" : 0,
+  "numCompletedIndices" : 2,
+  "executorRunTime" : 163,
+  "executorCpuTime" : 46609695,
+  "submissionTime" : "2018-10-02T00:43:46.408GMT",
+  "firstTaskLaunchedTime" : "2018-10-02T00:43:46.450GMT",
+  "completionTime" : "2018-10-02T00:43:46.701GMT",
+  "inputBytes" : 108,
+  "inputRecords" : 12,
+  "outputBytes" : 0,
+  "outputRecords" : 0,
+  "shuffleReadBytes" : 0,
+  "shuffleReadRecords" : 0,
+  "shuffleWriteBytes" : 533,
+  "shuffleWriteRecords" : 3,
+  "memoryBytesSpilled" : 0,
+  "diskBytesSpilled" : 0,
+  "name" : "filter at BisectingKMeans.scala:213",
+  "details" : "org.apache.spark.rdd.RDD.filter(RDD.scala:387)\norg.apache.spark.mllib.clustering.BisectingKMeans$$anonfun$run$1.apply$mcVI$sp(BisectingKMeans.scala:213)\nscala.collection.immutable.Range.foreach$mVc$sp(Range.scala:160)\norg.apache.spark.mllib.clustering.BisectingKMeans.run(BisectingKMeans.scala:210)\norg.apache.spark.mllib.clustering.BisectingKMeans.run(BisectingKMeans.scala:255)\norg.apache.spark.mllib.clustering.BisectingKMeans.run(BisectingKMeans.scala:261)\norg.apache.spark.mllib.api.python.PythonMLLibAPI.trainBisectingKMeans(PythonMLLibAPI.scala:135)\nsun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)\nsun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)\nsun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)\njava.lang.reflect.Method.invoke(Method.java:498)\npy4j.reflection.MethodInvoker.invoke(MethodInvoker.java:244)\npy4j.reflection.ReflectionEngine.invoke(ReflectionEngine.java:357)\npy4j.Gateway.invoke(Gateway.java:282)\npy4j.commands.AbstractCommand.invokeMethod(AbstractCommand.java:132)\npy4j.commands.CallCommand.execute(CallCommand.java:79)\npy4j.GatewayConnection.run(GatewayConnection.java:238)\njava.lang.Thread.run(Thread.java:745)",
+  "schedulingPool" : "default",
+  "rddIds" : [ 28, 27, 8, 7, 0, 2, 6, 3, 5, 1 ],
+  "accumulatorUpdates" : [ ],
+  "killedTasksSummary" : { },
+  "peakExecutorMetrics" : {
+    "JVMHeapMemory" : 249428840,
+    "JVMOffHeapMemory" : 104016864,
+    "OnHeapExecutionMemory" : 0,
+    "OffHeapExecutionMemory" : 0,
+    "OnHeapStorageMemory" : 554933,
+    "OffHeapStorageMemory" : 0,
+    "OnHeapUnifiedMemory" : 554933,
+    "OffHeapUnifiedMemory" : 0,
+    "DirectPoolMemory" : 228407,
+    "MappedPoolMemory" : 0,
+    "ProcessTreeJVMVMemory" : 5040721920,
+    "ProcessTreeJVMRSSMemory" : 705302528,
+    "ProcessTreePythonVMemory" : 958894080,
+    "ProcessTreePythonRSSMemory" : 106696704,
+    "ProcessTreeOtherVMemory" : 0,
+    "ProcessTreeOtherRSSMemory" : 0,
+    "MinorGCCount" : 0,
+    "MinorGCTime" : 0,
+    "MajorGCCount" : 0,
+    "MajorGCTime" : 0
+  }
+}, {
+  "status" : "COMPLETE",
+  "stageId" : 10,
+  "attemptId" : 0,
+  "numTasks" : 2,
+  "numActiveTasks" : 0,
+  "numCompleteTasks" : 2,
+  "numFailedTasks" : 0,
+  "numKilledTasks" : 0,
+  "numCompletedIndices" : 2,
+  "executorRunTime" : 127,
+  "executorCpuTime" : 98937761,
+  "submissionTime" : "2018-10-02T00:43:46.138GMT",
+  "firstTaskLaunchedTime" : "2018-10-02T00:43:46.147GMT",
+  "completionTime" : "2018-10-02T00:43:46.376GMT",
+  "inputBytes" : 0,
+  "inputRecords" : 0,
+  "outputBytes" : 0,
+  "outputRecords" : 0,
+  "shuffleReadBytes" : 533,
+  "shuffleReadRecords" : 3,
+  "shuffleWriteBytes" : 0,
+  "shuffleWriteRecords" : 0,
+  "memoryBytesSpilled" : 0,
+  "diskBytesSpilled" : 0,
+  "name" : "collect at BisectingKMeans.scala:304",
+  "details" : "org.apache.spark.rdd.RDD.collect(RDD.scala:944)\norg.apache.spark.mllib.clustering.BisectingKMeans$.org$apache$spark$mllib$clustering$BisectingKMeans$$summarize(BisectingKMeans.scala:304)\norg.apache.spark.mllib.clustering.BisectingKMeans$$anonfun$run$1.apply$mcVI$sp(BisectingKMeans.scala:216)\nscala.collection.immutable.Range.foreach$mVc$sp(Range.scala:160)\norg.apache.spark.mllib.clustering.BisectingKMeans.run(BisectingKMeans.scala:210)\norg.apache.spark.mllib.clustering.BisectingKMeans.run(BisectingKMeans.scala:255)\norg.apache.spark.mllib.clustering.BisectingKMeans.run(BisectingKMeans.scala:261)\norg.apache.spark.mllib.api.python.PythonMLLibAPI.trainBisectingKMeans(PythonMLLibAPI.scala:135)\nsun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)\nsun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)\nsun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)\njava.lang.reflect.Method.invoke(Method.java:498)\npy4j.reflection.MethodInvoker.invoke(MethodInvoker.java:244)\npy4j.reflection.ReflectionEngine.invoke(ReflectionEngine.java:357)\npy4j.Gateway.invoke(Gateway.java:282)\npy4j.commands.AbstractCommand.invokeMethod(AbstractCommand.java:132)\npy4j.commands.CallCommand.execute(CallCommand.java:79)\npy4j.GatewayConnection.run(GatewayConnection.java:238)\njava.lang.Thread.run(Thread.java:745)",
+  "schedulingPool" : "default",
+  "rddIds" : [ 26, 25 ],
+  "accumulatorUpdates" : [ ],
+  "killedTasksSummary" : { },
+  "peakExecutorMetrics" : {
+    "JVMHeapMemory" : 341644736,
+    "JVMOffHeapMemory" : 103378144,
+    "OnHeapExecutionMemory" : 0,
+    "OffHeapExecutionMemory" : 0,
+    "OnHeapStorageMemory" : 541469,
+    "OffHeapStorageMemory" : 0,
+    "OnHeapUnifiedMemory" : 541469,
+    "OffHeapUnifiedMemory" : 0,
+    "DirectPoolMemory" : 228406,
+    "MappedPoolMemory" : 0,
+    "ProcessTreeJVMVMemory" : 5011247104,
+    "ProcessTreeJVMRSSMemory" : 658989056,
+    "ProcessTreePythonVMemory" : 958894080,
+    "ProcessTreePythonRSSMemory" : 106696704,
+    "ProcessTreeOtherVMemory" : 0,
+    "ProcessTreeOtherRSSMemory" : 0,
+    "MinorGCCount" : 0,
+    "MinorGCTime" : 0,
+    "MajorGCCount" : 0,
+    "MajorGCTime" : 0
+  }
+}, {
+  "status" : "COMPLETE",
+  "stageId" : 9,
+  "attemptId" : 0,
+  "numTasks" : 2,
+  "numActiveTasks" : 0,
+  "numCompleteTasks" : 2,
+  "numFailedTasks" : 3,
+  "numKilledTasks" : 0,
+  "numCompletedIndices" : 2,
+  "executorRunTime" : 3179,
+  "executorCpuTime" : 1357796664,
+  "submissionTime" : "2018-10-02T00:43:26.612GMT",
+  "firstTaskLaunchedTime" : "2018-10-02T00:43:26.622GMT",
+  "completionTime" : "2018-10-02T00:43:46.137GMT",
+  "inputBytes" : 144,
+  "inputRecords" : 12,
+  "outputBytes" : 0,
+  "outputRecords" : 0,
+  "shuffleReadBytes" : 0,
+  "shuffleReadRecords" : 0,
+  "shuffleWriteBytes" : 533,
+  "shuffleWriteRecords" : 3,
+  "memoryBytesSpilled" : 0,
+  "diskBytesSpilled" : 0,
+  "name" : "filter at BisectingKMeans.scala:213",
+  "details" : "org.apache.spark.rdd.RDD.filter(RDD.scala:387)\norg.apache.spark.mllib.clustering.BisectingKMeans$$anonfun$run$1.apply$mcVI$sp(BisectingKMeans.scala:213)\nscala.collection.immutable.Range.foreach$mVc$sp(Range.scala:160)\norg.apache.spark.mllib.clustering.BisectingKMeans.run(BisectingKMeans.scala:210)\norg.apache.spark.mllib.clustering.BisectingKMeans.run(BisectingKMeans.scala:255)\norg.apache.spark.mllib.clustering.BisectingKMeans.run(BisectingKMeans.scala:261)\norg.apache.spark.mllib.api.python.PythonMLLibAPI.trainBisectingKMeans(PythonMLLibAPI.scala:135)\nsun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)\nsun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)\nsun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)\njava.lang.reflect.Method.invoke(Method.java:498)\npy4j.reflection.MethodInvoker.invoke(MethodInvoker.java:244)\npy4j.reflection.ReflectionEngine.invoke(ReflectionEngine.java:357)\npy4j.Gateway.invoke(Gateway.java:282)\npy4j.commands.AbstractCommand.invokeMethod(AbstractCommand.java:132)\npy4j.commands.CallCommand.execute(CallCommand.java:79)\npy4j.GatewayConnection.run(GatewayConnection.java:238)\njava.lang.Thread.run(Thread.java:745)",
+  "schedulingPool" : "default",
+  "rddIds" : [ 24, 23, 8, 7, 0, 2, 6, 3, 5, 1 ],
+  "accumulatorUpdates" : [ ],
+  "killedTasksSummary" : { },
+  "peakExecutorMetrics" : {
+    "JVMHeapMemory" : 332727504,
+    "JVMOffHeapMemory" : 103237664,
+    "OnHeapExecutionMemory" : 0,
+    "OffHeapExecutionMemory" : 0,
+    "OnHeapStorageMemory" : 1482103,
+    "OffHeapStorageMemory" : 0,
+    "OnHeapUnifiedMemory" : 1482103,
+    "OffHeapUnifiedMemory" : 0,
+    "DirectPoolMemory" : 228406,
+    "MappedPoolMemory" : 0,
+    "ProcessTreeJVMVMemory" : 5011247104,
+    "ProcessTreeJVMRSSMemory" : 658915328,
+    "ProcessTreePythonVMemory" : 958894080,
+    "ProcessTreePythonRSSMemory" : 106696704,
+    "ProcessTreeOtherVMemory" : 0,
+    "ProcessTreeOtherRSSMemory" : 0,
+    "MinorGCCount" : 0,
+    "MinorGCTime" : 0,
+    "MajorGCCount" : 0,
+    "MajorGCTime" : 0
+  }
+}, {
+  "status" : "COMPLETE",
+  "stageId" : 8,
+  "attemptId" : 0,
+  "numTasks" : 2,
+  "numActiveTasks" : 0,
+  "numCompleteTasks" : 2,
+  "numFailedTasks" : 1,
+  "numKilledTasks" : 0,
+  "numCompletedIndices" : 2,
+  "executorRunTime" : 1305,
+  "executorCpuTime" : 430107464,
+  "submissionTime" : "2018-10-02T00:43:15.698GMT",
+  "firstTaskLaunchedTime" : "2018-10-02T00:43:15.710GMT",
+  "completionTime" : "2018-10-02T00:43:26.585GMT",
+  "inputBytes" : 0,
+  "inputRecords" : 0,
+  "outputBytes" : 0,
+  "outputRecords" : 0,
+  "shuffleReadBytes" : 533,
+  "shuffleReadRecords" : 3,
+  "shuffleWriteBytes" : 0,
+  "shuffleWriteRecords" : 0,
+  "memoryBytesSpilled" : 0,
+  "diskBytesSpilled" : 0,
+  "name" : "collect at BisectingKMeans.scala:304",
+  "details" : "org.apache.spark.rdd.RDD.collect(RDD.scala:944)\norg.apache.spark.mllib.clustering.BisectingKMeans$.org$apache$spark$mllib$clustering$BisectingKMeans$$summarize(BisectingKMeans.scala:304)\norg.apache.spark.mllib.clustering.BisectingKMeans$$anonfun$run$1.apply$mcVI$sp(BisectingKMeans.scala:216)\nscala.collection.immutable.Range.foreach$mVc$sp(Range.scala:160)\norg.apache.spark.mllib.clustering.BisectingKMeans.run(BisectingKMeans.scala:210)\norg.apache.spark.mllib.clustering.BisectingKMeans.run(BisectingKMeans.scala:255)\norg.apache.spark.mllib.clustering.BisectingKMeans.run(BisectingKMeans.scala:261)\norg.apache.spark.mllib.api.python.PythonMLLibAPI.trainBisectingKMeans(PythonMLLibAPI.scala:135)\nsun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)\nsun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)\nsun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)\njava.lang.reflect.Method.invoke(Method.java:498)\npy4j.reflection.MethodInvoker.invoke(MethodInvoker.java:244)\npy4j.reflection.ReflectionEngine.invoke(ReflectionEngine.java:357)\npy4j.Gateway.invoke(Gateway.java:282)\npy4j.commands.AbstractCommand.invokeMethod(AbstractCommand.java:132)\npy4j.commands.CallCommand.execute(CallCommand.java:79)\npy4j.GatewayConnection.run(GatewayConnection.java:238)\njava.lang.Thread.run(Thread.java:745)",
+  "schedulingPool" : "default",
+  "rddIds" : [ 22, 21 ],
+  "accumulatorUpdates" : [ ],
+  "killedTasksSummary" : { },
+  "peakExecutorMetrics" : {
+    "JVMHeapMemory" : 263995432,
+    "JVMOffHeapMemory" : 101978136,
+    "OnHeapExecutionMemory" : 0,
+    "OffHeapExecutionMemory" : 0,
+    "OnHeapStorageMemory" : 498888,
+    "OffHeapStorageMemory" : 0,
+    "OnHeapUnifiedMemory" : 498888,
+    "OffHeapUnifiedMemory" : 0,
+    "DirectPoolMemory" : 191656,
+    "MappedPoolMemory" : 0,
+    "ProcessTreeJVMVMemory" : 5008089088,
+    "ProcessTreeJVMRSSMemory" : 663732224,
+    "ProcessTreePythonVMemory" : 958963712,
+    "ProcessTreePythonRSSMemory" : 106639360,
+    "ProcessTreeOtherVMemory" : 0,
+    "ProcessTreeOtherRSSMemory" : 0,
+    "MinorGCCount" : 0,
+    "MinorGCTime" : 0,
+    "MajorGCCount" : 0,
+    "MajorGCTime" : 0
+  }
+}, {
+  "status" : "COMPLETE",
+  "stageId" : 7,
+  "attemptId" : 0,
+  "numTasks" : 2,
+  "numActiveTasks" : 0,
+  "numCompleteTasks" : 2,
+  "numFailedTasks" : 2,
+  "numKilledTasks" : 0,
+  "numCompletedIndices" : 2,
+  "executorRunTime" : 4348,
+  "executorCpuTime" : 1498513857,
+  "submissionTime" : "2018-10-02T00:43:00.015GMT",
+  "firstTaskLaunchedTime" : "2018-10-02T00:43:00.049GMT",
+  "completionTime" : "2018-10-02T00:43:15.697GMT",
+  "inputBytes" : 144,
+  "inputRecords" : 12,
+  "outputBytes" : 0,
+  "outputRecords" : 0,
+  "shuffleReadBytes" : 0,
+  "shuffleReadRecords" : 0,
+  "shuffleWriteBytes" : 533,
+  "shuffleWriteRecords" : 3,
+  "memoryBytesSpilled" : 0,
+  "diskBytesSpilled" : 0,
+  "name" : "filter at BisectingKMeans.scala:213",
+  "details" : "org.apache.spark.rdd.RDD.filter(RDD.scala:387)\norg.apache.spark.mllib.clustering.BisectingKMeans$$anonfun$run$1.apply$mcVI$sp(BisectingKMeans.scala:213)\nscala.collection.immutable.Range.foreach$mVc$sp(Range.scala:160)\norg.apache.spark.mllib.clustering.BisectingKMeans.run(BisectingKMeans.scala:210)\norg.apache.spark.mllib.clustering.BisectingKMeans.run(BisectingKMeans.scala:255)\norg.apache.spark.mllib.clustering.BisectingKMeans.run(BisectingKMeans.scala:261)\norg.apache.spark.mllib.api.python.PythonMLLibAPI.trainBisectingKMeans(PythonMLLibAPI.scala:135)\nsun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)\nsun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)\nsun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)\njava.lang.reflect.Method.invoke(Method.java:498)\npy4j.reflection.MethodInvoker.invoke(MethodInvoker.java:244)\npy4j.reflection.ReflectionEngine.invoke(ReflectionEngine.java:357)\npy4j.Gateway.invoke(Gateway.java:282)\npy4j.commands.AbstractCommand.invokeMethod(AbstractCommand.java:132)\npy4j.commands.CallCommand.execute(CallCommand.java:79)\npy4j.GatewayConnection.run(GatewayConnection.java:238)\njava.lang.Thread.run(Thread.java:745)",
+  "schedulingPool" : "default",
+  "rddIds" : [ 20, 8, 19, 7, 0, 2, 6, 3, 5, 1 ],
+  "accumulatorUpdates" : [ ],
+  "killedTasksSummary" : { },
+  "peakExecutorMetrics" : {
+    "JVMHeapMemory" : 258718872,
+    "JVMOffHeapMemory" : 100867584,
+    "OnHeapExecutionMemory" : 0,
+    "OffHeapExecutionMemory" : 0,
+    "OnHeapStorageMemory" : 1482102,
+    "OffHeapStorageMemory" : 0,
+    "OnHeapUnifiedMemory" : 1482102,
+    "OffHeapUnifiedMemory" : 0,
+    "DirectPoolMemory" : 171571,
+    "MappedPoolMemory" : 0,
+    "ProcessTreeJVMVMemory" : 4971368448,
+    "ProcessTreeJVMRSSMemory" : 663375872,
+    "ProcessTreePythonVMemory" : 958963712,
+    "ProcessTreePythonRSSMemory" : 106639360,
+    "ProcessTreeOtherVMemory" : 0,
+    "ProcessTreeOtherRSSMemory" : 0,
+    "MinorGCCount" : 0,
+    "MinorGCTime" : 0,
+    "MajorGCCount" : 0,
+    "MajorGCTime" : 0
+  }
+}, {
+  "status" : "COMPLETE",
+  "stageId" : 6,
+  "attemptId" : 0,
+  "numTasks" : 2,
+  "numActiveTasks" : 0,
+  "numCompleteTasks" : 2,
+  "numFailedTasks" : 0,
+  "numKilledTasks" : 0,
+  "numCompletedIndices" : 2,
+  "executorRunTime" : 46,
+  "executorCpuTime" : 30636031,
+  "submissionTime" : "2018-10-02T00:42:59.854GMT",
+  "firstTaskLaunchedTime" : "2018-10-02T00:42:59.869GMT",
+  "completionTime" : "2018-10-02T00:42:59.973GMT",
+  "inputBytes" : 0,
+  "inputRecords" : 0,
+  "outputBytes" : 0,
+  "outputRecords" : 0,
+  "shuffleReadBytes" : 533,
+  "shuffleReadRecords" : 3,
+  "shuffleWriteBytes" : 0,
+  "shuffleWriteRecords" : 0,
+  "memoryBytesSpilled" : 0,
+  "diskBytesSpilled" : 0,
+  "name" : "collect at BisectingKMeans.scala:304",
+  "details" : "org.apache.spark.rdd.RDD.collect(RDD.scala:944)\norg.apache.spark.mllib.clustering.BisectingKMeans$.org$apache$spark$mllib$clustering$BisectingKMeans$$summarize(BisectingKMeans.scala:304)\norg.apache.spark.mllib.clustering.BisectingKMeans$$anonfun$run$1.apply$mcVI$sp(BisectingKMeans.scala:216)\nscala.collection.immutable.Range.foreach$mVc$sp(Range.scala:160)\norg.apache.spark.mllib.clustering.BisectingKMeans.run(BisectingKMeans.scala:210)\norg.apache.spark.mllib.clustering.BisectingKMeans.run(BisectingKMeans.scala:255)\norg.apache.spark.mllib.clustering.BisectingKMeans.run(BisectingKMeans.scala:261)\norg.apache.spark.mllib.api.python.PythonMLLibAPI.trainBisectingKMeans(PythonMLLibAPI.scala:135)\nsun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)\nsun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)\nsun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)\njava.lang.reflect.Method.invoke(Method.java:498)\npy4j.reflection.MethodInvoker.invoke(MethodInvoker.java:244)\npy4j.reflection.ReflectionEngine.invoke(ReflectionEngine.java:357)\npy4j.Gateway.invoke(Gateway.java:282)\npy4j.commands.AbstractCommand.invokeMethod(AbstractCommand.java:132)\npy4j.commands.CallCommand.execute(CallCommand.java:79)\npy4j.GatewayConnection.run(GatewayConnection.java:238)\njava.lang.Thread.run(Thread.java:745)",
+  "schedulingPool" : "default",
+  "rddIds" : [ 18, 17 ],
+  "accumulatorUpdates" : [ ],
+  "killedTasksSummary" : { },
+  "peakExecutorMetrics" : {
+    "JVMHeapMemory" : 350990264,
+    "JVMOffHeapMemory" : 97710440,
+    "OnHeapExecutionMemory" : 0,
+    "OffHeapExecutionMemory" : 0,
+    "OnHeapStorageMemory" : 456312,
+    "OffHeapStorageMemory" : 0,
+    "OnHeapUnifiedMemory" : 456312,
+    "OffHeapUnifiedMemory" : 0,
+    "DirectPoolMemory" : 135031,
+    "MappedPoolMemory" : 0,
+    "ProcessTreeJVMVMemory" : 4932550656,
+    "ProcessTreeJVMRSSMemory" : 604299264,
+    "ProcessTreePythonVMemory" : 408375296,
+    "ProcessTreePythonRSSMemory" : 40284160,
+    "ProcessTreeOtherVMemory" : 0,
+    "ProcessTreeOtherRSSMemory" : 0,
+    "MinorGCCount" : 0,
+    "MinorGCTime" : 0,
+    "MajorGCCount" : 0,
+    "MajorGCTime" : 0
+  }
+}, {
+  "status" : "COMPLETE",
+  "stageId" : 5,
+  "attemptId" : 0,
+  "numTasks" : 2,
+  "numActiveTasks" : 0,
+  "numCompleteTasks" : 2,
+  "numFailedTasks" : 0,
+  "numKilledTasks" : 0,
+  "numCompletedIndices" : 2,
+  "executorRunTime" : 157,
+  "executorCpuTime" : 45135674,
+  "submissionTime" : "2018-10-02T00:42:59.619GMT",
+  "firstTaskLaunchedTime" : "2018-10-02T00:42:59.638GMT",
+  "completionTime" : "2018-10-02T00:42:59.852GMT",
+  "inputBytes" : 108,
+  "inputRecords" : 12,
+  "outputBytes" : 0,
+  "outputRecords" : 0,
+  "shuffleReadBytes" : 0,
+  "shuffleReadRecords" : 0,
+  "shuffleWriteBytes" : 533,
+  "shuffleWriteRecords" : 3,
+  "memoryBytesSpilled" : 0,
+  "diskBytesSpilled" : 0,
+  "name" : "filter at BisectingKMeans.scala:213",
+  "details" : "org.apache.spark.rdd.RDD.filter(RDD.scala:387)\norg.apache.spark.mllib.clustering.BisectingKMeans$$anonfun$run$1.apply$mcVI$sp(BisectingKMeans.scala:213)\nscala.collection.immutable.Range.foreach$mVc$sp(Range.scala:160)\norg.apache.spark.mllib.clustering.BisectingKMeans.run(BisectingKMeans.scala:210)\norg.apache.spark.mllib.clustering.BisectingKMeans.run(BisectingKMeans.scala:255)\norg.apache.spark.mllib.clustering.BisectingKMeans.run(BisectingKMeans.scala:261)\norg.apache.spark.mllib.api.python.PythonMLLibAPI.trainBisectingKMeans(PythonMLLibAPI.scala:135)\nsun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)\nsun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)\nsun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)\njava.lang.reflect.Method.invoke(Method.java:498)\npy4j.reflection.MethodInvoker.invoke(MethodInvoker.java:244)\npy4j.reflection.ReflectionEngine.invoke(ReflectionEngine.java:357)\npy4j.Gateway.invoke(Gateway.java:282)\npy4j.commands.AbstractCommand.invokeMethod(AbstractCommand.java:132)\npy4j.commands.CallCommand.execute(CallCommand.java:79)\npy4j.GatewayConnection.run(GatewayConnection.java:238)\njava.lang.Thread.run(Thread.java:745)",
+  "schedulingPool" : "default",
+  "rddIds" : [ 16, 8, 7, 0, 2, 15, 6, 3, 5, 1 ],
+  "accumulatorUpdates" : [ ],
+  "killedTasksSummary" : { },
+  "peakExecutorMetrics" : {
+    "JVMHeapMemory" : 341682304,
+    "JVMOffHeapMemory" : 97514672,
+    "OnHeapExecutionMemory" : 0,
+    "OffHeapExecutionMemory" : 0,
+    "OnHeapStorageMemory" : 526317,
+    "OffHeapStorageMemory" : 0,
+    "OnHeapUnifiedMemory" : 526317,
+    "OffHeapUnifiedMemory" : 0,
+    "DirectPoolMemory" : 135031,
+    "MappedPoolMemory" : 0,
+    "ProcessTreeJVMVMemory" : 4935254016,
+    "ProcessTreeJVMRSSMemory" : 597999616,
+    "ProcessTreePythonVMemory" : 958914560,
+    "ProcessTreePythonRSSMemory" : 106622976,
+    "ProcessTreeOtherVMemory" : 0,
+    "ProcessTreeOtherRSSMemory" : 0,
+    "MinorGCCount" : 0,
+    "MinorGCTime" : 0,
+    "MajorGCCount" : 0,
+    "MajorGCTime" : 0
+  }
+}, {
+  "status" : "COMPLETE",
+  "stageId" : 4,
+  "attemptId" : 0,
+  "numTasks" : 2,
+  "numActiveTasks" : 0,
+  "numCompleteTasks" : 2,
+  "numFailedTasks" : 0,
+  "numKilledTasks" : 0,
+  "numCompletedIndices" : 2,
+  "executorRunTime" : 46,
+  "executorCpuTime" : 34176533,
+  "submissionTime" : "2018-10-02T00:42:59.446GMT",
+  "firstTaskLaunchedTime" : "2018-10-02T00:42:59.462GMT",
+  "completionTime" : "2018-10-02T00:42:59.573GMT",
+  "inputBytes" : 0,
+  "inputRecords" : 0,
+  "outputBytes" : 0,
+  "outputRecords" : 0,
+  "shuffleReadBytes" : 533,
+  "shuffleReadRecords" : 3,
+  "shuffleWriteBytes" : 0,
+  "shuffleWriteRecords" : 0,
+  "memoryBytesSpilled" : 0,
+  "diskBytesSpilled" : 0,
+  "name" : "collect at BisectingKMeans.scala:304",
+  "details" : "org.apache.spark.rdd.RDD.collect(RDD.scala:944)\norg.apache.spark.mllib.clustering.BisectingKMeans$.org$apache$spark$mllib$clustering$BisectingKMeans$$summarize(BisectingKMeans.scala:304)\norg.apache.spark.mllib.clustering.BisectingKMeans$$anonfun$run$1.apply$mcVI$sp(BisectingKMeans.scala:216)\nscala.collection.immutable.Range.foreach$mVc$sp(Range.scala:160)\norg.apache.spark.mllib.clustering.BisectingKMeans.run(BisectingKMeans.scala:210)\norg.apache.spark.mllib.clustering.BisectingKMeans.run(BisectingKMeans.scala:255)\norg.apache.spark.mllib.clustering.BisectingKMeans.run(BisectingKMeans.scala:261)\norg.apache.spark.mllib.api.python.PythonMLLibAPI.trainBisectingKMeans(PythonMLLibAPI.scala:135)\nsun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)\nsun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)\nsun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)\njava.lang.reflect.Method.invoke(Method.java:498)\npy4j.reflection.MethodInvoker.invoke(MethodInvoker.java:244)\npy4j.reflection.ReflectionEngine.invoke(ReflectionEngine.java:357)\npy4j.Gateway.invoke(Gateway.java:282)\npy4j.commands.AbstractCommand.invokeMethod(AbstractCommand.java:132)\npy4j.commands.CallCommand.execute(CallCommand.java:79)\npy4j.GatewayConnection.run(GatewayConnection.java:238)\njava.lang.Thread.run(Thread.java:745)",
+  "schedulingPool" : "default",
+  "rddIds" : [ 14, 13 ],
+  "accumulatorUpdates" : [ ],
+  "killedTasksSummary" : { },
+  "peakExecutorMetrics" : {
+    "JVMHeapMemory" : 329919832,
+    "JVMOffHeapMemory" : 96756344,
+    "OnHeapExecutionMemory" : 0,
+    "OffHeapExecutionMemory" : 0,
+    "OnHeapStorageMemory" : 505748,
+    "OffHeapStorageMemory" : 0,
+    "OnHeapUnifiedMemory" : 505748,
+    "OffHeapUnifiedMemory" : 0,
+    "DirectPoolMemory" : 135031,
+    "MappedPoolMemory" : 0,
+    "ProcessTreeJVMVMemory" : 4935208960,
+    "ProcessTreeJVMRSSMemory" : 585252864,
+    "ProcessTreePythonVMemory" : 958914560,
+    "ProcessTreePythonRSSMemory" : 106622976,
+    "ProcessTreeOtherVMemory" : 0,
+    "ProcessTreeOtherRSSMemory" : 0,
+    "MinorGCCount" : 0,
+    "MinorGCTime" : 0,
+    "MajorGCCount" : 0,
+    "MajorGCTime" : 0
+  }
+}, {
+  "status" : "COMPLETE",
+  "stageId" : 3,
+  "attemptId" : 0,
+  "numTasks" : 2,
+  "numActiveTasks" : 0,
+  "numCompleteTasks" : 2,
+  "numFailedTasks" : 0,
+  "numKilledTasks" : 0,
+  "numCompletedIndices" : 2,
+  "executorRunTime" : 172,
+  "executorCpuTime" : 63170744,
+  "submissionTime" : "2018-10-02T00:42:59.163GMT",
+  "firstTaskLaunchedTime" : "2018-10-02T00:42:59.184GMT",
+  "completionTime" : "2018-10-02T00:42:59.444GMT",
+  "inputBytes" : 108,
+  "inputRecords" : 12,
+  "outputBytes" : 0,
+  "outputRecords" : 0,
+  "shuffleReadBytes" : 0,
+  "shuffleReadRecords" : 0,
+  "shuffleWriteBytes" : 533,
+  "shuffleWriteRecords" : 3,
+  "memoryBytesSpilled" : 0,
+  "diskBytesSpilled" : 0,
+  "name" : "filter at BisectingKMeans.scala:213",
+  "details" : "org.apache.spark.rdd.RDD.filter(RDD.scala:387)\norg.apache.spark.mllib.clustering.BisectingKMeans$$anonfun$run$1.apply$mcVI$sp(BisectingKMeans.scala:213)\nscala.collection.immutable.Range.foreach$mVc$sp(Range.scala:160)\norg.apache.spark.mllib.clustering.BisectingKMeans.run(BisectingKMeans.scala:210)\norg.apache.spark.mllib.clustering.BisectingKMeans.run(BisectingKMeans.scala:255)\norg.apache.spark.mllib.clustering.BisectingKMeans.run(BisectingKMeans.scala:261)\norg.apache.spark.mllib.api.python.PythonMLLibAPI.trainBisectingKMeans(PythonMLLibAPI.scala:135)\nsun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)\nsun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)\nsun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)\njava.lang.reflect.Method.invoke(Method.java:498)\npy4j.reflection.MethodInvoker.invoke(MethodInvoker.java:244)\npy4j.reflection.ReflectionEngine.invoke(ReflectionEngine.java:357)\npy4j.Gateway.invoke(Gateway.java:282)\npy4j.commands.AbstractCommand.invokeMethod(AbstractCommand.java:132)\npy4j.commands.CallCommand.execute(CallCommand.java:79)\npy4j.GatewayConnection.run(GatewayConnection.java:238)\njava.lang.Thread.run(Thread.java:745)",
+  "schedulingPool" : "default",
+  "rddIds" : [ 12, 8, 7, 0, 11, 2, 6, 3, 5, 1 ],
+  "accumulatorUpdates" : [ ],
+  "killedTasksSummary" : { },
+  "peakExecutorMetrics" : {
+    "JVMHeapMemory" : 318926040,
+    "JVMOffHeapMemory" : 96521592,
+    "OnHeapExecutionMemory" : 0,
+    "OffHeapExecutionMemory" : 0,
+    "OnHeapStorageMemory" : 483726,
+    "OffHeapStorageMemory" : 0,
+    "OnHeapUnifiedMemory" : 483726,
+    "OffHeapUnifiedMemory" : 0,
+    "DirectPoolMemory" : 135031,
+    "MappedPoolMemory" : 0,
+    "ProcessTreeJVMVMemory" : 4932550656,
+    "ProcessTreeJVMRSSMemory" : 569753600,
+    "ProcessTreePythonVMemory" : 958914560,
+    "ProcessTreePythonRSSMemory" : 106622976,
+    "ProcessTreeOtherVMemory" : 0,
+    "ProcessTreeOtherRSSMemory" : 0,
+    "MinorGCCount" : 0,
+    "MinorGCTime" : 0,
+    "MajorGCCount" : 0,
+    "MajorGCTime" : 0
+  }
+}, {
+  "status" : "COMPLETE",
+  "stageId" : 2,
+  "attemptId" : 0,
+  "numTasks" : 2,
+  "numActiveTasks" : 0,
+  "numCompleteTasks" : 2,
+  "numFailedTasks" : 0,
+  "numKilledTasks" : 0,
+  "numCompletedIndices" : 2,
+  "executorRunTime" : 148,
+  "executorCpuTime" : 103444383,
+  "submissionTime" : "2018-10-02T00:42:58.830GMT",
+  "firstTaskLaunchedTime" : "2018-10-02T00:42:58.844GMT",
+  "completionTime" : "2018-10-02T00:42:59.086GMT",
+  "inputBytes" : 0,
+  "inputRecords" : 0,
+  "outputBytes" : 0,
+  "outputRecords" : 0,
+  "shuffleReadBytes" : 356,
+  "shuffleReadRecords" : 2,
+  "shuffleWriteBytes" : 0,
+  "shuffleWriteRecords" : 0,
+  "memoryBytesSpilled" : 0,
+  "diskBytesSpilled" : 0,
+  "name" : "collect at BisectingKMeans.scala:304",
+  "details" : "org.apache.spark.rdd.RDD.collect(RDD.scala:944)\norg.apache.spark.mllib.clustering.BisectingKMeans$.org$apache$spark$mllib$clustering$BisectingKMeans$$summarize(BisectingKMeans.scala:304)\norg.apache.spark.mllib.clustering.BisectingKMeans.run(BisectingKMeans.scala:171)\norg.apache.spark.mllib.clustering.BisectingKMeans.run(BisectingKMeans.scala:255)\norg.apache.spark.mllib.clustering.BisectingKMeans.run(BisectingKMeans.scala:261)\norg.apache.spark.mllib.api.python.PythonMLLibAPI.trainBisectingKMeans(PythonMLLibAPI.scala:135)\nsun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)\nsun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)\nsun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)\njava.lang.reflect.Method.invoke(Method.java:498)\npy4j.reflection.MethodInvoker.invoke(MethodInvoker.java:244)\npy4j.reflection.ReflectionEngine.invoke(ReflectionEngine.java:357)\npy4j.Gateway.invoke(Gateway.java:282)\npy4j.commands.AbstractCommand.invokeMethod(AbstractCommand.java:132)\npy4j.commands.CallCommand.execute(CallCommand.java:79)\npy4j.GatewayConnection.run(GatewayConnection.java:238)\njava.lang.Thread.run(Thread.java:745)",
+  "schedulingPool" : "default",
+  "rddIds" : [ 10, 9 ],
+  "accumulatorUpdates" : [ ],
+  "killedTasksSummary" : { },
+  "peakExecutorMetrics" : {
+    "JVMHeapMemory" : 303792496,
+    "JVMOffHeapMemory" : 95545824,
+    "OnHeapExecutionMemory" : 0,
+    "OffHeapExecutionMemory" : 0,
+    "OnHeapStorageMemory" : 463135,
+    "OffHeapStorageMemory" : 0,
+    "OnHeapUnifiedMemory" : 463135,
+    "OffHeapUnifiedMemory" : 0,
+    "DirectPoolMemory" : 135031,
+    "MappedPoolMemory" : 0,
+    "ProcessTreeJVMVMemory" : 4931497984,
+    "ProcessTreeJVMRSSMemory" : 549777408,
+    "ProcessTreePythonVMemory" : 958914560,
+    "ProcessTreePythonRSSMemory" : 106622976,
+    "ProcessTreeOtherVMemory" : 0,
+    "ProcessTreeOtherRSSMemory" : 0,
+    "MinorGCCount" : 0,
+    "MinorGCTime" : 0,
+    "MajorGCCount" : 0,
+    "MajorGCTime" : 0
+  }
+}, {
+  "status" : "COMPLETE",
+  "stageId" : 1,
+  "attemptId" : 0,
+  "numTasks" : 2,
+  "numActiveTasks" : 0,
+  "numCompleteTasks" : 2,
+  "numFailedTasks" : 0,
+  "numKilledTasks" : 0,
+  "numCompletedIndices" : 2,
+  "executorRunTime" : 825,
+  "executorCpuTime" : 238787567,
+  "submissionTime" : "2018-10-02T00:42:57.793GMT",
+  "firstTaskLaunchedTime" : "2018-10-02T00:42:57.816GMT",
+  "completionTime" : "2018-10-02T00:42:58.821GMT",
+  "inputBytes" : 144,
+  "inputRecords" : 12,
+  "outputBytes" : 0,
+  "outputRecords" : 0,
+  "shuffleReadBytes" : 0,
+  "shuffleReadRecords" : 0,
+  "shuffleWriteBytes" : 356,
+  "shuffleWriteRecords" : 2,
+  "memoryBytesSpilled" : 0,
+  "diskBytesSpilled" : 0,
+  "name" : "map at BisectingKMeans.scala:170",
+  "details" : "org.apache.spark.rdd.RDD.map(RDD.scala:370)\norg.apache.spark.mllib.clustering.BisectingKMeans.run(BisectingKMeans.scala:170)\norg.apache.spark.mllib.clustering.BisectingKMeans.run(BisectingKMeans.scala:255)\norg.apache.spark.mllib.clustering.BisectingKMeans.run(BisectingKMeans.scala:261)\norg.apache.spark.mllib.api.python.PythonMLLibAPI.trainBisectingKMeans(PythonMLLibAPI.scala:135)\nsun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)\nsun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)\nsun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)\njava.lang.reflect.Method.invoke(Method.java:498)\npy4j.reflection.MethodInvoker.invoke(MethodInvoker.java:244)\npy4j.reflection.ReflectionEngine.invoke(ReflectionEngine.java:357)\npy4j.Gateway.invoke(Gateway.java:282)\npy4j.commands.AbstractCommand.invokeMethod(AbstractCommand.java:132)\npy4j.commands.CallCommand.execute(CallCommand.java:79)\npy4j.GatewayConnection.run(GatewayConnection.java:238)\njava.lang.Thread.run(Thread.java:745)",
+  "schedulingPool" : "default",
+  "rddIds" : [ 8, 7, 0, 2, 6, 3, 5, 1 ],
+  "accumulatorUpdates" : [ ],
+  "killedTasksSummary" : { },
+  "peakExecutorMetrics" : {
+    "JVMHeapMemory" : 292935952,
+    "JVMOffHeapMemory" : 95141200,
+    "OnHeapExecutionMemory" : 0,
+    "OffHeapExecutionMemory" : 0,
+    "OnHeapStorageMemory" : 1492038,
+    "OffHeapStorageMemory" : 0,
+    "OnHeapUnifiedMemory" : 1492038,
+    "OffHeapUnifiedMemory" : 0,
+    "DirectPoolMemory" : 135031,
+    "MappedPoolMemory" : 0,
+    "ProcessTreeJVMVMemory" : 4929392640,
+    "ProcessTreeJVMRSSMemory" : 539996160,
+    "ProcessTreePythonVMemory" : 958914560,
+    "ProcessTreePythonRSSMemory" : 106622976,
+    "ProcessTreeOtherVMemory" : 0,
+    "ProcessTreeOtherRSSMemory" : 0,
+    "MinorGCCount" : 0,
+    "MinorGCTime" : 0,
+    "MajorGCCount" : 0,
+    "MajorGCTime" : 0
+  }
+}, {
+  "status" : "COMPLETE",
+  "stageId" : 0,
+  "attemptId" : 0,
+  "numTasks" : 1,
+  "numActiveTasks" : 0,
+  "numCompleteTasks" : 1,
+  "numFailedTasks" : 0,
+  "numKilledTasks" : 0,
+  "numCompletedIndices" : 1,
+  "executorRunTime" : 2307,
+  "executorCpuTime" : 1105071149,
+  "submissionTime" : "2018-10-02T00:42:49.044GMT",
+  "firstTaskLaunchedTime" : "2018-10-02T00:42:53.735GMT",
+  "completionTime" : "2018-10-02T00:42:57.644GMT",
+  "inputBytes" : 72,
+  "inputRecords" : 4,
+  "outputBytes" : 0,
+  "outputRecords" : 0,
+  "shuffleReadBytes" : 0,
+  "shuffleReadRecords" : 0,
+  "shuffleWriteBytes" : 0,
+  "shuffleWriteRecords" : 0,
+  "memoryBytesSpilled" : 0,
+  "diskBytesSpilled" : 0,
+  "name" : "first at BisectingKMeans.scala:163",
+  "details" : "org.apache.spark.rdd.RDD.first(RDD.scala:1377)\norg.apache.spark.mllib.clustering.BisectingKMeans.run(BisectingKMeans.scala:163)\norg.apache.spark.mllib.clustering.BisectingKMeans.run(BisectingKMeans.scala:255)\norg.apache.spark.mllib.clustering.BisectingKMeans.run(BisectingKMeans.scala:261)\norg.apache.spark.mllib.api.python.PythonMLLibAPI.trainBisectingKMeans(PythonMLLibAPI.scala:135)\nsun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)\nsun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)\nsun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)\njava.lang.reflect.Method.invoke(Method.java:498)\npy4j.reflection.MethodInvoker.invoke(MethodInvoker.java:244)\npy4j.reflection.ReflectionEngine.invoke(ReflectionEngine.java:357)\npy4j.Gateway.invoke(Gateway.java:282)\npy4j.commands.AbstractCommand.invokeMethod(AbstractCommand.java:132)\npy4j.commands.CallCommand.execute(CallCommand.java:79)\npy4j.GatewayConnection.run(GatewayConnection.java:238)\njava.lang.Thread.run(Thread.java:745)",
+  "schedulingPool" : "default",
+  "rddIds" : [ 4, 0, 2, 3, 1 ],
+  "accumulatorUpdates" : [ ],
+  "killedTasksSummary" : { },
+  "peakExecutorMetrics" : {
+    "JVMHeapMemory" : 256071440,
+    "JVMOffHeapMemory" : 92211424,
+    "OnHeapExecutionMemory" : 0,
+    "OffHeapExecutionMemory" : 0,
+    "OnHeapStorageMemory" : 1086483,
+    "OffHeapStorageMemory" : 0,
+    "OnHeapUnifiedMemory" : 1086483,
+    "OffHeapUnifiedMemory" : 0,
+    "DirectPoolMemory" : 134726,
+    "MappedPoolMemory" : 0,
+    "ProcessTreeJVMVMemory" : 4926242816,
+    "ProcessTreeJVMRSSMemory" : 525656064,
+    "ProcessTreePythonVMemory" : 626200576,
+    "ProcessTreePythonRSSMemory" : 69218304,
+    "ProcessTreeOtherVMemory" : 0,
+    "ProcessTreeOtherRSSMemory" : 0,
+    "MinorGCCount" : 0,
+    "MinorGCTime" : 0,
+    "MajorGCCount" : 0,
+    "MajorGCTime" : 0
+  }
+} ]
diff --git a/core/src/test/resources/HistoryServerExpectations/stage_with_peak_metrics_expectation.json b/core/src/test/resources/HistoryServerExpectations/stage_with_peak_metrics_expectation.json
new file mode 100644
index 0000000000000..1690c95d9ea7c
--- /dev/null
+++ b/core/src/test/resources/HistoryServerExpectations/stage_with_peak_metrics_expectation.json
@@ -0,0 +1,189 @@
+{
+  "status" : "COMPLETE",
+  "stageId" : 0,
+  "attemptId" : 0,
+  "numTasks" : 1,
+  "numActiveTasks" : 0,
+  "numCompleteTasks" : 1,
+  "numFailedTasks" : 0,
+  "numKilledTasks" : 0,
+  "numCompletedIndices" : 1,
+  "executorRunTime" : 2307,
+  "executorCpuTime" : 1105071149,
+  "submissionTime" : "2018-10-02T00:42:49.044GMT",
+  "firstTaskLaunchedTime" : "2018-10-02T00:42:53.735GMT",
+  "completionTime" : "2018-10-02T00:42:57.644GMT",
+  "inputBytes" : 72,
+  "inputRecords" : 4,
+  "outputBytes" : 0,
+  "outputRecords" : 0,
+  "shuffleReadBytes" : 0,
+  "shuffleReadRecords" : 0,
+  "shuffleWriteBytes" : 0,
+  "shuffleWriteRecords" : 0,
+  "memoryBytesSpilled" : 0,
+  "diskBytesSpilled" : 0,
+  "name" : "first at BisectingKMeans.scala:163",
+  "details" : "org.apache.spark.rdd.RDD.first(RDD.scala:1377)\norg.apache.spark.mllib.clustering.BisectingKMeans.run(BisectingKMeans.scala:163)\norg.apache.spark.mllib.clustering.BisectingKMeans.run(BisectingKMeans.scala:255)\norg.apache.spark.mllib.clustering.BisectingKMeans.run(BisectingKMeans.scala:261)\norg.apache.spark.mllib.api.python.PythonMLLibAPI.trainBisectingKMeans(PythonMLLibAPI.scala:135)\nsun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)\nsun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)\nsun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)\njava.lang.reflect.Method.invoke(Method.java:498)\npy4j.reflection.MethodInvoker.invoke(MethodInvoker.java:244)\npy4j.reflection.ReflectionEngine.invoke(ReflectionEngine.java:357)\npy4j.Gateway.invoke(Gateway.java:282)\npy4j.commands.AbstractCommand.invokeMethod(AbstractCommand.java:132)\npy4j.commands.CallCommand.execute(CallCommand.java:79)\npy4j.GatewayConnection.run(GatewayConnection.java:238)\njava.lang.Thread.run(Thread.java:745)",
+  "schedulingPool" : "default",
+  "rddIds" : [ 4, 0, 2, 3, 1 ],
+  "accumulatorUpdates" : [ ],
+  "tasks" : {
+    "0" : {
+      "taskId" : 0,
+      "index" : 0,
+      "attempt" : 0,
+      "launchTime" : "2018-10-02T00:42:53.735GMT",
+      "duration" : 3893,
+      "executorId" : "1",
+      "host" : "rezamemory-2.gce.something.com",
+      "status" : "SUCCESS",
+      "taskLocality" : "NODE_LOCAL",
+      "speculative" : false,
+      "accumulatorUpdates" : [ ],
+      "taskMetrics" : {
+        "executorDeserializeTime" : 1322,
+        "executorDeserializeCpuTime" : 651096062,
+        "executorRunTime" : 2307,
+        "executorCpuTime" : 1105071149,
+        "resultSize" : 1448,
+        "jvmGcTime" : 208,
+        "resultSerializationTime" : 1,
+        "memoryBytesSpilled" : 0,
+        "diskBytesSpilled" : 0,
+        "peakExecutionMemory" : 0,
+        "inputMetrics" : {
+          "bytesRead" : 72,
+          "recordsRead" : 4
+        },
+        "outputMetrics" : {
+          "bytesWritten" : 0,
+          "recordsWritten" : 0
+        },
+        "shuffleReadMetrics" : {
+          "remoteBlocksFetched" : 0,
+          "localBlocksFetched" : 0,
+          "fetchWaitTime" : 0,
+          "remoteBytesRead" : 0,
+          "remoteBytesReadToDisk" : 0,
+          "localBytesRead" : 0,
+          "recordsRead" : 0
+        },
+        "shuffleWriteMetrics" : {
+          "bytesWritten" : 0,
+          "writeTime" : 0,
+          "recordsWritten" : 0
+        }
+      },
+      "executorLogs" : {
+        "stdout" : "http://rezamemory-2.gce.something.com:8042/node/containerlogs/container_1538416563558_0014_01_000002/root/stdout?start=-4096",
+        "stderr" : "http://rezamemory-2.gce.something.com:8042/node/containerlogs/container_1538416563558_0014_01_000002/root/stderr?start=-4096"
+      },
+      "schedulerDelay" : 263,
+      "gettingResultTime" : 0
+    }
+  },
+  "executorSummary" : {
+    "1" : {
+      "taskTime" : 3893,
+      "failedTasks" : 0,
+      "succeededTasks" : 1,
+      "killedTasks" : 0,
+      "inputBytes" : 72,
+      "inputRecords" : 4,
+      "outputBytes" : 0,
+      "outputRecords" : 0,
+      "shuffleRead" : 0,
+      "shuffleReadRecords" : 0,
+      "shuffleWrite" : 0,
+      "shuffleWriteRecords" : 0,
+      "memoryBytesSpilled" : 0,
+      "diskBytesSpilled" : 0,
+      "isBlacklistedForStage" : false,
+      "peakMemoryMetrics" : {
+        "JVMHeapMemory" : 182536928,
+        "JVMOffHeapMemory" : 58263224,
+        "OnHeapExecutionMemory" : 0,
+        "OffHeapExecutionMemory" : 0,
+        "OnHeapStorageMemory" : 1086483,
+        "OffHeapStorageMemory" : 0,
+        "OnHeapUnifiedMemory" : 1086483,
+        "OffHeapUnifiedMemory" : 0,
+        "DirectPoolMemory" : 20304,
+        "MappedPoolMemory" : 0,
+        "ProcessTreeJVMVMemory" : 3009855488,
+        "ProcessTreeJVMRSSMemory" : 404488192,
+        "ProcessTreePythonVMemory" : 626200576,
+        "ProcessTreePythonRSSMemory" : 69218304,
+        "ProcessTreeOtherVMemory" : 0,
+        "ProcessTreeOtherRSSMemory" : 0,
+        "MinorGCCount" : 0,
+        "MinorGCTime" : 0,
+        "MajorGCCount" : 0,
+        "MajorGCTime" : 0
+      }
+    },
+    "driver" : {
+      "taskTime" : 0,
+      "failedTasks" : 0,
+      "succeededTasks" : 0,
+      "killedTasks" : 0,
+      "inputBytes" : 0,
+      "inputRecords" : 0,
+      "outputBytes" : 0,
+      "outputRecords" : 0,
+      "shuffleRead" : 0,
+      "shuffleReadRecords" : 0,
+      "shuffleWrite" : 0,
+      "shuffleWriteRecords" : 0,
+      "memoryBytesSpilled" : 0,
+      "diskBytesSpilled" : 0,
+      "isBlacklistedForStage" : false,
+      "peakMemoryMetrics" : {
+        "JVMHeapMemory" : 256071440,
+        "JVMOffHeapMemory" : 92211424,
+        "OnHeapExecutionMemory" : 0,
+        "OffHeapExecutionMemory" : 0,
+        "OnHeapStorageMemory" : 333371,
+        "OffHeapStorageMemory" : 0,
+        "OnHeapUnifiedMemory" : 333371,
+        "OffHeapUnifiedMemory" : 0,
+        "DirectPoolMemory" : 134726,
+        "MappedPoolMemory" : 0,
+        "ProcessTreeJVMVMemory" : 4926242816,
+        "ProcessTreeJVMRSSMemory" : 525656064,
+        "ProcessTreePythonVMemory" : 408375296,
+        "ProcessTreePythonRSSMemory" : 40284160,
+        "ProcessTreeOtherVMemory" : 0,
+        "ProcessTreeOtherRSSMemory" : 0,
+        "MinorGCCount" : 0,
+        "MinorGCTime" : 0,
+        "MajorGCCount" : 0,
+        "MajorGCTime" : 0
+      }
+    }
+  },
+  "killedTasksSummary" : { },
+  "peakExecutorMetrics" : {
+    "JVMHeapMemory" : 256071440,
+    "JVMOffHeapMemory" : 92211424,
+    "OnHeapExecutionMemory" : 0,
+    "OffHeapExecutionMemory" : 0,
+    "OnHeapStorageMemory" : 1086483,
+    "OffHeapStorageMemory" : 0,
+    "OnHeapUnifiedMemory" : 1086483,
+    "OffHeapUnifiedMemory" : 0,
+    "DirectPoolMemory" : 134726,
+    "MappedPoolMemory" : 0,
+    "ProcessTreeJVMVMemory" : 4926242816,
+    "ProcessTreeJVMRSSMemory" : 525656064,
+    "ProcessTreePythonVMemory" : 626200576,
+    "ProcessTreePythonRSSMemory" : 69218304,
+    "ProcessTreeOtherVMemory" : 0,
+    "ProcessTreeOtherRSSMemory" : 0,
+    "MinorGCCount" : 0,
+    "MinorGCTime" : 0,
+    "MajorGCCount" : 0,
+    "MajorGCTime" : 0
+  }
+}
diff --git a/core/src/test/scala/org/apache/spark/deploy/history/HistoryServerSuite.scala b/core/src/test/scala/org/apache/spark/deploy/history/HistoryServerSuite.scala
index c99bcf2de5b0a..7e8024c13d75e 100644
--- a/core/src/test/scala/org/apache/spark/deploy/history/HistoryServerSuite.scala
+++ b/core/src/test/scala/org/apache/spark/deploy/history/HistoryServerSuite.scala
@@ -168,6 +168,8 @@ class HistoryServerSuite extends SparkFunSuite with BeforeAndAfter with Matchers
       "applications/local-1426533911241/2/stages/0/0/taskList",
     "blacklisting for stage" -> "applications/app-20180109111548-0000/stages/0/0",
     "blacklisting node for stage" -> "applications/application_1516285256255_0012/stages/0/0",
+    "stage list with peak metrics" -> "applications/application_1538416563558_0014/stages",
+    "stage with peak metrics" -> "applications/application_1538416563558_0014/stages/0/0",
 
     "rdd list storage json" -> "applications/local-1422981780767/storage/rdd",
     "executor node blacklisting" -> "applications/app-20161116163331-0000/executors",
diff --git a/core/src/test/scala/org/apache/spark/status/AppStatusListenerSuite.scala b/core/src/test/scala/org/apache/spark/status/AppStatusListenerSuite.scala
index b5800661efa7f..37e15eadd521a 100644
--- a/core/src/test/scala/org/apache/spark/status/AppStatusListenerSuite.scala
+++ b/core/src/test/scala/org/apache/spark/status/AppStatusListenerSuite.scala
@@ -1374,6 +1374,11 @@ class AppStatusListenerSuite extends SparkFunSuite with BeforeAndAfter {
     }
   }
 
+  /** expected stage executor metrics */
+  private case class StageExecutorMetrics(
+      peakExecutorMetrics: ExecutorMetrics,
+      executorMetrics: Map[String, ExecutorMetrics])
+
   test("executor metrics updates") {
     val listener = new AppStatusListener(store, conf, true)
 
@@ -1460,14 +1465,29 @@ class AppStatusListenerSuite extends SparkFunSuite with BeforeAndAfter {
         assert(exec.info.id === id)
         exec.info.peakMemoryMetrics match {
           case Some(actual) =>
-            ExecutorMetricType.metricToOffset.foreach { metric =>
-              assert(actual.getMetricValue(metric._1) === metrics.getMetricValue(metric._1))
-            }
+            checkExecutorMetrics(metrics, actual)
           case _ =>
             assert(false)
         }
       }
     }
+
+    // check stage level executor metrics
+    val expectedStageValues = Map(
+      0 -> StageExecutorMetrics(
+        new ExecutorMetrics(Array(7000L, 80L, 50L, 20L, 50L, 10L, 100L, 30L,
+          80L, 40L, 9000L, 4000L, 8000L, 3000L, 7000L, 2000L)),
+        Map("1" -> new ExecutorMetrics(Array(5000L, 50L, 50L, 20L, 50L, 10L, 100L, 30L,
+          70L, 20L, 8000L, 4000L, 7000L, 3000L, 6000L, 2000L)),
+          "2" -> new ExecutorMetrics(Array(7000L, 80L, 50L, 20L, 10L, 10L, 50L, 30L,
+            80L, 40L, 9000L, 4000L, 8000L, 3000L, 7000L, 2000L)))),
+      1 -> StageExecutorMetrics(new ExecutorMetrics(Array(7000L, 80L, 50L, 40L, 60L, 30L, 80L, 60L,
+        50L, 40L, 8000L, 5000L, 7000L, 4000L, 6000L, 3000L)),
+        Map("1" -> new ExecutorMetrics(Array(7000L, 70L, 50L, 30L, 60L, 30L, 80L, 55L,
+          50L, 0L, 5000L, 3000L, 4000L, 2000L, 3000L, 1000L)),
+          "2" -> new ExecutorMetrics(Array(7000L, 80L, 50L, 40L, 10L, 30L, 50L, 60L,
+            40L, 40L, 8000L, 5000L, 7000L, 4000L, 6000L, 3000L)))))
+    checkStageExecutorMetrics(expectedStageValues)
   }
 
   test("stage executor metrics") {
@@ -1510,14 +1530,66 @@ class AppStatusListenerSuite extends SparkFunSuite with BeforeAndAfter {
         assert(exec.info.id === id)
         exec.info.peakMemoryMetrics match {
           case Some(actual) =>
-            ExecutorMetricType.metricToOffset.foreach { metric =>
-              assert(actual.getMetricValue(metric._1) === metrics.getMetricValue(metric._1))
-            }
+            checkExecutorMetrics(metrics, actual)
           case _ =>
             assert(false)
         }
       }
     }
+
+    // check stage level executor metrics
+    val expectedStageValues = Map(
+      0 -> StageExecutorMetrics(
+        new ExecutorMetrics(Array(7000L, 70L, 50L, 20L, 50L, 10L, 100L, 30L,
+          80L, 40L, 9000L, 4000L, 8000L, 3000L, 7000L, 2000L)),
+        Map("1" -> new ExecutorMetrics(Array(5000L, 50L, 50L, 20L, 50L, 10L, 100L, 30L,
+            70L, 20L, 8000L, 4000L, 7000L, 3000L, 6000L, 2000L)),
+          "2" -> new ExecutorMetrics(Array(7000L, 70L, 50L, 20L, 10L, 10L, 50L, 30L,
+            80L, 40L, 9000L, 4000L, 8000L, 3000L, 7000L, 2000L)))),
+      1 -> StageExecutorMetrics(new ExecutorMetrics(Array(7000L, 80L, 50L, 40L, 60L, 30L, 80L, 60L,
+        50L, 40L, 8000L, 5000L, 7000L, 4000L, 6000L, 3000L)),
+        Map("1" -> new ExecutorMetrics(Array(7000L, 70L, 50L, 30L, 60L, 30L, 80L, 55L,
+          50L, 0L, 5000L, 3000L, 4000L, 2000L, 3000L, 1000L)),
+          "2" -> new ExecutorMetrics(Array(7000L, 80L, 50L, 40L, 10L, 30L, 50L, 60L,
+            40L, 40L, 8000L, 5000L, 7000L, 4000L, 6000L, 3000L)))))
+    checkStageExecutorMetrics(expectedStageValues)
+  }
+
+  private def checkExecutorMetrics(expected: ExecutorMetrics, actual: ExecutorMetrics): Unit = {
+    ExecutorMetricType.metricToOffset.foreach { metric =>
+      assert(actual.getMetricValue(metric._1) === expected.getMetricValue(metric._1))
+    }
+  }
+
+  /** check stage level peak executor metric values, and executor peak values for each stage */
+  private def checkStageExecutorMetrics(expectedStageValues: Map[Int, StageExecutorMetrics]) = {
+    // check stage level peak executor metric values for each stage
+    for ((stageId, expectedMetrics) <- expectedStageValues) {
+      check[StageDataWrapper](Array(stageId, 0)) { stage =>
+        stage.info.peakExecutorMetrics match {
+          case Some(actual) =>
+            checkExecutorMetrics(expectedMetrics.peakExecutorMetrics, actual)
+          case None =>
+            assert(false)
+        }
+      }
+    }
+
+    // check peak executor metric values for each stage and executor
+    val stageExecSummaries = store.view(classOf[ExecutorStageSummaryWrapper]).asScala.toSeq
+    stageExecSummaries.foreach { exec =>
+      expectedStageValues.get(exec.stageId) match {
+        case Some(stageValue) =>
+          (stageValue.executorMetrics.get(exec.executorId), exec.info.peakMemoryMetrics) match {
+            case (Some(expected), Some(actual)) =>
+              checkExecutorMetrics(expected, actual)
+            case _ =>
+              assert(false)
+          }
+        case None =>
+          assert(false)
+      }
+    }
   }
 
   test("storage information on executor lost/down") {
diff --git a/core/src/test/scala/org/apache/spark/ui/StagePageSuite.scala b/core/src/test/scala/org/apache/spark/ui/StagePageSuite.scala
index 5e976ae4e91da..a6ddece7026dc 100644
--- a/core/src/test/scala/org/apache/spark/ui/StagePageSuite.scala
+++ b/core/src/test/scala/org/apache/spark/ui/StagePageSuite.scala
@@ -78,7 +78,8 @@ class StagePageSuite extends SparkFunSuite with LocalSparkContext {
         accumulatorUpdates = Seq(new UIAccumulableInfo(0L, "acc", None, "value")),
         tasks = None,
         executorSummary = None,
-        killedTasksSummary = Map.empty
+        killedTasksSummary = Map.empty,
+        peakExecutorMetrics = None
       )
       val taskTable = new TaskPagedTable(
         stageData,