[GitHub] [spark] jaceklaskowski commented on a change in pull request #32388: [SPARK-35258][SHUFFLE][YARN] Add new metrics to ExternalShuffleService for better monitoring

[GitBox <gi...@apache.org>]

jaceklaskowski commented on a change in pull request #32388:
URL: https://github.com/apache/spark/pull/32388#discussion_r622875712



##########
File path: common/network-shuffle/src/main/java/org/apache/spark/network/shuffle/ExternalBlockHandler.java
##########
@@ -98,10 +99,10 @@ public ExternalBlockHandler(
       OneForOneStreamManager streamManager,
       ExternalShuffleBlockResolver blockManager,
       MergedShuffleFileManager mergeManager) {
-    this.metrics = new ShuffleMetrics();
     this.streamManager = streamManager;
     this.blockManager = blockManager;
     this.mergeManager = mergeManager;
+    this.metrics = new ShuffleMetrics();

Review comment:
       Why has this been moved down? Curious.

##########
File path: common/network-shuffle/src/main/java/org/apache/spark/network/shuffle/ExternalBlockHandler.java
##########
@@ -291,9 +294,18 @@ public ShuffleMetrics() {
       allMetrics.put("openBlockRequestLatencyMillis", openBlockRequestLatencyMillis);
       allMetrics.put("registerExecutorRequestLatencyMillis", registerExecutorRequestLatencyMillis);
       allMetrics.put("finalizeShuffleMergeLatencyMillis", finalizeShuffleMergeLatencyMillis);
+      allMetrics.put("blockTransferRate", blockTransferRate);
       allMetrics.put("blockTransferRateBytes", blockTransferRateBytes);
+      allMetrics.put("blockTransferAvgSize_1min", new RatioGauge() {
+        @Override
+        protected Ratio getRatio() {
+          return Ratio.of(
+              blockTransferRateBytes.getOneMinuteRate(),
+              blockTransferRate.getOneMinuteRate());
+        }
+      });
       allMetrics.put("registeredExecutorsSize",
-                     (Gauge<Integer>) () -> blockManager.getRegisteredExecutorsSize());
+          (Gauge<Integer>) blockManager::getRegisteredExecutorsSize);

Review comment:
       Been a while since I was with Java, and I'm curious whether the change is going to work with Java 8 (that I assume Spark 3.x keeps supporting)?

##########
File path: resource-managers/yarn/src/test/scala/org/apache/spark/network/yarn/YarnShuffleServiceSuite.scala
##########
@@ -403,7 +403,9 @@ class YarnShuffleServiceSuite extends SparkFunSuite with Matchers with BeforeAnd
     val metrics = metricSetRef.get(metricsSource).asInstanceOf[MetricSet].getMetrics
 
     assert(metrics.keySet().asScala == Set(
+      "blockTransferRate",

Review comment:
       Why not to use the same trick "// Use sorted Seq instead of Set for easier comparison when there is a mismatch" here?

##########
File path: docs/monitoring.md
##########
@@ -1361,7 +1361,9 @@ Note: applies when running in Spark standalone as worker
 ### Component instance = shuffleService
 Note: applies to the shuffle service
 
+- blockTransferRate (meter)
 - blockTransferRateBytes (meter)
+- blockTransferAvgTime_1min (gauge - 1 minute moving average)

Review comment:
       nit: 1-minute (with the dash)

##########
File path: resource-managers/yarn/src/test/scala/org/apache/spark/network/yarn/YarnShuffleServiceMetricsSuite.scala
##########
@@ -37,28 +38,57 @@ class YarnShuffleServiceMetricsSuite extends SparkFunSuite with Matchers {
   val metrics = new ExternalBlockHandler(streamManager, blockResolver).getAllMetrics
 
   test("metrics named as expected") {
-    val allMetrics = Set(
+    val allMetrics = Seq(

Review comment:
       Why `Seq` here?




-- 
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.

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



---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org