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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
