You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@celeborn.apache.org by "yew1eb (via GitHub)" <gi...@apache.org> on 2026/10/13 15:43:13 UTC

[PR] [CELEBORN-2479] Show shuffle compression ratio in Spark UI [celeborn]

yew1eb opened a new pull request, #3872:
URL: https://github.com/apache/celeborn/pull/3872

   ## What changes were proposed in this PR?
   
   Track the total bytes handed to `ShuffleClient` by each mapper **before** compression/encryption, and surface the aggregated **compression ratio (uncompressed/compressed)** on the Celeborn tab of the Spark UI.
   
   Close #CELEBORN-2479. (issue: apache/celeborn CELEBORN-2479)
   
   ### Change breakdown
   
   - **Writers** (`HashBasedShuffleWriter`, `SortBasedPusher` via `SortBasedShuffleWriter`) count uncompressed bytes and report them at `mapperEnd` through a new extensible `MapperMetrics` message.
   - **Protocol**: new `PbMapperMetrics` message carried by `PbMapperEnd` as field 10 (`metrics`); `ControlMessages.MapperEnd` gains a defaulted `MapperMetrics` parameter, so existing callers remain source/binary compatible. Future mapper-side metrics should extend `MapperMetrics` instead of `PbMapperEnd`.
   - **LifecycleManager** gains `registerMapperEndCallback`; on each successfully finished mapper attempt it invokes the callback with the metrics (only effective when LifecycleManager runs in the driver JVM, which is the Spark case).
   - **SparkShuffleManager** registers a callback that posts `CelebornMapperMetricsEvent` (a `SparkListenerEvent` with `logEvent() = true`) onto the driver listener bus, so the metric is available in the live UI **and** replayable by the History Server from the event log.
   - **UI**: `CelebornListener` aggregates the events into `AggregatedTaskInfoUIData.uncompressedShuffleBytes`; `CelebornShufflePage` renders `Compression Ratio (uncompressed/compressed): <uncompressed>/<written> = x.xx`. Old clients / unreported values degrade to `N/A`.
   - **docker**: register `spark.plugins=...CelebornPlugin` in `docker/spark/conf/spark-defaults.conf` so the local docker-compose cluster exercises the Celeborn UI tab end-to-end.
   
   ## How was this patch tested?
   
   - New unit tests: `CelebornListenerSuite.aggregateMapperMetricsEvents`, `CelebornListenerSuite.mapperMetricsEventJsonRoundtrip` (event-log roundtrip), extended `UtilsSuite` MapperEnd ser/de roundtrip with `MapperMetrics`.
   - End-to-end on the docker-compose stack (celeborn:dev + celeborn-spark:dev, Spark 3.5.9, hash writer, lz4): ran a Spark job whose shuffle went through Celeborn and confirmed the Celeborn tab shows the compression ratio while the driver is alive.
   
   πŸ€– Generated with [Claude Code](https://claude.com/claude-code)


-- 
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: issues-unsubscribe@celeborn.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] [CELEBORN-2479] Show shuffle compression ratio in Spark UI [celeborn]

Posted by "zaynt4606 (via GitHub)" <gi...@apache.org>.
zaynt4606 commented on code in PR #3872:
URL: https://github.com/apache/celeborn/pull/3872#discussion_r4332762439


##########
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:
   Thanks for the update. There is still a gap in the assumption that failed attempts never report MapperEnd: the writer can finish and report MapperEnd before Spark finishes the task. A task-completion listener can then throw, causing that attempt to fail and be retried.
   
   I reproduced this on ea7a4fa3 with Spark 3.5.8, local[2,2], two map partitions, and an injected task-completion-listener failure on partition 0 / attempt 0 after the shuffle writer completes. The retry succeeds and the job result is correct, but the metrics are:
   
   - MapperMetrics events (uncompressed bytes): 20, 20, 20, including the failed attempt.
   - Successful map attempts (compressed bytes): 94, 94.
   - Failed map attempt (compressed bytes): 94, excluded by onTaskEnd.
   - UI totals: 60 / 188, rather than 40 / 188 for the successful attempts, overstating the ratio by 50%.
   
   The onTaskEnd and mapper-event accumulation code, both writers, and the event payload are unchanged in 96f8cc84. The event still lacks map/attempt identity to reconcile it with the final task outcome. Could we derive both counters from the same set of attempts (with enough identity to reconcile them if using task-success filtering) and add a regression where the task fails after writer completion? This is distinct from deduplicating MapperEnd RPC retries.



