[SPARK-26329][CORE] Faster polling of executor memory metrics. #23767

wypoon · 2019-02-12T16:51:42Z

What changes were proposed in this pull request?

Prior to this change, in an executor, on each heartbeat, memory metrics are polled and sent in the heartbeat. The heartbeat interval is 10s by default. With this change, in an executor, memory metrics can optionally be polled in a separate poller at a shorter interval.

For each executor, we use a map of (stageId, stageAttemptId) to (count of running tasks, executor metric peaks) to track what stages are active as well as the per-stage memory metric peaks. When polling the executor memory metrics, we attribute the memory to the active stage(s), and update the peaks. In a heartbeat, we send the per-stage peaks (for stages active at that time), and then reset the peaks. The semantics would be that the per-stage peaks sent in each heartbeat are the peaks since the last heartbeat.

We also keep a map of taskId to memory metric peaks. This tracks the metric peaks during the lifetime of the task. The polling thread updates this as well. At end of a task, we send the peak metric values in the task result. In case of task failure, we send the peak metric values in the TaskFailedReason.

We continue to do the stage-level aggregation in the EventLoggingListener.

For the driver, we still only poll on heartbeats. What the driver sends will be the current values of the metrics in the driver at the time of the heartbeat. This is semantically the same as before.

How was this patch tested?

Unit tests. Manually tested applications on an actual system and checked the event logs; the metrics appear in the SparkListenerTaskEnd and SparkListenerStageExecutorMetrics events.

SparkQA · 2019-02-12T17:01:27Z

Test build #102260 has finished for PR 23767 at commit ad8b5e5.

This patch fails to build.
This patch merges cleanly.
This patch adds no public classes.

SparkQA · 2019-02-13T00:14:12Z

Test build #102266 has finished for PR 23767 at commit faa3b09.

This patch fails MiMa tests.
This patch merges cleanly.
This patch adds no public classes.

SparkQA · 2019-02-13T05:44:49Z

Test build #102271 has finished for PR 23767 at commit 0dd8600.

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

SparkQA · 2019-02-13T08:05:02Z

Test build #102280 has finished for PR 23767 at commit 3db5924.

This patch fails due to an unknown error code, -9.
This patch merges cleanly.
This patch adds no public classes.

wypoon · 2019-02-13T16:29:13Z

Retest this please.

SparkQA · 2019-02-13T20:53:40Z

Test build #102301 has finished for PR 23767 at commit 3db5924.

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

wypoon · 2019-02-13T22:24:02Z

Retest this please.

SparkQA · 2019-02-14T02:17:42Z

Test build #102311 has finished for PR 23767 at commit 3db5924.

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

HyukjinKwon · 2019-02-14T03:02:38Z

retest this please

HyukjinKwon · 2019-02-14T03:07:19Z

PySpark test failures look irrelevant.

SparkQA · 2019-02-14T08:05:02Z

Test build #102329 has finished for PR 23767 at commit 3db5924.

This patch fails due to an unknown error code, -9.
This patch merges cleanly.
This patch adds no public classes.

wypoon · 2019-02-14T17:28:56Z

The PySpark tests and PySpark packaging tests were successful this time, but the build errored in running SparkR tests.
In any case, Java and Scala unit tests are green.
@edwinalu can you please take a look at my change?

HyukjinKwon · 2019-02-14T17:34:29Z

retest this please

SparkQA · 2019-02-14T22:17:22Z

Test build #102358 has finished for PR 23767 at commit 3db5924.

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

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

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

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

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

squito · 2019-02-15T22:47:55Z

@mccheah you might be interested in this too

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

wypoon · 2019-02-17T05:28:00Z

@edwinalu thank you for your feedback. I'll be on vacation next week. When I come back, I'll look into your comments further.

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

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

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

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

squito · 2019-02-20T21:04:58Z

btw we should probably also update the docs on ExecutorMetricType to explain threading concerns, as its less clear now. IIUC, implementations do not need to be thread-safe currently -- they will only be called by one thread at a time.

SparkQA · 2019-03-06T08:05:01Z

Test build #103079 has finished for PR 23767 at commit 4a947e9.

This patch fails due to an unknown error code, -9.
This patch merges cleanly.
This patch adds no public classes.

wypoon · 2019-03-06T17:48:29Z

Retest this please.

SparkQA · 2019-03-06T22:59:53Z

Test build #103104 has finished for PR 23767 at commit 4a947e9.

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

wypoon · 2019-03-09T00:20:08Z

btw we should probably also update the docs on ExecutorMetricType to explain threading concerns, as its less clear now. IIUC, implementations do not need to be thread-safe currently -- they will only be called by one thread at a time.

For each executor (or the driver), the executor metrics are polled by a single thread, so I don't think there are any concerns there. It's only in the tracking of per-stage and per-task metric peaks that there are concurrency concerns around the data structures used to track them.

SparkQA · 2019-03-09T04:09:53Z

Test build #103242 has finished for PR 23767 at commit e13edaa.

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

wypoon · 2019-03-10T20:08:01Z

