You are viewing a plain text version of this content. The canonical link for it is here.
Posted to reviews@spark.apache.org by "wangyum (via GitHub)" <gi...@apache.org> on 2026/09/22 15:31:47 UTC
[PR] [SPARK-59711][CORE] Stop LiveExecutorStageSummary from retaining a full v1.TaskMetrics tree [spark]
wangyum opened a new pull request, #58968:
URL: https://github.com/apache/spark/pull/58968
### What changes were proposed in this pull request?
`LiveExecutorStageSummary` no longer keeps a `v1.TaskMetrics` object graph (`createMetrics` / `addMetrics`). It stores only the 10 longs that `ExecutorStageSummary` already exposes, and accumulates them with `addTaskMetrics`.
`TaskEnd` and `ExecutorMetricsUpdate` call `addTaskMetrics` instead of allocating a new metrics tree. `doUpdate()` still writes the same `ExecutorStageSummary` fields, so the REST/UI payload is unchanged.
### Why are the changes needed?
The executor-stage UI only needs:
`inputBytes`, `inputRecords`, `outputBytes`, `outputRecords`, `shuffleRead`, `shuffleReadRecords`, `shuffleWrite`, `shuffleWriteRecords`, `memoryBytesSpilled`, `diskBytesSpilled`.
The live object previously held a full `TaskMetrics` + `Input`/`Output`/`ShuffleRead`/`ShufflePushRead`/`ShuffleWrite` tree per `(stage, executor)`. Most of those fields are never shown. `addMetrics` also allocated a new tree on every task update, so even short jobs paid extra GC.
On a long-lived driver (many live stages × many executors) this shows up as tens of millions of `TaskMetrics` instances. This change is a memory-shape fix: same numbers, fewer objects.
### Does this PR introduce _any_ user-facing change?
No.
### How was this patch tested?
- New unit test: `LiveExecutorStageSummary` has no `v1.TaskMetrics` field.
- `AppStatusListenerWithInMemoryStoreSuite` (24 tests), including SPARK-24415 late task metrics and executor-summary cases.
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: Cursor Grok 4.6
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-59711][CORE] Stop LiveExecutorStageSummary from retaining a full v1.TaskMetrics tree [spark]
Posted by "wangyum (via GitHub)" <gi...@apache.org>.
wangyum commented on PR #58968:
URL: https://github.com/apache/spark/pull/58968#issuecomment-5865544134
> Thank you for updating the PR, @wangyum. I reviewed the latest commit ([90dda44](https://github.com/apache/spark/commit/90dda44fe000335eb8ef7f6355f21c369b53b4d9)).
>
> The production change is a behavior-preserving refactoring, and the latest commit addresses my previous comments. Summary of the remaining comments:
>
> 1. **PR description**: The `How was this patch tested?` section is outdated. The structural test now lives in `LiveEntitySuite` instead of `AppStatusListenerSuite`. It checks a field-type allow-list and whether `addTaskMetrics` increments every declared `Long` field. Since the PR description becomes the commit message, could you update it? For example:
> ```
> `AppStatusListenerSuite`:
> - A live stage receives `SparkListenerExecutorMetricsUpdate` and then `SparkListenerTaskEnd`, each with distinct metric values. The test checks all 10 `ExecutorStageSummary` metric fields after each event.
>
> `LiveEntitySuite`:
> - `LiveExecutorStageSummary` declares only `Long`, `Int`, `Boolean`, `String`, or `ExecutorMetrics` fields.
> - `addTaskMetrics` increments every declared `Long` field except `taskTime`.
> ```
> 2. **Test (nit)**: `buildMetrics` leaves the push-based merged bytes at zero. An extra term such as `remoteMergedBytesRead` in `addTaskMetrics` would not be caught.
> 3. **Pre-existing (not introduced by this PR)**: `speculationStageSummary.numActiveTasks` can become negative on `Resubmitted`. This is out of scope and can be handled with the `taskTime` issue in a separate JIRA.
Fix these issues.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-59711][CORE] Stop LiveExecutorStageSummary from retaining a full v1.TaskMetrics tree [spark]
Posted by "dongjoon-hyun (via GitHub)" <gi...@apache.org>.
dongjoon-hyun commented on code in PR #58968:
URL: https://github.com/apache/spark/pull/58968#discussion_r4118704544
##########
core/src/main/scala/org/apache/spark/status/AppStatusListener.scala:
##########
@@ -771,7 +771,7 @@ private[spark] class AppStatusListener(
esummary.failedTasks += failedDelta
Review Comment:
There is one more pre-existing `Resubmitted` issue in the same `onTaskEnd`. `stage.speculationStageSummary.numActiveTasks -= 1` (line 788) is still unconditional. SPARK-41187 changed the other active counters to use `activeDelta`, which is 0 for `Resubmitted`.
Suppose a speculative copy of a shuffle map task succeeds, and its executor is lost while the stage is still running. Then `TaskSetManager.executorLost` posts `Resubmitted` with the original `TaskInfo` (`speculative = true`). As a result, the counter is decremented twice and stays at -1 in the speculation summary of the stage page and the REST API.
This is also out of the scope of this PR. It can be fixed (`-= activeDelta`) together with the `taskTime` issue in a separate JIRA.
##########
core/src/test/scala/org/apache/spark/status/AppStatusListenerSuite.scala:
##########
@@ -1917,6 +1917,71 @@ abstract class AppStatusListenerSuite extends SparkFunSuite with BeforeAndAfter
checkInfoPopulated(listener, logUrlMap, processId)
}
+ test("SPARK-59711: LiveExecutorStageSummary accumulates all exposed task metrics") {
+ // Give every field a distinct, non-zero value derived from `base` so a dropped or
+ // mismatched term in addTaskMetrics is caught.
+ def buildMetrics(base: Long): TaskMetrics = {
+ val metrics = TaskMetrics.empty
+ metrics.inputMetrics.incBytesRead(base + 1)
+ metrics.inputMetrics.incRecordsRead(base + 2)
+ metrics.outputMetrics.setBytesWritten(base + 3)
+ metrics.outputMetrics.setRecordsWritten(base + 4)
+ metrics.shuffleReadMetrics.incRemoteBytesRead(base + 5)
+ metrics.shuffleReadMetrics.incLocalBytesRead(base + 6)
+ metrics.shuffleReadMetrics.incRecordsRead(base + 7)
+ metrics.shuffleWriteMetrics.incBytesWritten(base + 8)
+ metrics.shuffleWriteMetrics.incRecordsWritten(base + 9)
+ metrics.incMemoryBytesSpilled(base + 10)
+ metrics.incDiskBytesSpilled(base + 11)
Review Comment:
nit. `ShuffleBlockFetcherIterator.shuffleRemoteMetricsUpdate` and `shuffleLocalMetricsUpdate` add the same `buf.size` to both the merged bytes and `remoteBytesRead`/`localBytesRead`. So adding merged bytes to `shuffleRead` would double-count them. Currently, both SPARK-59711 tests would still pass in that case, because the merged bytes are 0 here. How about giving them non-zero values too?
```suggestion
metrics.incDiskBytesSpilled(base + 11)
// Merged bytes are already included in remote/local bytes read, so addTaskMetrics
// must not add them to shuffleRead again.
metrics.shuffleReadMetrics.incRemoteMergedBytesRead(base + 12)
metrics.shuffleReadMetrics.incLocalMergedBytesRead(base + 13)
```
##########
core/src/main/scala/org/apache/spark/status/LiveEntity.scala:
##########
@@ -379,34 +379,57 @@ private class LiveExecutorStageSummary(
attemptId: Int,
executorId: String) extends LiveEntity {
- import LiveEntityHelpers._
-
var taskTime = 0L
var succeededTasks = 0
var failedTasks = 0
var killedTasks = 0
var isExcluded = false
- var metrics = createMetrics(default = 0L)
+ // Only the longs that ExecutorStageSummary exposes. Do not hold a v1.TaskMetrics graph
+ // (Input/Output/ShuffleRead/ShuffleWrite/ShufflePushRead) per (stage, executor).
+ var inputBytes = 0L
+ var inputRecords = 0L
+ var outputBytes = 0L
+ var outputRecords = 0L
+ var shuffleRead = 0L
+ var shuffleReadRecords = 0L
+ var shuffleWrite = 0L
+ var shuffleWriteRecords = 0L
+ var memoryBytesSpilled = 0L
+ var diskBytesSpilled = 0L
val peakExecutorMetrics = new ExecutorMetrics()
+ def addTaskMetrics(delta: v1.TaskMetrics): Unit = {
Review Comment:
Thank you. The new `LiveEntitySuite` check enumerates the declared `Long` fields, so a newly added var which `addTaskMetrics` misses would fail it. This addresses my concern.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-59711][CORE] Stop LiveExecutorStageSummary from retaining a full v1.TaskMetrics tree [spark]
Posted by "wangyum (via GitHub)" <gi...@apache.org>.
wangyum commented on code in PR #58968:
URL: https://github.com/apache/spark/pull/58968#discussion_r4111512694
##########
core/src/main/scala/org/apache/spark/status/LiveEntity.scala:
##########
@@ -379,34 +379,57 @@ private class LiveExecutorStageSummary(
attemptId: Int,
executorId: String) extends LiveEntity {
- import LiveEntityHelpers._
-
var taskTime = 0L
var succeededTasks = 0
var failedTasks = 0
var killedTasks = 0
var isExcluded = false
- var metrics = createMetrics(default = 0L)
+ // Only the longs that ExecutorStageSummary exposes. Do not hold a v1.TaskMetrics graph
Review Comment:
ok
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-59711][CORE] Stop LiveExecutorStageSummary from retaining a full v1.TaskMetrics tree [spark]
Posted by "wangyum (via GitHub)" <gi...@apache.org>.
wangyum commented on code in PR #58968:
URL: https://github.com/apache/spark/pull/58968#discussion_r4089159495
##########
core/src/main/scala/org/apache/spark/status/LiveEntity.scala:
##########
@@ -379,34 +379,60 @@ private class LiveExecutorStageSummary(
attemptId: Int,
executorId: String) extends LiveEntity {
- import LiveEntityHelpers._
-
var taskTime = 0L
var succeededTasks = 0
var failedTasks = 0
var killedTasks = 0
var isExcluded = false
- var metrics = createMetrics(default = 0L)
+ // Only the longs that ExecutorStageSummary exposes. Do not hold a v1.TaskMetrics graph
+ // (Input/Output/ShuffleRead/ShuffleWrite/ShufflePushRead) per (stage, executor).
+ var inputBytes = 0L
+ var inputRecords = 0L
+ var outputBytes = 0L
+ var outputRecords = 0L
+ var shuffleRead = 0L
+ var shuffleReadRecords = 0L
+ var shuffleWrite = 0L
+ var shuffleWriteRecords = 0L
+ var memoryBytesSpilled = 0L
+ var diskBytesSpilled = 0L
val peakExecutorMetrics = new ExecutorMetrics()
+ def addTaskMetrics(delta: v1.TaskMetrics): Unit = {
+ if (delta == null) {
Review Comment:
removed it
##########
core/src/test/scala/org/apache/spark/status/AppStatusListenerSuite.scala:
##########
@@ -1917,6 +1917,12 @@ abstract class AppStatusListenerSuite extends SparkFunSuite with BeforeAndAfter
checkInfoPopulated(listener, logUrlMap, processId)
}
+ test("LiveExecutorStageSummary does not hold v1.TaskMetrics") {
Review Comment:
added a test
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-59711][CORE] Stop LiveExecutorStageSummary from retaining a full v1.TaskMetrics tree [spark]
Posted by "wangyum (via GitHub)" <gi...@apache.org>.
wangyum commented on code in PR #58968:
URL: https://github.com/apache/spark/pull/58968#discussion_r4111601351
##########
core/src/test/scala/org/apache/spark/status/AppStatusListenerSuite.scala:
##########
@@ -1917,6 +1917,120 @@ abstract class AppStatusListenerSuite extends SparkFunSuite with BeforeAndAfter
checkInfoPopulated(listener, logUrlMap, processId)
}
+ test("LiveExecutorStageSummary does not hold v1.TaskMetrics") {
Review Comment:
removed it
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-59711][CORE] Stop LiveExecutorStageSummary from retaining a full v1.TaskMetrics tree [spark]
Posted by "dongjoon-hyun (via GitHub)" <gi...@apache.org>.
dongjoon-hyun commented on code in PR #58968:
URL: https://github.com/apache/spark/pull/58968#discussion_r4080318401
##########
core/src/main/scala/org/apache/spark/status/LiveEntity.scala:
##########
@@ -379,34 +379,60 @@ private class LiveExecutorStageSummary(
attemptId: Int,
executorId: String) extends LiveEntity {
- import LiveEntityHelpers._
-
var taskTime = 0L
var succeededTasks = 0
var failedTasks = 0
var killedTasks = 0
var isExcluded = false
- var metrics = createMetrics(default = 0L)
+ // Only the longs that ExecutorStageSummary exposes. Do not hold a v1.TaskMetrics graph
+ // (Input/Output/ShuffleRead/ShuffleWrite/ShufflePushRead) per (stage, executor).
+ var inputBytes = 0L
+ var inputRecords = 0L
+ var outputBytes = 0L
+ var outputRecords = 0L
+ var shuffleRead = 0L
+ var shuffleReadRecords = 0L
+ var shuffleWrite = 0L
+ var shuffleWriteRecords = 0L
+ var memoryBytesSpilled = 0L
+ var diskBytesSpilled = 0L
val peakExecutorMetrics = new ExecutorMetrics()
+ def addTaskMetrics(delta: v1.TaskMetrics): Unit = {
+ if (delta == null) {
Review Comment:
This null check seems unnecessary. `onTaskEnd` already guards with `if (metricsDelta != null)`, and in `onExecutorMetricsUpdate` the delta comes from `TaskMetrics.fromAccumulatorInfos`, which is never null. The previous code didn't handle null here either. Could we remove it? It also lets us avoid `return`, which the Databricks Scala style guide discourages.
##########
core/src/test/scala/org/apache/spark/status/AppStatusListenerSuite.scala:
##########
@@ -1917,6 +1917,12 @@ abstract class AppStatusListenerSuite extends SparkFunSuite with BeforeAndAfter
checkInfoPopulated(listener, logUrlMap, processId)
}
+ test("LiveExecutorStageSummary does not hold v1.TaskMetrics") {
Review Comment:
This reflection-based test only checks the class structure, not the behavior. Currently, `AppStatusListenerSuite` verifies only `memoryBytesSpilled` of `ExecutorStageSummary`, so a wrong field mapping in `addTaskMetrics` (e.g., input/output/shuffle read/shuffle write) would not be caught. Could you add (or replace this with) a test which sends `SparkListenerTaskEnd` and `SparkListenerExecutorMetricsUpdate` with distinct metric values and verifies all 10 metric fields of `ExecutorStageSummary`?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-59711][CORE] Stop LiveExecutorStageSummary from retaining a full v1.TaskMetrics tree [spark]
Posted by "dongjoon-hyun (via GitHub)" <gi...@apache.org>.
dongjoon-hyun commented on code in PR #58968:
URL: https://github.com/apache/spark/pull/58968#discussion_r4108634059
##########
core/src/main/scala/org/apache/spark/status/AppStatusListener.scala:
##########
@@ -771,7 +771,7 @@ private[spark] class AppStatusListener(
esummary.failedTasks += failedDelta
Review Comment:
Not introduced by this PR, but I noticed this while reviewing `onTaskEnd`.
For `Resubmitted`, `TaskSetManager.executorLost` re-posts the task end with the original finished `TaskInfo`, whose `duration` is non-zero. The metrics are skipped for `Resubmitted` (`if (event.reason != Resubmitted)`), but `esummary.taskTime += event.taskInfo.duration` (line 769) and `exec.totalDuration += event.taskInfo.duration` are not. As a result, a successful shuffle map task whose executor is lost has its duration counted twice.
This is out of the scope of this PR. We can handle it with a separate JIRA.
##########
core/src/main/scala/org/apache/spark/status/LiveEntity.scala:
##########
@@ -379,34 +379,57 @@ private class LiveExecutorStageSummary(
attemptId: Int,
executorId: String) extends LiveEntity {
- import LiveEntityHelpers._
-
var taskTime = 0L
var succeededTasks = 0
var failedTasks = 0
var killedTasks = 0
var isExcluded = false
- var metrics = createMetrics(default = 0L)
+ // Only the longs that ExecutorStageSummary exposes. Do not hold a v1.TaskMetrics graph
Review Comment:
The PR description motivates this change by retention and allocation churn, but `LiveTask` still holds a full `v1.TaskMetrics` tree per live task (`createMetrics(default = -1L)`), `updateMetrics` allocates two trees per update (`createMetrics` + `subtractMetrics`), and `doUpdate` allocates another one via `makeNegative`. For typical jobs, live tasks outnumber the live `(stage, executor)` summaries.
Could you mention in the PR description that this PR covers only the executor-summary path, and which workloads benefit from it?
##########
core/src/test/scala/org/apache/spark/status/AppStatusListenerSuite.scala:
##########
@@ -1917,6 +1917,120 @@ abstract class AppStatusListenerSuite extends SparkFunSuite with BeforeAndAfter
checkInfoPopulated(listener, logUrlMap, processId)
}
+ test("LiveExecutorStageSummary does not hold v1.TaskMetrics") {
+ val fieldTypes = classOf[LiveExecutorStageSummary].getDeclaredFields.map(_.getType)
+ assert(!fieldTypes.contains(classOf[v1.TaskMetrics]),
+ s"LiveExecutorStageSummary still holds TaskMetrics: ${fieldTypes.mkString(", ")}")
+ }
+
+ test("LiveExecutorStageSummary accumulates all exposed task metrics") {
+ // Remote and local shuffle bytes stay separate so shuffleRead is their sum, which makes
+ // 11 source values. The summary itself still exposes 10 fields.
+ // scalastyle:off argcount
+ case class Snapshot(
Review Comment:
The same metric set is listed four times in this test: the `Snapshot` fields, the `toTaskMetrics` setters, the `accum(...)` list, and the `assertSummary` assertions. Could we build `TaskMetrics` directly (e.g., a builder with distinct values per field) and assert against its getters (e.g., `shuffleReadMetrics.totalBytesRead`)? It would make the test much shorter.
##########
core/src/test/scala/org/apache/spark/status/AppStatusListenerSuite.scala:
##########
@@ -1917,6 +1917,120 @@ abstract class AppStatusListenerSuite extends SparkFunSuite with BeforeAndAfter
checkInfoPopulated(listener, logUrlMap, processId)
}
+ test("LiveExecutorStageSummary does not hold v1.TaskMetrics") {
+ val fieldTypes = classOf[LiveExecutorStageSummary].getDeclaredFields.map(_.getType)
+ assert(!fieldTypes.contains(classOf[v1.TaskMetrics]),
+ s"LiveExecutorStageSummary still holds TaskMetrics: ${fieldTypes.mkString(", ")}")
+ }
+
+ test("LiveExecutorStageSummary accumulates all exposed task metrics") {
+ // Remote and local shuffle bytes stay separate so shuffleRead is their sum, which makes
+ // 11 source values. The summary itself still exposes 10 fields.
+ // scalastyle:off argcount
+ case class Snapshot(
+ inputBytes: Long,
+ inputRecords: Long,
+ outputBytes: Long,
+ outputRecords: Long,
+ shuffleRemoteBytes: Long,
+ shuffleLocalBytes: Long,
+ shuffleReadRecords: Long,
+ shuffleWrite: Long,
+ shuffleWriteRecords: Long,
+ memoryBytesSpilled: Long,
+ diskBytesSpilled: Long) {
+
+ def shuffleRead: Long = shuffleRemoteBytes + shuffleLocalBytes
+ }
+ // scalastyle:on argcount
+
+ def toTaskMetrics(snapshot: Snapshot): TaskMetrics = {
+ val metrics = TaskMetrics.empty
+ metrics.inputMetrics.incBytesRead(snapshot.inputBytes)
+ metrics.inputMetrics.incRecordsRead(snapshot.inputRecords)
+ metrics.outputMetrics.setBytesWritten(snapshot.outputBytes)
+ metrics.outputMetrics.setRecordsWritten(snapshot.outputRecords)
+ metrics.shuffleReadMetrics.incRemoteBytesRead(snapshot.shuffleRemoteBytes)
+ metrics.shuffleReadMetrics.incLocalBytesRead(snapshot.shuffleLocalBytes)
+ metrics.shuffleReadMetrics.incRecordsRead(snapshot.shuffleReadRecords)
+ metrics.shuffleWriteMetrics.incBytesWritten(snapshot.shuffleWrite)
+ metrics.shuffleWriteMetrics.incRecordsWritten(snapshot.shuffleWriteRecords)
+ metrics.incMemoryBytesSpilled(snapshot.memoryBytesSpilled)
+ metrics.incDiskBytesSpilled(snapshot.diskBytesSpilled)
+ metrics
+ }
+
+ def accum(name: String, value: Long): AccumulableInfo = {
+ AccumulableInfo(1L, Some(name), Some(value), None, true, false, None)
+ }
+
+ def assertSummary(stage: StageInfo, snapshot: Snapshot): Unit = {
+ val execs = KVUtils.viewToSeq(store.view(classOf[ExecutorStageSummaryWrapper])
Review Comment:
`AppStatusStore.executorSummary(stageId, attemptId)` already provides this lookup. Using `new AppStatusStore(store).executorSummary(...)` would verify the actual UI/REST read path and allow `execs(task.executorId)` instead of `execs.head`.
##########
core/src/main/scala/org/apache/spark/status/LiveEntity.scala:
##########
@@ -379,34 +379,57 @@ private class LiveExecutorStageSummary(
attemptId: Int,
executorId: String) extends LiveEntity {
- import LiveEntityHelpers._
-
var taskTime = 0L
var succeededTasks = 0
var failedTasks = 0
var killedTasks = 0
var isExcluded = false
- var metrics = createMetrics(default = 0L)
+ // Only the longs that ExecutorStageSummary exposes. Do not hold a v1.TaskMetrics graph
+ // (Input/Output/ShuffleRead/ShuffleWrite/ShufflePushRead) per (stage, executor).
+ var inputBytes = 0L
+ var inputRecords = 0L
+ var outputBytes = 0L
+ var outputRecords = 0L
+ var shuffleRead = 0L
+ var shuffleReadRecords = 0L
+ var shuffleWrite = 0L
+ var shuffleWriteRecords = 0L
+ var memoryBytesSpilled = 0L
+ var diskBytesSpilled = 0L
val peakExecutorMetrics = new ExecutorMetrics()
+ def addTaskMetrics(delta: v1.TaskMetrics): Unit = {
Review Comment:
This introduces a second hand-maintained field mapping. Previously, a new `ExecutorStageSummary` metric needed only a `metrics.xxx` read in `doUpdate()` because `addMetrics` accumulated every field. Now it needs 3 edits (`var`, `addTaskMetrics`, `doUpdate`), and only `doUpdate` is checked by the compiler. A missed `+=` silently publishes 0, and the new test would not catch it because it lists the current 10 fields only.
It may be worth a short comment here noting that `addTaskMetrics` and `doUpdate` must be kept in sync.
##########
core/src/test/scala/org/apache/spark/status/AppStatusListenerSuite.scala:
##########
@@ -1917,6 +1917,120 @@ abstract class AppStatusListenerSuite extends SparkFunSuite with BeforeAndAfter
checkInfoPopulated(listener, logUrlMap, processId)
}
+ test("LiveExecutorStageSummary does not hold v1.TaskMetrics") {
Review Comment:
This test uses neither `store` nor `listener`, but it lives in the abstract `AppStatusListenerSuite`, so it runs once per KVStore subclass (InMemory, LevelDB, RocksDB, Protobuf) and creates a temp dir and a KVStore each time. If we keep it, `LiveEntitySuite` (a plain `SparkFunSuite` for `LiveEntity` internals) seems to be a better place.
##########
core/src/test/scala/org/apache/spark/status/AppStatusListenerSuite.scala:
##########
@@ -1917,6 +1917,120 @@ abstract class AppStatusListenerSuite extends SparkFunSuite with BeforeAndAfter
checkInfoPopulated(listener, logUrlMap, processId)
}
+ test("LiveExecutorStageSummary does not hold v1.TaskMetrics") {
+ val fieldTypes = classOf[LiveExecutorStageSummary].getDeclaredFields.map(_.getType)
+ assert(!fieldTypes.contains(classOf[v1.TaskMetrics]),
+ s"LiveExecutorStageSummary still holds TaskMetrics: ${fieldTypes.mkString(", ")}")
+ }
+
+ test("LiveExecutorStageSummary accumulates all exposed task metrics") {
Review Comment:
nit. Please add the JIRA ID prefix to the new test names, e.g., `SPARK-59711: LiveExecutorStageSummary accumulates all exposed task metrics`, like the neighboring `SPARK-41187: ...` test.
##########
core/src/test/scala/org/apache/spark/status/AppStatusListenerSuite.scala:
##########
@@ -1917,6 +1917,120 @@ abstract class AppStatusListenerSuite extends SparkFunSuite with BeforeAndAfter
checkInfoPopulated(listener, logUrlMap, processId)
}
+ test("LiveExecutorStageSummary does not hold v1.TaskMetrics") {
+ val fieldTypes = classOf[LiveExecutorStageSummary].getDeclaredFields.map(_.getType)
Review Comment:
This guard only rejects a field whose erased type is exactly `v1.TaskMetrics`. A field such as `v1.ShuffleReadMetrics`, `Option[v1.TaskMetrics]` (erased to `scala.Option`), or `AnyRef` would bring back the per-`(stage, executor)` graph that the comment in `LiveEntity.scala` forbids, and this test would still pass.
Since the behavioral test below covers the behavior, could we remove this test? Otherwise, an allow-list check would be more robust, e.g., every declared field type is primitive, `String`, or `ExecutorMetrics`.
##########
core/src/test/scala/org/apache/spark/status/AppStatusListenerSuite.scala:
##########
@@ -1917,6 +1917,120 @@ abstract class AppStatusListenerSuite extends SparkFunSuite with BeforeAndAfter
checkInfoPopulated(listener, logUrlMap, processId)
}
+ test("LiveExecutorStageSummary does not hold v1.TaskMetrics") {
+ val fieldTypes = classOf[LiveExecutorStageSummary].getDeclaredFields.map(_.getType)
+ assert(!fieldTypes.contains(classOf[v1.TaskMetrics]),
+ s"LiveExecutorStageSummary still holds TaskMetrics: ${fieldTypes.mkString(", ")}")
+ }
+
+ test("LiveExecutorStageSummary accumulates all exposed task metrics") {
+ // Remote and local shuffle bytes stay separate so shuffleRead is their sum, which makes
+ // 11 source values. The summary itself still exposes 10 fields.
+ // scalastyle:off argcount
Review Comment:
This `scalastyle:off/on argcount` pair has no effect. `argcount` is `ParameterNumberChecker`, which checks only `def` parameter lists, not class constructors (e.g., `v1.StageData` has more than 60 constructor parameters without suppression). Could you remove these lines and the comment above?
##########
core/src/test/scala/org/apache/spark/status/AppStatusListenerSuite.scala:
##########
@@ -1917,6 +1917,120 @@ abstract class AppStatusListenerSuite extends SparkFunSuite with BeforeAndAfter
checkInfoPopulated(listener, logUrlMap, processId)
}
+ test("LiveExecutorStageSummary does not hold v1.TaskMetrics") {
+ val fieldTypes = classOf[LiveExecutorStageSummary].getDeclaredFields.map(_.getType)
+ assert(!fieldTypes.contains(classOf[v1.TaskMetrics]),
+ s"LiveExecutorStageSummary still holds TaskMetrics: ${fieldTypes.mkString(", ")}")
+ }
+
+ test("LiveExecutorStageSummary accumulates all exposed task metrics") {
+ // Remote and local shuffle bytes stay separate so shuffleRead is their sum, which makes
+ // 11 source values. The summary itself still exposes 10 fields.
+ // scalastyle:off argcount
+ case class Snapshot(
+ inputBytes: Long,
+ inputRecords: Long,
+ outputBytes: Long,
+ outputRecords: Long,
+ shuffleRemoteBytes: Long,
+ shuffleLocalBytes: Long,
+ shuffleReadRecords: Long,
+ shuffleWrite: Long,
+ shuffleWriteRecords: Long,
+ memoryBytesSpilled: Long,
+ diskBytesSpilled: Long) {
+
+ def shuffleRead: Long = shuffleRemoteBytes + shuffleLocalBytes
+ }
+ // scalastyle:on argcount
+
+ def toTaskMetrics(snapshot: Snapshot): TaskMetrics = {
+ val metrics = TaskMetrics.empty
+ metrics.inputMetrics.incBytesRead(snapshot.inputBytes)
+ metrics.inputMetrics.incRecordsRead(snapshot.inputRecords)
+ metrics.outputMetrics.setBytesWritten(snapshot.outputBytes)
+ metrics.outputMetrics.setRecordsWritten(snapshot.outputRecords)
+ metrics.shuffleReadMetrics.incRemoteBytesRead(snapshot.shuffleRemoteBytes)
+ metrics.shuffleReadMetrics.incLocalBytesRead(snapshot.shuffleLocalBytes)
+ metrics.shuffleReadMetrics.incRecordsRead(snapshot.shuffleReadRecords)
+ metrics.shuffleWriteMetrics.incBytesWritten(snapshot.shuffleWrite)
+ metrics.shuffleWriteMetrics.incRecordsWritten(snapshot.shuffleWriteRecords)
+ metrics.incMemoryBytesSpilled(snapshot.memoryBytesSpilled)
+ metrics.incDiskBytesSpilled(snapshot.diskBytesSpilled)
+ metrics
+ }
+
+ def accum(name: String, value: Long): AccumulableInfo = {
+ AccumulableInfo(1L, Some(name), Some(value), None, true, false, None)
+ }
+
+ def assertSummary(stage: StageInfo, snapshot: Snapshot): Unit = {
+ val execs = KVUtils.viewToSeq(store.view(classOf[ExecutorStageSummaryWrapper])
+ .index("stage").first(key(stage)).last(key(stage)))
+ assert(execs.size === 1)
+ val info = execs.head.info
+ assert(info.inputBytes === snapshot.inputBytes)
+ assert(info.inputRecords === snapshot.inputRecords)
+ assert(info.outputBytes === snapshot.outputBytes)
+ assert(info.outputRecords === snapshot.outputRecords)
+ assert(info.shuffleRead === snapshot.shuffleRead)
+ assert(info.shuffleReadRecords === snapshot.shuffleReadRecords)
+ assert(info.shuffleWrite === snapshot.shuffleWrite)
+ assert(info.shuffleWriteRecords === snapshot.shuffleWriteRecords)
+ assert(info.memoryBytesSpilled === snapshot.memoryBytesSpilled)
+ assert(info.diskBytesSpilled === snapshot.diskBytesSpilled)
+ }
+
+ val listener = new AppStatusListener(store, conf, true)
+ listener.onExecutorAdded(createExecutorAddedEvent(1))
+ val stage = new StageInfo(1, 0, "stage", 1, Nil, Nil, "details",
+ resourceProfileId = ResourceProfile.DEFAULT_RESOURCE_PROFILE_ID)
+ listener.onJobStart(SparkListenerJobStart(1, time, Seq(stage), null))
+ time += 1
+ stage.submissionTime = Some(time)
+ listener.onStageSubmitted(SparkListenerStageSubmitted(stage, new Properties()))
+
+ val task = createTasks(1, Array("1")).head
+ listener.onTaskStart(SparkListenerTaskStart(stage.stageId, stage.attemptNumber(), task))
+
+ // Heartbeat values are cumulative. Each summary field is distinct, and both shuffle
+ // byte components are non-zero so a dropped remote or local term fails shuffleRead.
+ val heartbeat = Snapshot(
+ inputBytes = 101, inputRecords = 102, outputBytes = 103, outputRecords = 104,
+ shuffleRemoteBytes = 11, shuffleLocalBytes = 19, shuffleReadRecords = 105,
+ shuffleWrite = 106, shuffleWriteRecords = 107,
+ memoryBytesSpilled = 108, diskBytesSpilled = 109)
+ listener.onExecutorMetricsUpdate(SparkListenerExecutorMetricsUpdate(
+ task.executorId,
+ Seq((task.taskId, stage.stageId, stage.attemptNumber(), Seq(
Review Comment:
This hand-written list duplicates the metric-name mapping in `toTaskMetrics`. We can derive it from the same object, like `ListenerEventsTestHelper.createExecutorMetricsUpdateEvent` does:
```scala
Seq((task.taskId, stage.stageId, stage.attemptNumber(),
toTaskMetrics(heartbeat).accumulators().map(AccumulatorSuite.makeInfo)))
```
`TaskMetrics.fromAccumulatorInfos` matches by name only, so the result is the same, and the `accum` helper can be removed.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-59711][CORE] Stop LiveExecutorStageSummary from retaining a full v1.TaskMetrics tree [spark]
Posted by "wangyum (via GitHub)" <gi...@apache.org>.
wangyum commented on code in PR #58968:
URL: https://github.com/apache/spark/pull/58968#discussion_r4111598825
##########
core/src/main/scala/org/apache/spark/status/LiveEntity.scala:
##########
@@ -379,34 +379,57 @@ private class LiveExecutorStageSummary(
attemptId: Int,
executorId: String) extends LiveEntity {
- import LiveEntityHelpers._
-
var taskTime = 0L
var succeededTasks = 0
var failedTasks = 0
var killedTasks = 0
var isExcluded = false
- var metrics = createMetrics(default = 0L)
+ // Only the longs that ExecutorStageSummary exposes. Do not hold a v1.TaskMetrics graph
+ // (Input/Output/ShuffleRead/ShuffleWrite/ShufflePushRead) per (stage, executor).
+ var inputBytes = 0L
+ var inputRecords = 0L
+ var outputBytes = 0L
+ var outputRecords = 0L
+ var shuffleRead = 0L
+ var shuffleReadRecords = 0L
+ var shuffleWrite = 0L
+ var shuffleWriteRecords = 0L
+ var memoryBytesSpilled = 0L
+ var diskBytesSpilled = 0L
val peakExecutorMetrics = new ExecutorMetrics()
+ def addTaskMetrics(delta: v1.TaskMetrics): Unit = {
Review Comment:
add a test to make sure it will not silently publishes 0
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org