-- 
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: issues-unsubscribe@celeborn.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] [CELEBORN-2479] Show shuffle compression ratio in Spark UI [celeborn]

Posted by "zaynt4606 (via GitHub)" <gi...@apache.org>.
zaynt4606 commented on PR #3872:
URL: https://github.com/apache/celeborn/pull/3872#issuecomment-6358610023

   Thanks @yew1eb for the contribution!
   Merged to main(v1.0.0).
   
   Commit: https://github.com/apache/celeborn/commit/3cb708b05e4a21b3d336b2628c461b4990e9f2aa


-- 
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: issues-unsubscribe@celeborn.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] [CELEBORN-2479] Show shuffle compression ratio in Spark UI [celeborn]

Posted by "zaynt4606 (via GitHub)" <gi...@apache.org>.
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: issues-unsubscribe@celeborn.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] [CELEBORN-2479] Show shuffle compression ratio in Spark UI [celeborn]

Posted by "yew1eb (via GitHub)" <gi...@apache.org>.
yew1eb commented on PR #3872:
URL: https://github.com/apache/celeborn/pull/3872#issuecomment-6310215833

   Hi @zaynt4606, thanks for the detailed review and the runtime reproductions! All three comments are addressed in cd1636e79 β€” please see the inline replies for details.
   
   Verified with: full compile on the spark-3.5 and spark-4.0 (Scala 2.13) profiles, `CelebornListenerSuite`, both shuffle writer suites, `UtilsSuite` serde round-trip, and the spark-it `LifecycleManagerSuite` against a real mini cluster.
   
   PTAL, thanks!


-- 
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: issues-unsubscribe@celeborn.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] [CELEBORN-2479] Show shuffle compression ratio in Spark UI [celeborn]

Posted by "zaynt4606 (via GitHub)" <gi...@apache.org>.
zaynt4606 commented on PR #3872:
URL: https://github.com/apache/celeborn/pull/3872#issuecomment-6358588427

   **Merge Summary:**
   - merged into main https://github.com/apache/celeborn/commit/3cb708b0
   
   *Posted by `merge_pr.py`*


-- 
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: issues-unsubscribe@celeborn.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] [CELEBORN-2479] Show shuffle compression ratio in Spark UI [celeborn]

Posted by "zaynt4606 (via GitHub)" <gi...@apache.org>.
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: issues-unsubscribe@celeborn.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] [CELEBORN-2479] Show shuffle compression ratio in Spark UI [celeborn]

Posted by "ForVic (via GitHub)" <gi...@apache.org>.
ForVic commented on code in PR #3872:
URL: https://github.com/apache/celeborn/pull/3872#discussion_r4343473628