org.apache.spark.streaming.kafka010.DirectKafkaStreamSuite."offset recovery from kafka" failed.
Appears to be flakiness. I ran the suite locally and it passes.

…sult and TaskFailedReason.

…ggregation. Add an internal check within the test showing how the expected aggregated metrics are calculated. This documents how the expected values are arrived at.

…tor metrics aggregation.

In ExecutorMetricsPoller, combine onTaskLaunch and onTaskStart, and combine onTaskCompletion and onTaskCleanup. Simplify getExecutorUpdates, as there should never be a stage in stageTCMP with task count == 0. Some code format cleanup and minor improvements.

Add some explanatory comments around existing helper methods and classes and how they are currently invoked.

…utable.Map. In Heartbeat, since the Map we actually construct and send as executorUpdates is a mutable one, make the signature use a scala.collection,mutable.Map instead of scala.collection.Map.

There are three existing cases in HistoryServerSuite related to executor metrics. Replace them with a single case that shows all the executor metrics we currently collect; they include proc fs metrics and garbage collection netrics. Uploaded new event logs. These show executor metrics at task end.

…gle pass. Also, make driver stage key a constant; and use SAM syntax.

Fix some bugs in ExecutorSuite as well.

SparkQA · 2019-07-30T00:44:28Z

Test build #108351 has finished for PR 23767 at commit 7331b27.

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

squito

some super teeny comments, otherwise lgtm

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

core/src/test/scala/org/apache/spark/executor/ExecutorSuite.scala

… Rashid.

SparkQA · 2019-07-31T20:17:33Z

Test build #108484 has finished for PR 23767 at commit 7556d6a.

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

squito · 2019-08-01T14:10:45Z

merged to master, thanks @wypoon !

zsxwing · 2019-08-19T14:30:52Z

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

