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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