##########
client-spark/spark-3/src/main/java/org/apache/spark/shuffle/celeborn/SparkShuffleManager.java:
##########
@@ -162,6 +163,19 @@ private void initializeLifecycleManager(String appId) {
           lifecycleManager = new LifecycleManager(appUniqueId, celebornConf);
           lifecycleManager.applicationCount().increment();
           lifecycleManager.registerCancelShuffleCallback(SparkUtils::cancelShuffle);
+          // Forward mapper metrics to the Celeborn UI extension.
+          lifecycleManager.registerMapperEndCallback(

Review Comment:
   This is registered unconditionally, so every app logs one event per map task even without CelebornPlugin enabled. EventLoggingListener also flushes the event log for each custom event.
   
   Could we gate this on the plugin? Or, building on the LifecycleManager comment, aggregate per shuffle and post once at StageEnd.



##########
client-spark/spark-3-ui/src/main/scala/org/apache/spark/shuffle/celeborn/CelebornListener.scala:
##########
@@ -62,16 +67,42 @@ 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. The compression ratio

Review Comment:
   I think this is left over from the earlier fix. The ratio comes entirely from MapperEnd now.
   
   It also makes "Shuffle Duration / Task Duration" inconsistent: write time only counts successful attempts, but task duration still counts all of them. And Spark's AppStatusListener counts metrics from failed/killed attempts too, so this doesn't match Spark's stage accounting either. Can we revert it?



##########
client-spark/spark-3/src/test/java/org/apache/spark/shuffle/celeborn/CelebornShuffleWriterSuiteBase.java:
##########
@@ -288,6 +289,16 @@ private void checkWithIntegrity(
     int[] crcPerPartition = pushState.getCRC32PerPartition(true, numPartitions);
     long[] bytesPerPartition = pushState.getBytesWrittenPerPartition(true, numPartitions);
 
+    MapperMetrics mapperMetrics = client.getMapperEndMetrics(shuffleId, 0, 0);
+    assertNotNull(mapperMetrics);
+    long totalBytesPushed = 0;
+    for (long bytes : bytesPerPartition) {
+      totalBytesPushed += bytes;
+    }
+    assertEquals(totalBytesPushed, mapperMetrics.uncompressedBytesWritten());

Review Comment:
   DummyShuffleClient returns the input length, so uncompressed == compressed here and a swap in the writer would still pass.
   
   Could the dummy client return a different byte count so these assertions actually distinguish the two?



##########
client/src/main/java/org/apache/celeborn/client/ShuffleClient.java:
##########
@@ -372,9 +373,19 @@ public abstract int mergeData(
 
   // Report partition locations written by the completed map task of ReducePartition Shuffle Type.
   public abstract void mapperEnd(
-      int shuffleId, int mapId, int attemptId, int numMappers, int numPartitions)
+      int shuffleId,

Review Comment:
   Making the new overload abstract breaks external ShuffleClient subclasses, and it's the opposite of what the description says.
   
   Could we keep the 5-arg one abstract and make this one a default that ignores the metrics?



-- 
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: issues-unsubscribe@celeborn.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] [CELEBORN-2479] Show shuffle compression ratio in Spark UI [celeborn]

Posted by "yew1eb (via GitHub)" <gi...@apache.org>.
yew1eb commented on code in PR #3872:
URL: https://github.com/apache/celeborn/pull/3872#discussion_r4268166033


##########
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:
   Good catch. `pushGiantRecord` now increments a writer-level giant-bytes counter that is summed with the pusher's counter in the `MapperMetrics` reported at mapperEnd, mirroring `HashBasedShuffleWriter`. The writer suites now assert the reported uncompressed bytes equal the bytes that reached the shuffle client β€” the all-giant-record tests (`testGiantRecord`, `testGiantRecordWithFastWrite`, which force every record above `pushBufferMaxSize`) fail without the fix, and the mixed-path `checkWithIntegrity` covers giant + buffered records together.



##########
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:
   Aligned. Both sides now use the same set of mapper attempts:
   - The listener counts `shuffleWriteMetrics` only for successful task ends (failed/killed attempts push bytes whose MapperEnd is never reported).
   - `LifecycleManager` reports metrics for every successfully closed writer attempt β€” including speculative/rerun attempts whose MapperEnd the commit handler rejects β€” deduplicated by attemptId so a retried MapperEnd RPC of the same attempt is not counted twice.
   
   Covered by `CelebornListenerSuite.shuffleWriteMetricsCountOnlySuccessfulAttempts` (success + killed attempts + mapper events) and a `LifecycleManagerSuite` test driving the accepted / retried / duplicate-attempt cases over real RPC.



-- 
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: issues-unsubscribe@celeborn.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] [CELEBORN-2479] Show shuffle compression ratio in Spark UI [celeborn]

Posted by "yew1eb (via GitHub)" <gi...@apache.org>.
yew1eb commented on code in PR #3872:
URL: https://github.com/apache/celeborn/pull/3872#discussion_r4348055247


##########
client-spark/spark-3/src/test/java/org/apache/spark/shuffle/celeborn/CelebornShuffleWriterSuiteBase.java:
##########
@@ -288,6 +289,16 @@ private void checkWithIntegrity(
     int[] crcPerPartition = pushState.getCRC32PerPartition(true, numPartitions);
     long[] bytesPerPartition = pushState.getBytesWrittenPerPartition(true, numPartitions);
 
+    MapperMetrics mapperMetrics = client.getMapperEndMetrics(shuffleId, 0, 0);
+    assertNotNull(mapperMetrics);
+    long totalBytesPushed = 0;
+    for (long bytes : bytesPerPartition) {
+      totalBytesPushed += bytes;
+    }
+    assertEquals(totalBytesPushed, mapperMetrics.uncompressedBytesWritten());

Review Comment:
   Good catch β€” and it was slightly worse than that: the SortBased async path pushes through `DataPusher`'s bare `pushData` (CRC already recorded at enqueue time), which bypassed the WithCRC overloads entirely. Fixed in 45ee042a3 by making the dummy's `pushData`/`mergeData` themselves return `length * 3 / 4` β€” like the real client, every push return value is a post-compression byte count β€” so all paths agree. The suites now assert `uncompressed == bytes pushed` and `compressed == Ξ£ 3/4Β·batch` separately; I verified a temporary swap of the two `MapperMetrics` fields turns 6 of 13 SortBased tests red.



##########
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:
   Agreed, adopted in 90d35c165. Since both ratio sides travel in the same MapperMetrics payload, reporting only the accepted attempt keeps them aligned, `finishMapperAttempt` already dedups by mapId (same-attempt retries and rejected attempts all return false), and the dedup set is gone along with its expiry cleanup. The regression test now replays your probe (attempts 0, 0, 1, 1 / 100, 100, 250, 250) and asserts exactly `[100]`.



-- 
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: issues-unsubscribe@celeborn.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] [CELEBORN-2479] Show shuffle compression ratio in Spark UI [celeborn]

Posted by "zaynt4606 (via GitHub)" <gi...@apache.org>.
zaynt4606 closed pull request #3872: [CELEBORN-2479] Show shuffle compression ratio in Spark UI
URL: https://github.com/apache/celeborn/pull/3872


-- 
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: issues-unsubscribe@celeborn.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] [CELEBORN-2479] Show shuffle compression ratio in Spark UI [celeborn]

Posted by "yew1eb (via GitHub)" <gi...@apache.org>.
yew1eb commented on PR #3872:
URL: https://github.com/apache/celeborn/pull/3872#issuecomment-6287935742

   @zaynt4606 


-- 
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: issues-unsubscribe@celeborn.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] [CELEBORN-2479] Show shuffle compression ratio in Spark UI [celeborn]

