zaynt4606 commented on code in PR #3872:
URL: https://github.com/apache/celeborn/pull/3872#discussion_r4332759899
##########
client-spark/spark-3-ui/src/main/scala/org/apache/spark/shuffle/celeborn/CelebornListener.scala:
##########
@@ -62,16 +64,33 @@ private[celeborn] class CelebornListener(
return
}
Option(taskEnd.taskMetrics).foreach { metrics =>
- totalWriteBytes.addAndGet(metrics.shuffleWriteMetrics.bytesWritten)
- // writeTime is in nanoseconds; normalize to ms.
- totalWriteTimeMs.addAndGet(metrics.shuffleWriteMetrics.writeTime /
1000000L)
+ taskEnd.reason match {
+ case Success =>
+ // Write-side metrics only for successful attempts: failed ones
never report
+ // MapperEnd, so counting them would skew the compression ratio.
+ totalWriteBytes.addAndGet(metrics.shuffleWriteMetrics.bytesWritten)
+ // writeTime is in nanoseconds; normalize to ms.
+ totalWriteTimeMs.addAndGet(metrics.shuffleWriteMetrics.writeTime /
1000000L)
+ case _ =>
+ }
totalReadBytes.addAndGet(metrics.shuffleReadMetrics.totalBytesRead)
totalFetchWaitTimeMs.addAndGet(metrics.shuffleReadMetrics.fetchWaitTime)
totalTaskDurationMs.addAndGet(taskEnd.taskInfo.duration)
}
mayUpdate()
}
+ override def onOtherEvent(event: SparkListenerEvent): Unit = {
Review Comment:
[P1] Combine the two onOtherEvent overrides
After merging main, this new override coexists with the override at line 127
that handles CelebornBuildInfoEvent from #3870. Compiling the current listener
at 96f8cc84 fails with `method onOtherEvent is defined twice` (lines 83 and
127). Please put the CelebornMapperMetricsEvent and CelebornBuildInfoEvent
cases into a single onOtherEvent override and retain coverage for both event
types.
##########
client/src/main/scala/org/apache/celeborn/client/LifecycleManager.scala:
##########
@@ -980,6 +983,13 @@ class LifecycleManager(val appUniqueId: String, val conf:
CelebornConf) extends
numPartitions = numPartitions,
crc32PerPartition = crc32PerPartition,
bytesWrittenPerPartition = bytesWrittenPerPartition)
+ // Report every finished writer attempt, including ones the commit handler
rejects
+ // (duplicate speculative attempts); a rejected MapperEnd with the same
attemptId is
+ // just a retried RPC.
+ if (mapperAttemptFinishedSuccess ||
+ attemptId != commitManager.getMapperAttempts(shuffleId)(mapId)) {
+ mapperEndCallback.foreach(_.onMapperEnd(shuffleId, mapId, attemptId,
metrics))
Review Comment:
[P2] Deduplicate retried rejected attempts as well
getMapperAttempts(shuffleId)(mapId) retains the accepted attempt ID. If
attempt 0 is accepted and speculative attempt 1 is rejected, every retry of
attempt 1 still satisfies `1 != 0`, so its metrics are emitted repeatedly.
A real-RPC probe on ea7a4fa3 sent MapperEnd for attempts 0, 0, 1, 1 with
metrics 100, 100, 250, 250. The callbacks were 100, 250, 250 (600 total),
rather than 100, 250 (350 total). This entire handleMapperEnd method is
unchanged in 96f8cc84. The existing test retries the accepted attempt, so it
misses this case.
Could we deduplicate every reported (shuffleId, mapId, attemptId), including
rejected attempts, clean up the bookkeeping with the shuffle, and add a
regression that retries the rejected attempt?
--
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]
