You are viewing a plain text version of this content. The canonical link for it is here.
Posted to reviews@spark.apache.org by GitBox <gi...@apache.org> on 2022/04/27 23:14:19 UTC
[GitHub] [spark] otterc commented on a diff in pull request #36165: [SPARK-36620][SHUFFLE] Add Push Based Shuffle client side metrics
otterc commented on code in PR #36165:
URL: https://github.com/apache/spark/pull/36165#discussion_r860171613
##########
core/src/main/scala/org/apache/spark/InternalAccumulator.scala:
##########
@@ -54,13 +56,26 @@ private[spark] object InternalAccumulator {
val LOCAL_BYTES_READ = SHUFFLE_READ_METRICS_PREFIX + "localBytesRead"
val FETCH_WAIT_TIME = SHUFFLE_READ_METRICS_PREFIX + "fetchWaitTime"
val RECORDS_READ = SHUFFLE_READ_METRICS_PREFIX + "recordsRead"
+ val REMOTE_REQS_DURATION = SHUFFLE_READ_METRICS_PREFIX + "remoteReqsDuration"
+ val CORRUPT_MERGED_BLOCK_CHUNKS = SHUFFLE_PUSH_READ_METRICS_PREFIX + "corruptMergedBlockChunks"
+ val ORIGINAL_BLOCK_FALLBACK_COUNT = SHUFFLE_PUSH_READ_METRICS_PREFIX + "fallbackCount"
Review Comment:
Nit: `ORIGINAL_BLOCK_FALLBACK_COUNT` -> `FALLBACK_COUNT`. This keeps track of how many times fallback is triggered. The prefix `ORIGINAL_BLOCK` seems odd
##########
core/src/main/scala/org/apache/spark/executor/ShuffleReadMetrics.scala:
##########
@@ -84,13 +94,76 @@ class ShuffleReadMetrics private[spark] () extends Serializable {
*/
def totalBlocksFetched: Long = remoteBlocksFetched + localBlocksFetched
+ /**
+ * Number of corrupt merged shuffle block chunks encountered by this task (remote or local).
+ */
+ def corruptMergedBlockChunks: Long = _corruptMergedBlockChunks.sum
+
+ /**
+ * Number of times the task had to fallback to fetch original shuffle blocks for a merged
+ * shuffle block chunk (remote or local).
+ */
+ def fallbackCount: Long = _fallbackCount.sum
+
+ /**
+ * Number of remote merged blocks fetched.
+ */
+ def remoteMergedBlocksFetched: Long = _remoteMergedBlocksFetched.sum
+
+ /**
+ * Number of local merged blocks fetched.
+ */
+ def localMergedBlocksFetched: Long = _localMergedBlocksFetched.sum
+
+ /**
+ * Number of remote merged chunks fetched.
+ */
+ def remoteMergedChunksFetched: Long = _remoteMergedChunksFetched.sum
+
+ /**
+ * Number of local merged chunks fetched.
+ */
+ def localMergedChunksFetched: Long = _localMergedChunksFetched.sum
+
+ /**
+ * Total number of remote merged bytes read.
+ */
+ def remoteMergedBlocksBytesRead: Long = _remoteMergedBlocksBytesRead.sum
+
+ /**
+ * Total number of local merged bytes read.
+ */
+ def localMergedBlocksBytesRead: Long = _localMergedBlocksBytesRead.sum
+
+ /**
+ * Total time taken for remote requests.
+ */
+ def remoteReqsDuration: Long = _remoteReqsDuration.sum + _remoteMergedReqsDuration.sum
+
Review Comment:
In the PR for the jira SPARK-32922, it was decided to not add push-merge related metrics to original/unmerged metrics. For example totalRemoteShuffleBlocksFetched in master doesn't include remoteMergedShuffleBlocks. I prefer consistency so even for this metric, my recommendation is to not add `remoteMergedReqsDuration` to `remoteReqsDuration`
##########
core/src/main/scala/org/apache/spark/status/AppStatusStore.scala:
##########
@@ -627,9 +709,22 @@ private[spark] class AppStatusStore(
shuffleLocalBytesRead = stage.shuffleLocalBytesRead,
shuffleReadBytes = stage.shuffleReadBytes,
shuffleReadRecords = stage.shuffleReadRecords,
+ pushBasedShuffleCorruptMergedBlockChunks = stage.pushBasedShuffleCorruptMergedBlockChunks,
+ pushBasedShuffleFallbackCount = stage.pushBasedShuffleFallbackCount,
+ pushBasedShuffleMergedRemoteBlocksFetched = stage.pushBasedShuffleMergedRemoteBlocksFetched,
+ pushBasedShuffleMergedLocalBlocksFetched = stage.pushBasedShuffleMergedLocalBlocksFetched,
+ pushBasedShuffleMergedRemoteChunksFetched = stage.pushBasedShuffleMergedRemoteChunksFetched,
+ pushBasedShuffleMergedLocalChunksFetched = stage.pushBasedShuffleMergedLocalChunksFetched,
+ pushBasedShuffleMergedRemoteBytesRead = stage.pushBasedShuffleMergedRemoteBytesRead,
+ pushBasedShuffleMergedLocalBytesRead = stage.pushBasedShuffleMergedLocalBytesRead,
+ pushBasedShuffleRemoteReqsDuration = stage.pushBasedShuffleRemoteReqsDuration,
Review Comment:
This looks off. This shouldn't be prefixed by `pushBasedShuffle`
##########
core/src/main/scala/org/apache/spark/scheduler/StageInfo.scala:
##########
@@ -40,8 +40,8 @@ class StageInfo(
private[spark] val taskLocalityPreferences: Seq[Seq[TaskLocation]] = Seq.empty,
private[spark] val shuffleDepId: Option[Int] = None,
val resourceProfileId: Int,
- private[spark] var isPushBasedShuffleEnabled: Boolean = false,
- private[spark] var shuffleMergerCount: Int = 0) {
+ var isPushBasedShuffleEnabled: Boolean = false,
+ var shuffleMergerCount: Int = 0) {
Review Comment:
Is this change needed for a test?
##########
core/src/main/scala/org/apache/spark/status/AppStatusStore.scala:
##########
@@ -448,10 +504,36 @@ private[spark] class AppStatusStore(
shuffleRemoteBytesReadToDisk =
computedQuantiles.shuffleReadMetrics.remoteBytesReadToDisk(idx),
shuffleTotalBlocksFetched = computedQuantiles.shuffleReadMetrics.totalBlocksFetched(idx),
+ pushBasedShuffleCorruptMergedBlockChunks =
+ computedQuantiles.shuffleReadMetrics.pushBased.corruptMergedBlockChunks(idx),
+ pushBasedShuffleFallbackCount =
+ computedQuantiles.shuffleReadMetrics.pushBased.fallbackCount(idx),
+ pushBasedShuffleMergedRemoteBlocksFetched =
+ computedQuantiles.shuffleReadMetrics.pushBased.remoteMergedBlocksFetched(idx),
+ pushBasedShuffleMergedLocalBlocksFetched =
+ computedQuantiles.shuffleReadMetrics.pushBased.localMergedBlocksFetched(idx),
+ pushBasedShuffleMergedRemoteChunksFetched =
+ computedQuantiles.shuffleReadMetrics.pushBased.remoteMergedChunksFetched(idx),
+ pushBasedShuffleMergedLocalChunksFetched =
+ computedQuantiles.shuffleReadMetrics.pushBased.localMergedChunksFetched(idx),
+ pushBasedShuffleMergedRemoteBlocksBytesRead =
+ computedQuantiles.shuffleReadMetrics.pushBased.remoteMergedBlocksBytesRead(idx),
+ pushBasedShuffleMergedLocalBlocksBytesRead =
Review Comment:
Nit: Please check the indentation
##########
core/src/main/scala/org/apache/spark/executor/TaskMetrics.scala:
##########
@@ -227,9 +227,23 @@ class TaskMetrics private[spark] () extends Serializable {
shuffleRead.LOCAL_BYTES_READ -> shuffleReadMetrics._localBytesRead,
shuffleRead.FETCH_WAIT_TIME -> shuffleReadMetrics._fetchWaitTime,
shuffleRead.RECORDS_READ -> shuffleReadMetrics._recordsRead,
+ shuffleRead.CORRUPT_MERGED_BLOCK_CHUNKS ->
+ shuffleReadMetrics._corruptMergedBlockChunks,
+ shuffleRead.ORIGINAL_BLOCK_FALLBACK_COUNT -> shuffleReadMetrics._fallbackCount,
+ shuffleRead.REMOTE_MERGED_BLOCKS_FETCHED -> shuffleReadMetrics._remoteMergedBlocksFetched,
+ shuffleRead.LOCAL_MERGED_BLOCKS_FETCHED -> shuffleReadMetrics._localMergedBlocksFetched,
+ shuffleRead.REMOTE_MERGED_CHUNKS_FETCHED -> shuffleReadMetrics._remoteMergedChunksFetched,
+ shuffleRead.LOCAL_MERGED_CHUNKS_FETCHED -> shuffleReadMetrics._localMergedChunksFetched,
+ shuffleRead.REMOTE_MERGED_BLOCKS_READ -> shuffleReadMetrics._remoteMergedBlocksBytesRead,
+ shuffleRead.LOCAL_MERGED_BLOCKS_READ -> shuffleReadMetrics._localMergedBlocksBytesRead,
+ shuffleRead.REMOTE_REQS_DURATION -> shuffleReadMetrics._remoteReqsDuration,
Review Comment:
Nit: indentation
##########
core/src/main/scala/org/apache/spark/status/LiveEntity.scala:
##########
@@ -502,9 +527,30 @@ private class LiveStage(var info: StageInfo) extends LiveEntity {
shuffleReadBytes =
metrics.shuffleReadMetrics.localBytesRead + metrics.shuffleReadMetrics.remoteBytesRead,
shuffleReadRecords = metrics.shuffleReadMetrics.recordsRead,
+ pushBasedShuffleCorruptMergedBlockChunks =
+ metrics.shuffleReadMetrics.pushBased.corruptMergedBlockChunks,
+ pushBasedShuffleFallbackCount = metrics.shuffleReadMetrics.pushBased.fallbackCount,
+ pushBasedShuffleMergedRemoteBlocksFetched =
+ metrics.shuffleReadMetrics.pushBased.remoteMergedBlocksFetched,
+ pushBasedShuffleMergedLocalBlocksFetched =
+ metrics.shuffleReadMetrics.pushBased.localMergedBlocksFetched,
+ pushBasedShuffleMergedRemoteChunksFetched =
+ metrics.shuffleReadMetrics.pushBased.remoteMergedChunksFetched,
+ pushBasedShuffleMergedLocalChunksFetched =
+ metrics.shuffleReadMetrics.pushBased.localMergedChunksFetched,
+ pushBasedShuffleMergedRemoteBytesRead =
+ metrics.shuffleReadMetrics.pushBased.remoteMergedBlocksBytesRead,
+ pushBasedShuffleMergedLocalBytesRead =
+ metrics.shuffleReadMetrics.pushBased.localMergedBlocksBytesRead,
+ pushBasedShuffleRemoteReqsDuration = metrics.shuffleReadMetrics.remoteReqsDuration,
Review Comment:
Same here. This is not specific to pushBased shuffle
##########
core/src/main/scala/org/apache/spark/status/LiveEntity.scala:
##########
@@ -843,9 +936,35 @@ private[spark] object LiveEntityHelpers {
m2.shuffleReadMetrics.remoteBytesReadToDisk * mult,
m1.shuffleReadMetrics.localBytesRead + m2.shuffleReadMetrics.localBytesRead * mult,
m1.shuffleReadMetrics.recordsRead + m2.shuffleReadMetrics.recordsRead * mult,
+ m1.shuffleReadMetrics.pushBased.corruptMergedBlockChunks +
+ m2.shuffleReadMetrics.pushBased.corruptMergedBlockChunks * mult,
+ m1.shuffleReadMetrics.pushBased.fallbackCount +
+ m2.shuffleReadMetrics.pushBased.fallbackCount * mult,
+ m1.shuffleReadMetrics.pushBased.remoteMergedBlocksFetched +
+ m2.shuffleReadMetrics.pushBased.remoteMergedBlocksFetched * mult,
+ m1.shuffleReadMetrics.pushBased.localMergedBlocksFetched +
+ m2.shuffleReadMetrics.pushBased.localMergedBlocksFetched * mult,
+ m1.shuffleReadMetrics.pushBased.remoteMergedChunksFetched +
+ m2.shuffleReadMetrics.pushBased.remoteMergedChunksFetched * mult,
+ m1.shuffleReadMetrics.pushBased.localMergedChunksFetched +
+ m2.shuffleReadMetrics.pushBased.localMergedChunksFetched * mult,
+ m1.shuffleReadMetrics.pushBased.remoteMergedBlocksBytesRead +
+ m2.shuffleReadMetrics.pushBased.remoteMergedBlocksBytesRead * mult,
+ m1.shuffleReadMetrics.pushBased.localMergedBlocksBytesRead +
+ m2.shuffleReadMetrics.pushBased.localMergedBlocksBytesRead * mult,
+ m1.shuffleReadMetrics.remoteReqsDuration + m2.shuffleReadMetrics.remoteReqsDuration * mult,
+ m1.shuffleReadMetrics.pushBased.remoteMergedReqsDuration +
+ m2.shuffleReadMetrics.pushBased.remoteMergedChunksFetched * mult,
Review Comment:
This looks off.
--
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: reviews-unsubscribe@spark.apache.org
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