yew1eb commented on code in PR #3872:
URL: https://github.com/apache/celeborn/pull/3872#discussion_r4333972702
##########
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:
Fixed in cd1636e79 — the two overrides are merged into a single
`onOtherEvent` that handles both `CelebornBuildInfoEvent` and
`CelebornMapperMetricsEvent`; the existing tests cover both paths.
##########
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:
Addressed in cd1636e79. `LifecycleManager` now tracks every reported
`(shuffleId, mapId, attemptId)` — including rejected attempts — and invokes the
callback exactly once per finished writer attempt; retried MapperEnd RPCs of
rejected attempts are suppressed, and the bookkeeping is removed with the
shuffle in `removeExpiredShuffle`. The regression test now replays your probe
(attempts 0,0,1,1 with metrics 100,100,250,250) and asserts exactly [100, 250].
##########
client-spark/spark-3-ui/src/main/scala/org/apache/spark/shuffle/celeborn/CelebornListener.scala:
##########
@@ -71,6 +73,17 @@ private[celeborn] class CelebornListener(
mayUpdate()
}
+ override def onOtherEvent(event: SparkListenerEvent): Unit = {
+ event match {
+ case e: CelebornMapperMetricsEvent =>
+ if (pluginEnabled) {
+ totalUncompressedWriteBytes.addAndGet(e.getUncompressedBytesWritten)
Review Comment:
Addressed in cd1636e79 by deriving both counters from the same attempt set:
writers now also report post-compression bytes via a new additive
`PbMapperMetrics.compressedBytesWritten` field, and the listener accumulates
both ratio sides from MapperEnd reports. In your repro the page would now show
60/282 — the same ratio as the successful-only 40/188 — instead of the skewed
60/188. Added a listener regression for the fail-after-write scenario, plus
JSON round-trip coverage for event logs written before the new field existed.
--
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]