@@ -574,7 +600,8 @@ private[spark] class Executor(
          logInfo(s"Executor interrupted and killed $taskName (TID $taskId), reason: $killReason")

          val (accums, accUpdates) = collectAccumulatorsAndResetStatusOnFailure(taskStartTimeNs)
-          val serializedTK = ser.serialize(TaskKilled(killReason, accUpdates, accums))
+          val metricPeaks = WrappedArray.make(metricsPoller.getTaskMetricPeaks(taskId))
+          val serializedTK = ser.serialize(TaskKilled(killReason, accUpdates, accums, metricPeaks))
          execBackend.statusUpdate(taskId, TaskState.KILLED, serializedTK)

        case t: Throwable if hasFetchFailure && !Utils.isFatalError(t) =>


@wypoon Is there any special reason that metricPeaks is not send in this case and CommitDeniedException?

There is no special reason. I saw that in some cases, accumulators are collected and sent in the TaskFailedReason, and in those cases I send the task metric peaks as well; in the other cases, accumulators are not collected so I don't send the task metric peaks either.

…ted after apache#23767

…tion to the metrics system ## What changes were proposed in this pull request? This PR proposes to add instrumentation of memory usage via the Spark Dropwizard/Codahale metrics system. Memory usage metrics are available via the Executor metrics, recently implemented as detailed in https://issues.apache.org/jira/browse/SPARK-23206. Additional notes: This takes advantage of the metrics poller introduced in #23767. ## Why are the changes needed? Executor metrics bring have many useful insights on memory usage, in particular on the usage of storage memory and executor memory. This is useful for troubleshooting. Having the information in the metrics systems allows to add those metrics to Spark performance dashboards and study memory usage as a function of time, as in the example graph https://issues.apache.org/jira/secure/attachment/12962810/Example_dashboard_Spark_Memory_Metrics.PNG ## Does this PR introduce any user-facing change? Adds `ExecutorMetrics` source to publish executor metrics via the Dropwizard metrics system. Details of the available metrics in docs/monitoring.md Adds configuration parameter `spark.metrics.executormetrics.source.enabled` ## How was this patch tested? Tested on YARN cluster and with an existing setup for a Spark dashboard based on InfluxDB and Grafana. Closes #24132 from LucaCanali/memoryMetricsSource. Authored-by: Luca Canali <luca.canali@cern.ch> Signed-off-by: Imran Rashid <irashid@cloudera.com>

wypoon changed the title ~~[SPARK-26329][CORE] Faster polling of executor memory metrics.~~ [SPARK-26329][CORE][WIP] Faster polling of executor memory metrics. Feb 13, 2019

edwinalu reviewed Feb 15, 2019

View reviewed changes

edwinalu reviewed Feb 16, 2019

View reviewed changes

core/src/main/scala/org/apache/spark/SparkContext.scala Show resolved Hide resolved

squito reviewed Feb 20, 2019

View reviewed changes

squito mentioned this pull request Feb 20, 2019

[SPARK-25865][CORE] Add GC information to ExecutorMetrics #22874

Closed

wypoon force-pushed the wypoon_SPARK-26329 branch from 3db5924 to 4a947e9 Compare March 6, 2019 05:01

wypoon force-pushed the wypoon_SPARK-26329 branch from 4a947e9 to e13edaa Compare March 9, 2019 00:05

wypoon added 14 commits July 29, 2019 13:53

[SPARK-26329][CORE] Unit tests for sending executor metrics in TaskRe…

0cbfc04

…sult and TaskFailedReason.

[SPARK-26329][CORE] Add driver updates to test for executor metrics a…

8cb30a8

…ggregation. Add an internal check within the test showing how the expected aggregated metrics are calculated. This documents how the expected values are arrived at.

[SPARK-26329][CORE] Add SparkListenerTaskEnd events to test for execu…

077abb0

…tor metrics aggregation.

[SPARK-26329][CORE] Fix ExecutorSuite failures.

3ed583a

Add some explanatory comments around existing helper methods and classes and how they are currently invoked.

[SPARK-26329][CORE] Delete a comment on irashid's suggestion.

0a4828a

[SPARK-26329][CORE] Change executorUpdates to be a scala.collection.m…

9530b75

…utable.Map. In Heartbeat, since the Map we actually construct and send as executorUpdates is a mutable one, make the signature use a scala.collection,mutable.Map instead of scala.collection.Map.

[SPARK-26329][CORE] Get executor updates and reset the peaks in a sin…

e062e60

…gle pass. Also, make driver stage key a constant; and use SAM syntax.

[SPARK-26329][CORE] Test fixes after rebase on master.

20b4b7e

[SPARK-26329][CORE] Adopt some suggestions from attilapiros.

b898ad2

[SPARK-26329][CORE] Address feedback from Imran Rashid.

fbb55bf

Fix some bugs in ExecutorSuite as well.

[SPARK-26329][CORE] Make TCMP case class private.

99addf1

[SPARK-26329][CORE] Fix a test post-rebase.

7331b27

wypoon force-pushed the wypoon_SPARK-26329 branch from a21ac84 to 7331b27 Compare July 29, 2019 22:29

squito approved these changes Jul 31, 2019

View reviewed changes

core/src/main/scala/org/apache/spark/executor/ExecutorMetricsPoller.scala Outdated Show resolved Hide resolved

core/src/test/scala/org/apache/spark/executor/ExecutorSuite.scala Outdated Show resolved Hide resolved

[SPARK-26329][CORE] Update a doc comment based on feedback from Imran…

7556d6a

… Rashid.

asfgit closed this in 80ab19b Aug 1, 2019

zsxwing reviewed Aug 19, 2019

View reviewed changes

squito mentioned this pull request Aug 26, 2019

Updates and Fixes to Make spark-memory work on opensource spark squito/spark-memory#4

Open

wypoon mentioned this pull request Sep 4, 2019

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

Closed

LucaCanali added a commit to LucaCanali/spark that referenced this pull request Nov 1, 2019

Executor Metrics integration with the Spark metrics system re-impleme…

756e849

…ted after apache#23767

LucaCanali added a commit to LucaCanali/spark that referenced this pull request Nov 11, 2019

Executor Metrics integration with the Spark metrics system re-impleme…

005f95e

…ted after apache#23767

[SPARK-26329][CORE] Faster polling of executor memory metrics. #23767

[SPARK-26329][CORE] Faster polling of executor memory metrics. #23767

Uh oh!

Conversation

wypoon commented Feb 12, 2019 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

What changes were proposed in this pull request?

How was this patch tested?

Uh oh!

SparkQA commented Feb 12, 2019

Uh oh!

SparkQA commented Feb 13, 2019

Uh oh!

SparkQA commented Feb 13, 2019

Uh oh!

SparkQA commented Feb 13, 2019

Uh oh!

wypoon commented Feb 13, 2019

Uh oh!

SparkQA commented Feb 13, 2019

Uh oh!

wypoon commented Feb 13, 2019

Uh oh!

SparkQA commented Feb 14, 2019

Uh oh!

HyukjinKwon commented Feb 14, 2019

Uh oh!

HyukjinKwon commented Feb 14, 2019

Uh oh!

SparkQA commented Feb 14, 2019

Uh oh!

wypoon commented Feb 14, 2019

Uh oh!

HyukjinKwon commented Feb 14, 2019

Uh oh!

SparkQA commented Feb 14, 2019

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

squito commented Feb 15, 2019

Uh oh!

Uh oh!

wypoon commented Feb 17, 2019

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

squito commented Feb 20, 2019

Uh oh!

SparkQA commented Mar 6, 2019

Uh oh!

wypoon commented Mar 6, 2019

Uh oh!

SparkQA commented Mar 6, 2019

Uh oh!

wypoon commented Mar 9, 2019

Uh oh!

SparkQA commented Mar 9, 2019

Uh oh!

wypoon commented Mar 10, 2019 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

SparkQA commented Jul 30, 2019

Uh oh!

squito left a comment

Choose a reason for hiding this comment

wypoon commented Feb 12, 2019 •

edited

Loading

wypoon commented Mar 10, 2019 •

edited

Loading

wypoon Aug 19, 2019 •

edited

Loading