[SPARK-23429][CORE] Add executor memory metrics to heartbeat and expose in executors REST API #21221

edwinalu · 2018-05-02T23:57:26Z

Add new executor level memory metrics (JVM used memory, on/off heap execution memory, on/off heap storage memory, on/off heap unified memory, direct memory, and mapped memory), and expose via the executors REST API. This information will help provide insight into how executor and driver JVM memory is used, and for the different memory regions. It can be used to help determine good values for spark.executor.memory, spark.driver.memory, spark.memory.fraction, and spark.memory.storageFraction.

What changes were proposed in this pull request?

An ExecutorMetrics class is added, with jvmUsedHeapMemory, jvmUsedNonHeapMemory, onHeapExecutionMemory, offHeapExecutionMemory, onHeapStorageMemory, and offHeapStorageMemory, onHeapUnifiedMemory, offHeapUnifiedMemory, directMemory and mappedMemory. The new ExecutorMetrics is sent by executors to the driver as part of the Heartbeat. A heartbeat is added for the driver as well, to collect these metrics for the driver.

The EventLoggingListener store information about the peak values for each metric, per active stage and executor. When a StageCompleted event is seen, a StageExecutorsMetrics event will be logged for each executor, with peak values for the stage.

The AppStatusListener records the peak values for each memory metric.

The new memory metrics are added to the executors REST API.

How was this patch tested?

New unit tests have been added. This was also tested on our cluster.

…xecutors REST API Add new executor level memory metrics (JVM used memory, on/off heap execution memory, on/off heap storage memory), and expose via the executors REST API. This information will help provide insight into how executor and driver JVM memory is used, and for the different memory regions. It can be used to help determine good values for spark.executor.memory, spark.driver.memory, spark.memory.fraction, and spark.memory.storageFraction. Add an ExecutorMetrics class, with jvmUsedMemory, onHeapExecutionMemory, offHeapExecutionMemory, onHeapStorageMemory, and offHeapStorageMemory. The new ExecutorMetrics will be sent by executors to the driver as part of Heartbeat. A heartbeat will be added for the driver as well, to collect these metrics for the driver. Modify the EventLoggingListener to log ExecutorMetricsUpdate events if there is a new peak value for any of the memory metrics for an executor and stage. Only the ExecutorMetrics will be logged, and not the TaskMetrics, to minimize additional logging. Modify the AppStatusListener to record the peak values for each memory metric. Add the new memory metrics to the executors REST API.

…etricsUpdate

squito · 2018-05-03T02:04:33Z

Jenkins, ok to test

SparkQA · 2018-05-03T05:22:30Z

Test build #90087 has finished for PR 21221 at commit ad10d28.

This patch fails Spark unit tests.
This patch merges cleanly.
This patch adds no public classes.

squito · 2018-05-09T18:50:03Z

Jenkins, retest this please

SparkQA · 2018-05-09T23:21:31Z

Test build #90422 has finished for PR 21221 at commit ad10d28.

This patch passes all tests.
This patch merges cleanly.
This patch adds no public classes.

Ngone51 · 2018-05-10T04:22:45Z

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

