zaynt4606 commented on code in PR #3872:
URL: https://github.com/apache/celeborn/pull/3872#discussion_r4266251107
##########
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:
Could the numerator and denominator use the same set of mapper attempts?
`onTaskEnd` adds `shuffleWriteMetrics.bytesWritten` for every task end with
metrics, while `LifecycleManager.handleMapperEnd` emits this event only when
`finishMapperAttempt` accepts that attempt. If a failed or speculative attempt
writes data and a later attempt succeeds, its compressed bytes remain in the
denominator but its uncompressed bytes are absent from the numerator,
understating the ratio. Please align the accounting and test a
retry/duplicate-attempt case.
##########
client-spark/spark-3/src/main/java/org/apache/spark/shuffle/celeborn/SortBasedShuffleWriter.java:
##########
@@ -396,7 +397,13 @@ private void close(boolean iteratorHasNext) throws
IOException {
writeMetrics.incRecordsWritten(tmpRecordsWritten);
long waitStartTime = System.nanoTime();
- shuffleClient.mapperEnd(shuffleId, mapId, encodedAttemptId, numMappers,
numPartitions);
+ shuffleClient.mapperEnd(
+ shuffleId,
+ mapId,
+ encodedAttemptId,
+ numMappers,
+ numPartitions,
+ new MapperMetrics(pusher.getUncompressedBytesWritten()));
Review Comment:
Could we include bytes sent by `pushGiantRecord()` in this metric? That path
pushes directly through `shuffleClient.pushDataWithCRC()` and increments
`writeMetrics.bytesWritten`, but never updates `SortBasedPusher`'s
uncompressed-byte counter, which is the sole source used here. If all records
exceed `pushBufferMaxSize`, this reports zero and the UI shows `N/A` despite a
successful write. Please account for the direct-push path and cover it with a
giant-record 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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
