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: [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]