+  }
+
+  /** Reports heartbeat metrics for the driver. */
+  private def reportHeartBeat(): Unit = {


Why we need this for driver ? If spark run in local mode, there's a local executor, which will report heartbeat.

With cluster mode, including YARN, there isn't a local executor, so the metrics for the driver would not be collected. Perhaps this could be modified to skip this step for local mode.

With cluster mode, including YARN, there isn't a local executor, so the metrics for the driver would not be collected.

Yes. But the problem is can we use executor's getCurrentExecutorMetrics() method for collecting memory metrics for driver ? IIRC, driver do not acqurie memory from execution memory pool at least.

It's a bit redundant for fields that aren't used by the driver -- for the driver, execution memory gets set to 0.

Ngone51 · 2018-05-10T04:47:28Z

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

+ */
+private[spark] class Heartbeater(reportHeartbeat: () => Unit, intervalMs: Long) {
+  // Executor for the heartbeat task
+  private val heartbeater = ThreadUtils.newDaemonSingleThreadScheduledExecutor("driver-heartbeater")


I'm wondering should the prefix name of heartbeater thread be "executor-heartbeater" ?

How about "heartbeater", since it could be for the driver as well? Alternatively, we can also pass in the name to the constructor.

"pass in the name to the constructor" is better(if we do need to do this for the driver)

Ngone51 · 2018-05-10T04:50:16Z

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

+      (event.stageInfo.stageId, event.stageInfo.attemptNumber()))
+    executorMap.foreach {
+      executorEntry => {
+        for ((executorId, peakExecutorMetrics) <- executorEntry) {


How about case (executorId, peakExecutorMetrics) => ? It would be more readable.

The for loop (line 187) is going through the hashmap entries of executorId to peakExecutorMetrics, so there are multiple values. Could you please provide more detail for how "case (executorId, peakExecutorMetrics) =>" would work? If the for loop is OK, then I can add some comments.

I revisited the code, I think you're right. My mistake, sorry.

Ngone51 · 2018-05-10T05:44:19Z

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

@@ -169,6 +179,27 @@ private[spark] class EventLoggingListener(

  // Events that trigger a flush
  override def onStageCompleted(event: SparkListenerStageCompleted): Unit = {
+    // log the peak executor metrics for the stage, for each executor
+    val accumUpdates = new ArrayBuffer[(Long, Int, Int, Seq[AccumulableInfo])]()
+    val executorMap = liveStageExecutorMetrics.remove(


Do we always post a SparkListenerStageCompleted event for failed satges (I can't rememer clearly)? If not, I think we should clean up other attempts of the same stage here.

Yes, it's safer to clean up earlier attempts -- I can add some code to iterate through earlier attemptIDs.

Ngone51 · 2018-05-10T06:08:16Z

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

+ * Records the peak values for executor level metrics. If jvmUsedHeapMemory is -1, then no
+ * values have been recorded yet.
+ */
+private[spark] class PeakExecutorMetrics {


Do we really need this class? It seems ExecutorMetrics can already do the same work.

I got some errors when trying to add methods to ExecutorMetrics. I don't remember the details, but can try this again.

can you revisit this given the other refactoring that has taken place?

and if you do need this extra class, please include a comment here explaining the metrics array and referencing MetricGetter.

With ExecutorMetrics removed, it seems useful to have a class for tracking and setting peak metric values, that can be used by both EventLoggingListener and AppStatusListener.

Ngone51 · 2018-05-10T06:13:48Z

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

@@ -93,6 +94,10 @@ private[spark] class EventLoggingListener(
  // Visible for tests only.
  private[scheduler] val logPath = getLogPath(logBaseDir, appId, appAttemptId, compressionCodecName)

+  // map of live stages, to peak executor metrics for the stage
+  private val liveStageExecutorMetrics = mutable.HashMap[(Int, Int),


Why we should track executor's memory metrics for each stage?

This is tracking peak metric values for executors for each stage, so that the peak values for the stage can be dumped at stage end. The purpose is to reduce the amount of logging, to only number of stages * number of executors ExecutorMetricsUpdate events.

I originally tried logging for new peak values, resetting when a new stage begins -- this is simpler, but can lead to more events being logged.

Having stage level information is useful for users trying to identify which stages are more memory intensive. This information could be useful they are trying to reduce the amount of memory used, since they would know which stages (and the relevant code) to focus on.

…enabled to enable/disable executor metrics update logging. Code review comments.

SparkQA · 2018-05-15T05:08:26Z

Test build #90613 has finished for PR 21221 at commit 10ed328.

This patch passes all tests.
This patch merges cleanly.
This patch adds no public classes.

squito

some minor things as I do another round and page things back in

squito · 2018-05-23T18:03:19Z

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

+  def getCurrentExecutorMetrics(
+      memoryManager: MemoryManager,
+      direct: BufferPoolMXBean,
+      mapped: BufferPoolMXBean) : ExecutorMetrics = {


does it make more sense to move this inside Heartbeater? Then you don't need to pass in any BufferPoolMXBeans. also rename to "getCurrentMemoryMetrics"

Yes, and easier to share the code between driver and executor.

squito · 2018-05-23T18:05:55Z

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

@@ -81,7 +84,7 @@ private[spark] class EventLoggingListener(
  private val compressionCodecName = compressionCodec.map { c =>
    CompressionCodec.getShortName(c.getClass.getName)
  }
-
+logInfo("spark.eventLog.logExecutorMetricsUpdates.enabled is " + shouldLogExecutorMetricsUpdates)


doesn't really seem necessary at all, definitely not at INFO level (and indentation is wrong).

Removed. Thanks, I hadn't meant to push that.

squito · 2018-05-23T18:07:20Z

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

@@ -93,6 +96,10 @@ private[spark] class EventLoggingListener(
  // Visible for tests only.
  private[scheduler] val logPath = getLogPath(logBaseDir, appId, appAttemptId, compressionCodecName)

+  // map of live stages, to peak executor metrics for the stage
+  private val liveStageExecutorMetrics = mutable.HashMap[(Int, Int),
+    mutable.HashMap[String, PeakExecutorMetrics]]()


you could just import mutable.HashMap (added bonus -- fits on one line)

squito · 2018-05-23T18:20:11Z

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

+        liveStageExecutorMetrics.remove((event.stageInfo.stageId, attemptId))
+      }
+
+      // log the peak executor metrics for the stage, for each executor


I'd add a comment here that this will log metrics for all executors that were alive while the stage was running, whether or not they ran any tasks for that stage (I think that's what it will do here, right?)

Yes, it's all running executors, and does not filter based on if they have tasks for the stage. I've updated the comment.

squito · 2018-05-23T18:28:57Z

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

@@ -209,6 +210,16 @@ class DAGScheduler(
  private[scheduler] val eventProcessLoop = new DAGSchedulerEventProcessLoop(this)
  taskScheduler.setDAGScheduler(this)

+  /** driver heartbeat for collecting metrics */
+  private val heartbeater: Heartbeater = new Heartbeater(reportHeartBeat, "driver-heartbeater",


lets not put this in the DAGScheduler please -- this class is fragile enough as it is :)

I think this should just go in SparkContext.

squito · 2018-05-23T18:45:47Z

core/src/test/scala/org/apache/spark/scheduler/EventLoggingListenerSuite.scala

+    SparkListenerExecutorAdded(0L, executorId.toString, new ExecutorInfo("host1", 1, Map.empty))
+  }
+
+  /** Create an executor added event for the specified executor Id. */


added -> removed

though for that matter -- I'd just remove the doc comments on all these teeny helper methods

I'll remove -- they are pretty self-explanatory.

squito · 2018-05-23T19:00:10Z

core/src/test/scala/org/apache/spark/scheduler/EventLoggingListenerSuite.scala

+                i += 1
+             }
+            checkEvent(lines(i), event)
+            i += 1


I found this pretty confusing at first. I suggest renaming i to logIdx and including a comment about the j loop. Also we tend to use (1 to 2).foreach. eg.

// just before the SparkListenerStageCompleted gets logged, we expect to get a // SparkListenerExecutorMetricsUpdate for each executor (1 to 2).foreach { _ => checkExecutorMetricsUpdate(lines(logIdx), stageCompleted.stageInfo.stageId, expectedMetricsEvents) logIdx += 1 } // also check that we get the expected SparkListenerStageCompleted checkEvent(lines(logIdx), event) logIdx += 1

Changed for both.

squito · 2018-05-23T19:01:05Z

core/src/test/scala/org/apache/spark/scheduler/EventLoggingListenerSuite.scala

+    assert(line.contains(event.getClass.toString.split("\\.").last))
+    event match {
+      case executorMetrics: SparkListenerExecutorMetricsUpdate =>
+        JsonProtocol.sparkEventFromJson(parse(line)) match {


you can pull JsonProtocol.sparkEventFromJson(parse(line)) out to avoid repeating, along with the type comparison.

val parsed = JsonProtocol.sparkEventFromJson(parse(line)) assert(parsed.getClass === event.getClass) event match { ...

(also assertTypeError does something else entirely: http://doc.scalatest.org/2.2.6/#org.scalatest.Assertions)

Thanks, modified.

squito · 2018-05-23T19:04:12Z

core/src/test/scala/org/apache/spark/scheduler/EventLoggingListenerSuite.scala

+  private def checkEvent(line: String, event: SparkListenerEvent): Unit = {
+    assert(line.contains(event.getClass.toString.split("\\.").last))
+    event match {
+      case executorMetrics: SparkListenerExecutorMetricsUpdate =>


you're never using this w/ SparkListenerExecutorMetricsUpdate, right?

Nope, with the change in design to logging the executor metrics updates at stage end, this part is skipped -- I'll remove this.

squito · 2018-05-23T21:30:35Z

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

+      updated = true
+    }
+    if (executorMetrics.offHeapStorageMemory > _offHeapStorageMemory) {
+      _offHeapStorageMemory = executorMetrics.offHeapStorageMemory


I know spark has this kind of code all over the place already, but I really hate how error prone it is -- way too easy for a copy paste error to result in comparing the wrong two metrics, or updating the wrong value, or forgetting to update this when another metric is added, etc.

I just opened this edwinalu#1 as another way to do this that would eliminate a ton of boilerplate IMO.

Thanks! This is cleaner, and will make it easier to add new metrics. It is very easy to have a copy/paste error. I can merge and make the test changes -- let me know if that sounds good, or if you'd like to make some more changes first.

The more you can take it over from here, the better :) But let me know if there is anything which is confusing, or if the TODOs that I've left actually don't seem possible etc. and I can take a closer look.

Will do. Thanks!

Metric enums

SparkQA · 2018-06-03T02:47:54Z

Test build #91424 has finished for PR 21221 at commit 7879e66.

This patch fails to build.
This patch does not merge cleanly.
This patch adds the following public classes (experimental):
sealed trait MetricGetter
abstract class MemoryManagerMetricGetter(f: MemoryManager => Long) extends MetricGetter
abstract class MBeanMetricGetter(mBeanName: String) extends MetricGetter

felixcheung · 2018-06-10T19:25:54Z

ok to test

felixcheung · 2018-06-10T19:26:11Z

probably need to be rebased

SparkQA · 2018-06-10T19:42:46Z

Test build #91642 has finished for PR 21221 at commit 7879e66.

This patch fails to build.
This patch does not merge cleanly.
This patch adds the following public classes (experimental):
sealed trait MetricGetter
abstract class MemoryManagerMetricGetter(f: MemoryManager => Long) extends MetricGetter
abstract class MBeanMetricGetter(mBeanName: String) extends MetricGetter

edwinalu · 2018-06-11T01:52:57Z

@squito , I'm modifying ExecutorMetrics to take in the metrics array -- this will be easier for tests where we pass in set values, and seems fine for the actual code. It will check that the length of the passed in array is the same as MetricGetter.values.length. Let me know if you have any concerns.

@felixcheung , I'll finish the current changes, then rebase.

edwinalu · 2018-06-11T12:53:18Z

@squito For PeakMemoryMetrics in api.scala, changing to the array gives REST API output of:

"peakMemoryMetrics" : {
"metrics" : [ 755008624, 100519936, 0, 0, 47962185, 0, 47962185, 0, 98230, 0 ]
}

instead of:

"peakMemoryMetrics" : {
"jvmUsedHeapMemory" : 629553808,
"jvmUsedNonHeapMemory" : 205304696,
"onHeapExecutionMemory" : 0,
"offHeapExecutionMemory" : 0,
"onHeapStorageMemory" : 905801,
"offHeapStorageMemory" : 0,
"onHeapUnifiedMemory" : 905801,
"offHeapUnifiedMemory" : 0,
"directMemory" : 397602,
"mappedMemory" : 0
}

Would it be OK to revert back to the original version of PeakMemoryMetrics, where each field is listed as a separate element?

squito · 2018-06-11T17:19:51Z

well, I think you should change the way PeakExecutorMetrics gets converted to json, so that it uses a name from the relevant MetricGetter. You should be able to customize the way it gets converted to json here:

https://github.com/apache/spark/blob/master/core/src/main/scala/org/apache/spark/status/api/v1/JacksonMessageWriter.scala#L50

… move logic for getting metrics to Heartbeater), and modifiy tests for the new ExecutorMetrics format.

…xecutors REST API Add new executor level memory metrics (JVM used memory, on/off heap execution memory, on/off heap storage memory), and expose via the executors REST API. This information will help provide insight into how executor and driver JVM memory is used, and for the different memory regions. It can be used to help determine good values for spark.executor.memory, spark.driver.memory, spark.memory.fraction, and spark.memory.storageFraction. Add an ExecutorMetrics class, with jvmUsedMemory, onHeapExecutionMemory, offHeapExecutionMemory, onHeapStorageMemory, and offHeapStorageMemory. The new ExecutorMetrics will be sent by executors to the driver as part of Heartbeat. A heartbeat will be added for the driver as well, to collect these metrics for the driver. Modify the EventLoggingListener to log ExecutorMetricsUpdate events if there is a new peak value for any of the memory metrics for an executor and stage. Only the ExecutorMetrics will be logged, and not the TaskMetrics, to minimize additional logging. Modify the AppStatusListener to record the peak values for each memory metric. Add the new memory metrics to the executors REST API.

…etricsUpdate

…enabled to enable/disable executor metrics update logging. Code review comments.

felixcheung

LGTM, minor comments

felixcheung · 2018-08-16T06:53:16Z

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

@@ -216,8 +217,7 @@ private[spark] class Executor(

  def stop(): Unit = {
    env.metricsSystem.report()
-    heartbeater.shutdown()
-    heartbeater.awaitTermination(10, TimeUnit.SECONDS)
+    heartbeater.stop()


future: try {} catch { case NonFatal(e)?

felixcheung · 2018-08-16T06:54:31Z

core/src/main/scala/org/apache/spark/internal/config/package.scala

+  private[spark] val EVENT_LOG_STAGE_EXECUTOR_METRICS =
+    ConfigBuilder("spark.eventLog.logStageExecutorMetrics.enabled")
+      .booleanConf
+      .createWithDefault(true)


should this be "false" for now until we could test this out more, just to be on the safe side?

That would be safer. I'll change to false, and we can change change to true after people have had a chance to test it out.

felixcheung · 2018-08-16T06:58:46Z

Jenkins, retest this please

SparkQA · 2018-08-16T10:42:34Z

Test build #94842 has finished for PR 21221 at commit a14b82a.

This patch fails Spark unit tests.
This patch merges cleanly.
This patch adds no public classes.

SparkQA · 2018-08-17T00:48:46Z

Test build #94865 has finished for PR 21221 at commit 2897281.

This patch passes all tests.
This patch merges cleanly.
This patch adds no public classes.

mccheah · 2018-08-20T21:51:55Z

We're going to delay on merging this until after the 2.4 branch is cut. We can include this in Spark 2.5.

mccheah · 2018-09-06T22:29:25Z

@edwinalu - this can merge now that Spark 2.4's release branch has been cut, but there's conflicting files now. Can we clear the conflicts and then we can merge this?

SparkQA · 2018-09-07T05:23:36Z

Test build #95776 has finished for PR 21221 at commit ee4aa1d.

This patch passes all tests.
This patch merges cleanly.
This patch adds no public classes.

kiszk · 2018-09-07T09:25:01Z

core/src/main/java/org/apache/spark/SparkFirehoseListener.java

@@ -103,6 +103,12 @@ public final void onExecutorMetricsUpdate(
    onEvent(executorMetricsUpdate);
  }

+  @Override
+  public final void onStageExecutorMetrics(
+          SparkListenerStageExecutorMetrics executorMetrics) {


nit: remove extra spaces for better indent

SparkQA · 2018-09-07T17:38:05Z

Test build #95801 has finished for PR 21221 at commit 571285b.

This patch passes all tests.
This patch merges cleanly.
This patch adds no public classes.

mccheah · 2018-09-07T17:42:06Z

Thanks, I think this looks good. With a prior +1 from @felixcheung and @squito I'm going to merge this now. Let us know if there are any further concerns and we can follow up.

edwinalu · 2018-09-07T18:17:16Z

Thanks!

gatorsmile · 2018-09-30T15:52:58Z

For the other reviewers, this was merged to master (not 2.4)

gatorsmile · 2018-09-30T15:56:15Z

@mccheah When you merged the code, could you also leave the comments about which branches you did the merge?

### What changes were proposed in this pull request? At Apache Spark 3.0.0, [SPARK-23429](#21221) added the ability to collect executor metrics via heartbeats and to expose it as a REST API. This PR aims to extend it to support `Prometheus` format additionally. ### Why are the changes needed? Prometheus.io is a CNCF project used widely with K8s. - https://github.com/prometheus/prometheus ### Does this PR introduce any user-facing change? Yes. New web interfaces are added along with the existing JSON API. | | JSON End Point | Prometheus End Point | | ------- | ------------------------------------ | --------------------------------- | | Driver | /api/v1/applications/{id}/executors/ | /metrics/executors/prometheus/ | ### How was this patch tested? Manually connect to the new end-points with `curl` and compare with JSON. **SETUP** ``` $ sbin/start-master.sh $ sbin/start-slave.sh spark://`hostname`:7077 $ bin/spark-shell --master spark://`hostname`:7077 --conf spark.ui.prometheus.enabled=true ``` **JSON (existing after SPARK-23429)** ``` $ curl -s http://localhost:4040/api/v1/applications/app-20190911204823-0000/executors [ { "id" : "driver", "hostPort" : "localhost:52615", "isActive" : true, "rddBlocks" : 0, "memoryUsed" : 0, "diskUsed" : 0, "totalCores" : 0, "maxTasks" : 0, "activeTasks" : 0, "failedTasks" : 0, "completedTasks" : 0, "totalTasks" : 0, "totalDuration" : 0, "totalGCTime" : 0, "totalInputBytes" : 0, "totalShuffleRead" : 0, "totalShuffleWrite" : 0, "isBlacklisted" : false, "maxMemory" : 384093388, "addTime" : "2019-09-12T03:48:23.875GMT", "executorLogs" : { }, "memoryMetrics" : { "usedOnHeapStorageMemory" : 0, "usedOffHeapStorageMemory" : 0, "totalOnHeapStorageMemory" : 384093388, "totalOffHeapStorageMemory" : 0 }, "blacklistedInStages" : [ ], "peakMemoryMetrics" : { "JVMHeapMemory" : 229995952, "JVMOffHeapMemory" : 145872280, "OnHeapExecutionMemory" : 0, "OffHeapExecutionMemory" : 0, "OnHeapStorageMemory" : 0, "OffHeapStorageMemory" : 0, "OnHeapUnifiedMemory" : 0, "OffHeapUnifiedMemory" : 0, "DirectPoolMemory" : 75891, "MappedPoolMemory" : 0, "ProcessTreeJVMVMemory" : 0, "ProcessTreeJVMRSSMemory" : 0, "ProcessTreePythonVMemory" : 0, "ProcessTreePythonRSSMemory" : 0, "ProcessTreeOtherVMemory" : 0, "ProcessTreeOtherRSSMemory" : 0, "MinorGCCount" : 8, "MinorGCTime" : 82, "MajorGCCount" : 3, "MajorGCTime" : 128 }, "attributes" : { }, "resources" : { } }, { "id" : "0", "hostPort" : "127.0.0.1:52619", "isActive" : true, "rddBlocks" : 0, "memoryUsed" : 0, "diskUsed" : 0, "totalCores" : 16, "maxTasks" : 16, "activeTasks" : 0, "failedTasks" : 0, "completedTasks" : 0, "totalTasks" : 0, "totalDuration" : 0, "totalGCTime" : 0, "totalInputBytes" : 0, "totalShuffleRead" : 0, "totalShuffleWrite" : 0, "isBlacklisted" : false, "maxMemory" : 384093388, "addTime" : "2019-09-12T03:48:25.907GMT", "executorLogs" : { "stdout" : "http://127.0.0.1:8081/logPage/?appId=app-20190911204823-0000&executorId=0&logType=stdout", "stderr" : "http://127.0.0.1:8081/logPage/?appId=app-20190911204823-0000&executorId=0&logType=stderr" }, "memoryMetrics" : { "usedOnHeapStorageMemory" : 0, "usedOffHeapStorageMemory" : 0, "totalOnHeapStorageMemory" : 384093388, "totalOffHeapStorageMemory" : 0 }, "blacklistedInStages" : [ ], "attributes" : { }, "resources" : { } } ] ``` **Prometheus** ``` $ curl -s http://localhost:4040/metrics/executors/prometheus metrics_app_20190911204823_0000_driver_executor_rddBlocks_Count 0 metrics_app_20190911204823_0000_driver_executor_memoryUsed_Count 0 metrics_app_20190911204823_0000_driver_executor_diskUsed_Count 0 metrics_app_20190911204823_0000_driver_executor_totalCores_Count 0 metrics_app_20190911204823_0000_driver_executor_maxTasks_Count 0 metrics_app_20190911204823_0000_driver_executor_activeTasks_Count 0 metrics_app_20190911204823_0000_driver_executor_failedTasks_Count 0 metrics_app_20190911204823_0000_driver_executor_completedTasks_Count 0 metrics_app_20190911204823_0000_driver_executor_totalTasks_Count 0 metrics_app_20190911204823_0000_driver_executor_totalDuration_Value 0 metrics_app_20190911204823_0000_driver_executor_totalGCTime_Value 0 metrics_app_20190911204823_0000_driver_executor_totalInputBytes_Count 0 metrics_app_20190911204823_0000_driver_executor_totalShuffleRead_Count 0 metrics_app_20190911204823_0000_driver_executor_totalShuffleWrite_Count 0 metrics_app_20190911204823_0000_driver_executor_maxMemory_Count 384093388 metrics_app_20190911204823_0000_driver_executor_usedOnHeapStorageMemory_Count 0 metrics_app_20190911204823_0000_driver_executor_usedOffHeapStorageMemory_Count 0 metrics_app_20190911204823_0000_driver_executor_totalOnHeapStorageMemory_Count 384093388 metrics_app_20190911204823_0000_driver_executor_totalOffHeapStorageMemory_Count 0 metrics_app_20190911204823_0000_driver_executor_JVMHeapMemory_Count 230406336 metrics_app_20190911204823_0000_driver_executor_JVMOffHeapMemory_Count 146132592 metrics_app_20190911204823_0000_driver_executor_OnHeapExecutionMemory_Count 0 metrics_app_20190911204823_0000_driver_executor_OffHeapExecutionMemory_Count 0 metrics_app_20190911204823_0000_driver_executor_OnHeapStorageMemory_Count 0 metrics_app_20190911204823_0000_driver_executor_OffHeapStorageMemory_Count 0 metrics_app_20190911204823_0000_driver_executor_OnHeapUnifiedMemory_Count 0 metrics_app_20190911204823_0000_driver_executor_OffHeapUnifiedMemory_Count 0 metrics_app_20190911204823_0000_driver_executor_DirectPoolMemory_Count 97049 metrics_app_20190911204823_0000_driver_executor_MappedPoolMemory_Count 0 metrics_app_20190911204823_0000_driver_executor_ProcessTreeJVMVMemory_Count 0 metrics_app_20190911204823_0000_driver_executor_ProcessTreeJVMRSSMemory_Count 0 metrics_app_20190911204823_0000_driver_executor_ProcessTreePythonVMemory_Count 0 metrics_app_20190911204823_0000_driver_executor_ProcessTreePythonRSSMemory_Count 0 metrics_app_20190911204823_0000_driver_executor_ProcessTreeOtherVMemory_Count 0 metrics_app_20190911204823_0000_driver_executor_ProcessTreeOtherRSSMemory_Count 0 metrics_app_20190911204823_0000_driver_executor_MinorGCCount_Count 8 metrics_app_20190911204823_0000_driver_executor_MinorGCTime_Count 82 metrics_app_20190911204823_0000_driver_executor_MajorGCCount_Count 3 metrics_app_20190911204823_0000_driver_executor_MajorGCTime_Count 128 metrics_app_20190911204823_0000_0_executor_rddBlocks_Count 0 metrics_app_20190911204823_0000_0_executor_memoryUsed_Count 0 metrics_app_20190911204823_0000_0_executor_diskUsed_Count 0 metrics_app_20190911204823_0000_0_executor_totalCores_Count 16 metrics_app_20190911204823_0000_0_executor_maxTasks_Count 16 metrics_app_20190911204823_0000_0_executor_activeTasks_Count 0 metrics_app_20190911204823_0000_0_executor_failedTasks_Count 0 metrics_app_20190911204823_0000_0_executor_completedTasks_Count 0 metrics_app_20190911204823_0000_0_executor_totalTasks_Count 0 metrics_app_20190911204823_0000_0_executor_totalDuration_Value 0 metrics_app_20190911204823_0000_0_executor_totalGCTime_Value 0 metrics_app_20190911204823_0000_0_executor_totalInputBytes_Count 0 metrics_app_20190911204823_0000_0_executor_totalShuffleRead_Count 0 metrics_app_20190911204823_0000_0_executor_totalShuffleWrite_Count 0 metrics_app_20190911204823_0000_0_executor_maxMemory_Count 384093388 metrics_app_20190911204823_0000_0_executor_usedOnHeapStorageMemory_Count 0 metrics_app_20190911204823_0000_0_executor_usedOffHeapStorageMemory_Count 0 metrics_app_20190911204823_0000_0_executor_totalOnHeapStorageMemory_Count 384093388 metrics_app_20190911204823_0000_0_executor_totalOffHeapStorageMemory_Count 0 ``` Closes #25770 from dongjoon-hyun/SPARK-29064. Authored-by: Dongjoon Hyun <dhyun@apple.com> Signed-off-by: DB Tsai <d_tsai@apple.com>

### What changes were proposed in this pull request? At Apache Spark 3.0.0, [SPARK-23429](apache#21221) added the ability to collect executor metrics via heartbeats and to expose it as a REST API. This PR aims to extend it to support `Prometheus` format additionally. ### Why are the changes needed? Prometheus.io is a CNCF project used widely with K8s. - https://github.com/prometheus/prometheus ### Does this PR introduce any user-facing change? Yes. New web interfaces are added along with the existing JSON API. | | JSON End Point | Prometheus End Point | | ------- | ------------------------------------ | --------------------------------- | | Driver | /api/v1/applications/{id}/executors/ | /metrics/executors/prometheus/ | ### How was this patch tested? Manually connect to the new end-points with `curl` and compare with JSON. **SETUP** ``` $ sbin/start-master.sh $ sbin/start-slave.sh spark://`hostname`:7077 $ bin/spark-shell --master spark://`hostname`:7077 --conf spark.ui.prometheus.enabled=true ``` **JSON (existing after SPARK-23429)** ``` $ curl -s http://localhost:4040/api/v1/applications/app-20190911204823-0000/executors [ { "id" : "driver", "hostPort" : "localhost:52615", "isActive" : true, "rddBlocks" : 0, "memoryUsed" : 0, "diskUsed" : 0, "totalCores" : 0, "maxTasks" : 0, "activeTasks" : 0, "failedTasks" : 0, "completedTasks" : 0, "totalTasks" : 0, "totalDuration" : 0, "totalGCTime" : 0, "totalInputBytes" : 0, "totalShuffleRead" : 0, "totalShuffleWrite" : 0, "isBlacklisted" : false, "maxMemory" : 384093388, "addTime" : "2019-09-12T03:48:23.875GMT", "executorLogs" : { }, "memoryMetrics" : { "usedOnHeapStorageMemory" : 0, "usedOffHeapStorageMemory" : 0, "totalOnHeapStorageMemory" : 384093388, "totalOffHeapStorageMemory" : 0 }, "blacklistedInStages" : [ ], "peakMemoryMetrics" : { "JVMHeapMemory" : 229995952, "JVMOffHeapMemory" : 145872280, "OnHeapExecutionMemory" : 0, "OffHeapExecutionMemory" : 0, "OnHeapStorageMemory" : 0, "OffHeapStorageMemory" : 0, "OnHeapUnifiedMemory" : 0, "OffHeapUnifiedMemory" : 0, "DirectPoolMemory" : 75891, "MappedPoolMemory" : 0, "ProcessTreeJVMVMemory" : 0, "ProcessTreeJVMRSSMemory" : 0, "ProcessTreePythonVMemory" : 0, "ProcessTreePythonRSSMemory" : 0, "ProcessTreeOtherVMemory" : 0, "ProcessTreeOtherRSSMemory" : 0, "MinorGCCount" : 8, "MinorGCTime" : 82, "MajorGCCount" : 3, "MajorGCTime" : 128 }, "attributes" : { }, "resources" : { } }, { "id" : "0", "hostPort" : "127.0.0.1:52619", "isActive" : true, "rddBlocks" : 0, "memoryUsed" : 0, "diskUsed" : 0, "totalCores" : 16, "maxTasks" : 16, "activeTasks" : 0, "failedTasks" : 0, "completedTasks" : 0, "totalTasks" : 0, "totalDuration" : 0, "totalGCTime" : 0, "totalInputBytes" : 0, "totalShuffleRead" : 0, "totalShuffleWrite" : 0, "isBlacklisted" : false, "maxMemory" : 384093388, "addTime" : "2019-09-12T03:48:25.907GMT", "executorLogs" : { "stdout" : "http://127.0.0.1:8081/logPage/?appId=app-20190911204823-0000&executorId=0&logType=stdout", "stderr" : "http://127.0.0.1:8081/logPage/?appId=app-20190911204823-0000&executorId=0&logType=stderr" }, "memoryMetrics" : { "usedOnHeapStorageMemory" : 0, "usedOffHeapStorageMemory" : 0, "totalOnHeapStorageMemory" : 384093388, "totalOffHeapStorageMemory" : 0 }, "blacklistedInStages" : [ ], "attributes" : { }, "resources" : { } } ] ``` **Prometheus** ``` $ curl -s http://localhost:4040/metrics/executors/prometheus metrics_app_20190911204823_0000_driver_executor_rddBlocks_Count 0 metrics_app_20190911204823_0000_driver_executor_memoryUsed_Count 0 metrics_app_20190911204823_0000_driver_executor_diskUsed_Count 0 metrics_app_20190911204823_0000_driver_executor_totalCores_Count 0 metrics_app_20190911204823_0000_driver_executor_maxTasks_Count 0 metrics_app_20190911204823_0000_driver_executor_activeTasks_Count 0 metrics_app_20190911204823_0000_driver_executor_failedTasks_Count 0 metrics_app_20190911204823_0000_driver_executor_completedTasks_Count 0 metrics_app_20190911204823_0000_driver_executor_totalTasks_Count 0 metrics_app_20190911204823_0000_driver_executor_totalDuration_Value 0 metrics_app_20190911204823_0000_driver_executor_totalGCTime_Value 0 metrics_app_20190911204823_0000_driver_executor_totalInputBytes_Count 0 metrics_app_20190911204823_0000_driver_executor_totalShuffleRead_Count 0 metrics_app_20190911204823_0000_driver_executor_totalShuffleWrite_Count 0 metrics_app_20190911204823_0000_driver_executor_maxMemory_Count 384093388 metrics_app_20190911204823_0000_driver_executor_usedOnHeapStorageMemory_Count 0 metrics_app_20190911204823_0000_driver_executor_usedOffHeapStorageMemory_Count 0 metrics_app_20190911204823_0000_driver_executor_totalOnHeapStorageMemory_Count 384093388 metrics_app_20190911204823_0000_driver_executor_totalOffHeapStorageMemory_Count 0 metrics_app_20190911204823_0000_driver_executor_JVMHeapMemory_Count 230406336 metrics_app_20190911204823_0000_driver_executor_JVMOffHeapMemory_Count 146132592 metrics_app_20190911204823_0000_driver_executor_OnHeapExecutionMemory_Count 0 metrics_app_20190911204823_0000_driver_executor_OffHeapExecutionMemory_Count 0 metrics_app_20190911204823_0000_driver_executor_OnHeapStorageMemory_Count 0 metrics_app_20190911204823_0000_driver_executor_OffHeapStorageMemory_Count 0 metrics_app_20190911204823_0000_driver_executor_OnHeapUnifiedMemory_Count 0 metrics_app_20190911204823_0000_driver_executor_OffHeapUnifiedMemory_Count 0 metrics_app_20190911204823_0000_driver_executor_DirectPoolMemory_Count 97049 metrics_app_20190911204823_0000_driver_executor_MappedPoolMemory_Count 0 metrics_app_20190911204823_0000_driver_executor_ProcessTreeJVMVMemory_Count 0 metrics_app_20190911204823_0000_driver_executor_ProcessTreeJVMRSSMemory_Count 0 metrics_app_20190911204823_0000_driver_executor_ProcessTreePythonVMemory_Count 0 metrics_app_20190911204823_0000_driver_executor_ProcessTreePythonRSSMemory_Count 0 metrics_app_20190911204823_0000_driver_executor_ProcessTreeOtherVMemory_Count 0 metrics_app_20190911204823_0000_driver_executor_ProcessTreeOtherRSSMemory_Count 0 metrics_app_20190911204823_0000_driver_executor_MinorGCCount_Count 8 metrics_app_20190911204823_0000_driver_executor_MinorGCTime_Count 82 metrics_app_20190911204823_0000_driver_executor_MajorGCCount_Count 3 metrics_app_20190911204823_0000_driver_executor_MajorGCTime_Count 128 metrics_app_20190911204823_0000_0_executor_rddBlocks_Count 0 metrics_app_20190911204823_0000_0_executor_memoryUsed_Count 0 metrics_app_20190911204823_0000_0_executor_diskUsed_Count 0 metrics_app_20190911204823_0000_0_executor_totalCores_Count 16 metrics_app_20190911204823_0000_0_executor_maxTasks_Count 16 metrics_app_20190911204823_0000_0_executor_activeTasks_Count 0 metrics_app_20190911204823_0000_0_executor_failedTasks_Count 0 metrics_app_20190911204823_0000_0_executor_completedTasks_Count 0 metrics_app_20190911204823_0000_0_executor_totalTasks_Count 0 metrics_app_20190911204823_0000_0_executor_totalDuration_Value 0 metrics_app_20190911204823_0000_0_executor_totalGCTime_Value 0 metrics_app_20190911204823_0000_0_executor_totalInputBytes_Count 0 metrics_app_20190911204823_0000_0_executor_totalShuffleRead_Count 0 metrics_app_20190911204823_0000_0_executor_totalShuffleWrite_Count 0 metrics_app_20190911204823_0000_0_executor_maxMemory_Count 384093388 metrics_app_20190911204823_0000_0_executor_usedOnHeapStorageMemory_Count 0 metrics_app_20190911204823_0000_0_executor_usedOffHeapStorageMemory_Count 0 metrics_app_20190911204823_0000_0_executor_totalOnHeapStorageMemory_Count 384093388 metrics_app_20190911204823_0000_0_executor_totalOffHeapStorageMemory_Count 0 ``` Closes apache#25770 from dongjoon-hyun/SPARK-29064. Authored-by: Dongjoon Hyun <dhyun@apple.com> Signed-off-by: DB Tsai <d_tsai@apple.com>

edwinalu added 3 commits May 2, 2018 14:58

modify MimaExcludes.scala to filter changes to SparkListenerExecutorM…

5d6ae1c

…etricsUpdate

Address code review comments, change event logging to stage end.

ad10d28

edwinalu mentioned this pull request May 2, 2018

[SPARK-23429][CORE] Add executor memory metrics to heartbeat and expose in executors REST API #20940

Closed

Ngone51 reviewed May 10, 2018

View reviewed changes

Add configuration parameter spark.eventLog.logExecutorMetricsUpdates.…

10ed328

…enabled to enable/disable executor metrics update logging. Code review comments.

squito reviewed May 23, 2018

View reviewed changes

squito added 3 commits May 23, 2018 14:37

wip on enum based metrics

2d20367

wip ... has both enum and non-enum version

f904f1e

case objects, mostly complete

c502ec4

squito reviewed May 23, 2018

View reviewed changes

Merge pull request #1 from squito/metric_enums

7879e66

Metric enums

edwinalu and others added 6 commits June 13, 2018 16:23

Address comments (move heartbeater from DAGScheduler to SparkContext,…

2662f6f

… move logic for getting metrics to Heartbeater), and modifiy tests for the new ExecutorMetrics format.

modify MimaExcludes.scala to filter changes to SparkListenerExecutorM…

da83f2e

…etricsUpdate

Address code review comments, change event logging to stage end.

f25a44b

Add configuration parameter spark.eventLog.logExecutorMetricsUpdates.…

ca85c82

…enabled to enable/disable executor metrics update logging. Code review comments.

wip on enum based metrics

8b74ba8

felixcheung reviewed Aug 16, 2018

View reviewed changes

disable stage executor metrics logging by default

2897281

Merge branch 'master' into SPARK-23429.2

ee4aa1d

kiszk reviewed Sep 7, 2018

View reviewed changes

fix indentation

571285b

asfgit closed this in 9241e1e Sep 7, 2018

Fokko mentioned this pull request Sep 11, 2018

[SPARK-24601] Update Jackson to 2.9.6 #21596

Closed

LantaoJin mentioned this pull request Sep 18, 2018

[SPARK-25421][SQL] Abstract an output path field in trait DataWritingCommand #22411

Closed

rezasafi mentioned this pull request Oct 10, 2018

[SPARK-24958][CORE] Add memory from procfs to executor metrics. #22612

Closed

wypoon mentioned this pull request Sep 4, 2019

[SPARK-28770][CORE][TESTS]Ignore SparkListenerStageExecutorMetrics in testApplicationReplay test #25659

Closed

dongjoon-hyun mentioned this pull request Sep 12, 2019

[SPARK-29064][CORE] Add PrometheusResource to export Executor metrics #25770

Closed

[SPARK-23429][CORE] Add executor memory metrics to heartbeat and expose in executors REST API #21221

[SPARK-23429][CORE] Add executor memory metrics to heartbeat and expose in executors REST API #21221

Conversation

edwinalu commented May 2, 2018 • edited

What changes were proposed in this pull request?

How was this patch tested?

squito commented May 3, 2018

SparkQA commented May 3, 2018

squito commented May 9, 2018

SparkQA commented May 9, 2018

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

edwinalu May 11, 2018 • edited

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

SparkQA commented May 15, 2018

squito left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

SparkQA commented Jun 3, 2018

felixcheung commented Jun 10, 2018

felixcheung commented Jun 10, 2018

SparkQA commented Jun 10, 2018

edwinalu commented Jun 11, 2018

edwinalu commented Jun 11, 2018

squito commented Jun 11, 2018

felixcheung left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

felixcheung commented Aug 16, 2018

SparkQA commented Aug 16, 2018

SparkQA commented Aug 17, 2018

mccheah commented Aug 20, 2018

mccheah commented Sep 6, 2018

SparkQA commented Sep 7, 2018

Choose a reason for hiding this comment

SparkQA commented Sep 7, 2018

mccheah commented Sep 7, 2018

edwinalu commented Sep 7, 2018

gatorsmile commented Sep 30, 2018

edwinalu commented May 2, 2018 •

edited

edwinalu May 11, 2018 •

edited