Posted by "ForVic (via GitHub)" <gi...@apache.org>.
ForVic commented on code in PR #3872:
URL: https://github.com/apache/celeborn/pull/3872#discussion_r4343519444


##########
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:
   Now that compressed bytes also come through MapperEnd, do we still need to report rejected attempts?
   
   If we only report when mapperAttemptFinishedSuccess is true, both sides of the ratio still use the same attempt set, and finishMapperAttempt already dedups by mapId. Then this set can go away.
   
   It's a boxed Long per mapper on the driver, and it's populated even when no callback is registered.



-- 
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: issues-unsubscribe@celeborn.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] [CELEBORN-2479] Show shuffle compression ratio in Spark UI [celeborn]

Posted by "zaynt4606 (via GitHub)" <gi...@apache.org>.
zaynt4606 commented on PR #3872:
URL: https://github.com/apache/celeborn/pull/3872#issuecomment-6355979086

   Thanks @yew1eb for addressing the review feedback! I have approved e0771482 and updated the PR description to match the current implementation.
   
   Please sync this branch with the latest `main` and resolve the import conflict in `SparkShuffleManager.java`, then let CI run on the updated head before merging. The current SBT service failure is the existing Master HA test race in `exportedGauge` (`head of empty list`), unrelated to this change.


-- 
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: issues-unsubscribe@celeborn.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] [CELEBORN-2479] Show shuffle compression ratio in Spark UI [celeborn]

Posted by "yew1eb (via GitHub)" <gi...@apache.org>.
yew1eb commented on code in PR #3872:
URL: https://github.com/apache/celeborn/pull/3872#discussion_r4348052295


##########
client-spark/spark-3-ui/src/main/scala/org/apache/spark/shuffle/celeborn/CelebornListener.scala:
##########
@@ -62,16 +67,42 @@ 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. The compression ratio

Review Comment:
   You are right, it was left over from the earlier split-source ratio β€” reverted in 5643bb773. Write bytes/time now count every attempt like the read side and task duration, matching `AppStatusListener`'s stage accounting. `shuffleWriteMetricsCountEveryAttempt` covers a killed attempt being counted.



##########
client/src/main/java/org/apache/celeborn/client/ShuffleClient.java:
##########
@@ -372,9 +373,19 @@ public abstract int mergeData(
 
   // Report partition locations written by the completed map task of ReducePartition Shuffle Type.
   public abstract void mapperEnd(
-      int shuffleId, int mapId, int attemptId, int numMappers, int numPartitions)
+      int shuffleId,

Review Comment:
   Fixed in 45ee042a3 β€” the 5-arg `mapperEnd` is abstract again, and the new 6-arg overload is a default method that discards the metrics and delegates to the 5-arg one, so external subclasses keep compiling with just the 5-arg implementation. `ShuffleClientImpl` overrides the 6-arg one to pass the metrics through.



-- 
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: issues-unsubscribe@celeborn.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] [CELEBORN-2479] Show shuffle compression ratio in Spark UI [celeborn]

Posted by "yew1eb (via GitHub)" <gi...@apache.org>.
yew1eb commented on code in PR #3872:
URL: https://github.com/apache/celeborn/pull/3872#discussion_r4348048528


##########
client-spark/spark-3/src/main/java/org/apache/spark/shuffle/celeborn/SparkShuffleManager.java:
##########
@@ -162,6 +163,19 @@ private void initializeLifecycleManager(String appId) {
           lifecycleManager = new LifecycleManager(appUniqueId, celebornConf);
           lifecycleManager.applicationCount().increment();
           lifecycleManager.registerCancelShuffleCallback(SparkUtils::cancelShuffle);
+          // Forward mapper metrics to the Celeborn UI extension.
+          lifecycleManager.registerMapperEndCallback(

Review Comment:
   Gated in da2330f98 β€” the callback is now registered only when `SparkUtils.isCelebornPluginEnabled(conf)` finds the plugin class in `spark.plugins` or `spark.plugins.defaultList` (the same keys the listener checks on replay), so apps that never opted in no longer post one event per map task or flush the event log per event.
   
   Aggregating per shuffle and posting once at StageEnd would additionally cut the event volume for opted-in apps; I have left it as a follow-up to keep this change small.



-- 
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: issues-unsubscribe@celeborn.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] [CELEBORN-2479] Show shuffle compression ratio in Spark UI [celeborn]

Posted by "yew1eb (via GitHub)" <gi...@apache.org>.
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: issues-unsubscribe@celeborn.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org