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/12 17:52:54 UTC
[GitHub] [spark] thejdeep opened a new pull request, #36165: [SPARK-36620][SHUFFLE] Add Push Based Shuffle client side metrics
thejdeep opened a new pull request, #36165:
URL: https://github.com/apache/spark/pull/36165
### What changes were proposed in this pull request?
This adds the following push based shuffle metrics like :
- Merger count and magnet enabled/disabled for stage
- Time spent on results finalization
- Counting actual number of blocks other than chunks
- Corrupt shuffle blocks chunks and fallback
### Why are the changes needed?
These changes help to understand the push based shuffle metrics of the application
### Does this PR introduce _any_ user-facing change?
Changes to API responses by SHS (eg: /stages)
### How was this patch tested?
Modified existing unit tests and also tested API response on event log files in SHS
--
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
[GitHub] [spark] thejdeep commented on a diff in pull request #36165: [SPARK-36620][SHUFFLE] Add Push Based Shuffle client side metrics
Posted by GitBox <gi...@apache.org>.
thejdeep commented on code in PR #36165:
URL: https://github.com/apache/spark/pull/36165#discussion_r1010566209
##########
core/src/test/scala/org/apache/spark/util/JsonProtocolSuite.scala:
##########
@@ -2623,48 +2677,118 @@ private[spark] object JsonProtocolSuite extends Assertions {
| },
| {
| "ID": 19,
+ | "Name": "${shuffleRead.FALLBACK_COUNT}",
+ | "Update": 0,
+ | "Internal": true,
+ | "Count Failed Values": true
+ | },
+ | {
+ | "ID" : 20,
+ | "Name" : "${shuffleRead.REMOTE_MERGED_BLOCKS_FETCHED}",
+ | "Update" : 0,
+ | "Internal" : true,
+ | "Count Failed Values" : true
+ | },
+ | {
+ | "ID" : 21,
+ | "Name" : "${shuffleRead.LOCAL_MERGED_BLOCKS_FETCHED}",
+ | "Update" : 0,
+ | "Internal" : true,
+ | "Count Failed Values" : true
+ | },
+ | {
+ | "ID" : 22,
+ | "Name" : "${shuffleRead.REMOTE_MERGED_CHUNKS_FETCHED}",
+ | "Update" : 0,
+ | "Internal" : true,
+ | "Count Failed Values" : true
+ | },
+ | {
+ | "ID" : 23,
+ | "Name" : "${shuffleRead.LOCAL_MERGED_CHUNKS_FETCHED}",
+ | "Update" : 0,
+ | "Internal" : true,
+ | "Count Failed Values" : true
+ | },
+ | {
+ | "ID" : 24,
+ | "Name" : "${shuffleRead.REMOTE_MERGED_BLOCKS_READ}",
+ | "Update" : 0,
+ | "Internal" : true,
+ | "Count Failed Values" : true
+ | },
+ | {
+ | "ID" : 25,
+ | "Name" : "${shuffleRead.LOCAL_MERGED_BLOCKS_READ}",
+ | "Update" : 0,
+ | "Internal" : true,
+ | "Count Failed Values" : true
+ | },
+ | {
+ | "ID" : 26,
+ | "Name" : "${shuffleRead.REMOTE_REQS_DURATION}",
+ | "Update" : 0,
+ | "Internal" : true,
+ | "Count Failed Values" : true
+ | },
+ | {
+ | "ID" : 27,
+ | "Name" : "${shuffleRead.REMOTE_MERGED_REQS_DURATION}",
+ | "Update" : 0,
+ | "Internal" : true,
+ | "Count Failed Values" : true
+ | },
+ | {
+ | "ID": 28,
+ | "Name": "${shuffleWrite.BYTES_WRITTEN}",
+ | "Update": 0,
+ | "Internal": true,
+ | "Count Failed Values": true
+ | },
+ | {
+ | "ID": 29,
| "Name": "${shuffleWrite.RECORDS_WRITTEN}",
| "Update": 0,
| "Internal": true,
| "Count Failed Values": true
| },
| {
- | "ID": 20,
+ | "ID": 30,
| "Name": "${shuffleWrite.WRITE_TIME}",
| "Update": 0,
| "Internal": true,
| "Count Failed Values": true
| },
| {
- | "ID": 21,
+ | "ID": 31,
| "Name": "${input.BYTES_READ}",
| "Update": 2100,
| "Internal": true,
| "Count Failed Values": true
| },
| {
- | "ID": 22,
+ | "ID": 32,
| "Name": "${input.RECORDS_READ}",
| "Update": 21,
| "Internal": true,
| "Count Failed Values": true
| },
| {
- | "ID": 23,
+ | "ID": 33,
| "Name": "${output.BYTES_WRITTEN}",
| "Update": 1200,
| "Internal": true,
| "Count Failed Values": true
| },
| {
- | "ID": 24,
+ | "ID": 34,
| "Name": "${output.RECORDS_WRITTEN}",
Review Comment:
The values just changed since metrics were added that took the place of indices preceding this.
--
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
[GitHub] [spark] zhouyejoe commented on a diff in pull request #36165: [SPARK-36620][SHUFFLE] Add Push Based Shuffle client side metrics
Posted by GitBox <gi...@apache.org>.
zhouyejoe commented on code in PR #36165:
URL: https://github.com/apache/spark/pull/36165#discussion_r1012322498
##########
core/src/main/scala/org/apache/spark/executor/TaskMetrics.scala:
##########
@@ -227,6 +227,16 @@ 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.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_BYTES_READ -> shuffleReadMetrics._remoteMergedBlocksBytesRead,
+ shuffleRead.LOCAL_MERGED_BYTES_READ -> shuffleReadMetrics._localMergedBlocksBytesRead,
+ shuffleRead.REMOTE_REQS_DURATION -> shuffleReadMetrics._remoteReqsDuration,
+ shuffleRead.REMOTE_MERGED_REQS_DURATION -> shuffleReadMetrics._remoteMergedReqsDuration,
shuffleWrite.BYTES_WRITTEN -> shuffleWriteMetrics._bytesWritten,
shuffleWrite.RECORDS_WRITTEN -> shuffleWriteMetrics._recordsWritten,
shuffleWrite.WRITE_TIME -> shuffleWriteMetrics._writeTime,
Review Comment:
I think we are missing the three push related metrics for the ShuffleWriter, including "BlocksNotPushed", "BlocksTooLate" and "BlocksCollided" here, and also the other relevant part, including the metrics logging in the Executor, the rendering in SHS and also in the SQLShuffleMetricsReporter.
##########
core/src/main/scala/org/apache/spark/executor/ShuffleReadMetrics.scala:
##########
@@ -146,6 +268,16 @@ private[spark] class TempShuffleReadMetrics extends ShuffleReadMetricsReporter {
override def incLocalBytesRead(v: Long): Unit = _localBytesRead += v
override def incFetchWaitTime(v: Long): Unit = _fetchWaitTime += v
override def incRecordsRead(v: Long): Unit = _recordsRead += v
+ override def incCorruptMergedBlockChunks(v: Long): Unit = _corruptMergedBlockChunks += v
+ override def incFallbackCount(v: Long): Unit = _fallbackCount += v
+ override def incRemoteMergedBlocksFetched(v: Long): Unit = _remoteMergedBlocksFetched += v
+ override def incLocalMergedBlocksFetched(v: Long): Unit = _localMergedBlocksFetched += v
+ override def incRemoteMergedChunksFetched(v: Long): Unit = _remoteMergedChunksFetched += v
+ override def incLocalMergedChunksFetched(v: Long): Unit = _localMergedChunksFetched += v
+ override def incRemoteMergedBlocksBytesRead(v: Long): Unit = _remoteMergedBlocksBytesRead += v
+ override def incLocalMergedBlocksBytesRead(v: Long): Unit = _localMergedBlocksBytesRead += v
+ override def incRemoteReqsDuration(v: Long): Unit = _remoteReqsDuration += v
+ override def incRemoteMergedReqsDuration(v: Long): Unit = _remoteMergedReqsDuration += v
Review Comment:
I am able to find the calling for these two methods in SHS which can build the metrics up from JSON, but the caller in the ShuffleBlockFetcherIterator is actually missing in this patch. For the caller part in Executor, which measures the remote request duration, needs more refactoring in the Iterator. Probably we can carve these two metrics out in this PR. If needed, we can create another PR which tracks the remote request duration for unmerged and merged shuffle. @mridulm @otterc Thoughts?
##########
core/src/test/scala/org/apache/spark/ui/StagePageSuite.scala:
##########
@@ -78,6 +78,16 @@ class StagePageSuite extends SparkFunSuite with LocalSparkContext {
shuffleLocalBytesRead = 1L,
shuffleReadBytes = 1L,
shuffleReadRecords = 1L,
+ shuffleCorruptMergedBlockChunks = 2L,
+ shuffleFallbackCount = 2L,
Review Comment:
Any particular reason these are set to 2?
--
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
[GitHub] [spark] zhouyejoe commented on pull request #36165: [SPARK-36620][SHUFFLE] Add Push Based Shuffle client side read metrics
Posted by GitBox <gi...@apache.org>.
zhouyejoe commented on PR #36165:
URL: https://github.com/apache/spark/pull/36165#issuecomment-1351968304
There are also UT failures related to this PR @thejdeep
--
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
[GitHub] [spark] thejdeep commented on a diff in pull request #36165: [SPARK-36620][SHUFFLE] Add Push Based Shuffle client side read metrics
Posted by GitBox <gi...@apache.org>.
thejdeep commented on code in PR #36165:
URL: https://github.com/apache/spark/pull/36165#discussion_r1058019046
##########
core/src/main/protobuf/org/apache/spark/status/protobuf/store_types.proto:
##########
@@ -100,11 +100,21 @@ message TaskDataWrapper {
int64 shuffle_remote_bytes_read_to_disk = 34;
int64 shuffle_local_bytes_read = 35;
int64 shuffle_records_read = 36;
- int64 shuffle_bytes_written = 37;
- int64 shuffle_write_time = 38;
- int64 shuffle_records_written = 39;
- int64 stage_id = 40;
- int32 stage_attempt_id = 41;
+ int64 shuffle_corrupt_merged_block_chunks = 37;
+ int64 shuffle_merged_fetch_fallback_count = 38;
+ int64 shuffle_merged_remote_blocks_fetched = 39;
+ int64 shuffle_merged_local_blocks_fetched = 40;
+ int64 shuffle_merged_remote_chunks_fetched = 41;
+ int64 shuffle_merged_local_chunks_fetched = 42;
+ int64 shuffle_merged_remote_bytes_read = 43;
+ int64 shuffle_merged_local_bytes_read = 44;
+ int64 shuffle_remote_reqs_duration = 45;
+ int64 shuffle_merged_remote_req_duration = 46;
+ int64 shuffle_bytes_written = 47;
+ int64 shuffle_write_time = 48;
+ int64 shuffle_records_written = 49;
+ int64 stage_id = 50;
+ int32 stage_attempt_id = 51;
Review Comment:
@mridulm Do you suggest we keep the attribute `id`s same as before ? This would mean having to change the function definitions across most of the diff in this PR
--
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
[GitHub] [spark] mridulm commented on a diff in pull request #36165: [SPARK-36620][SHUFFLE] Add Push Based Shuffle client side read metrics
Posted by GitBox <gi...@apache.org>.
mridulm commented on code in PR #36165:
URL: https://github.com/apache/spark/pull/36165#discussion_r1040453682
##########
core/src/test/resources/HistoryServerExpectations/excludeOnFailure_node_for_stage_expectation.json:
##########
@@ -81,7 +93,19 @@
"remoteBytesRead" : 0,
"remoteBytesReadToDisk" : 0,
"localBytesRead" : 0,
- "recordsRead" : 0
+ "recordsRead" : 0,
+ "pushBased": {
Review Comment:
This should be "Push Based shuffle" right ?
--
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
[GitHub] [spark] mridulm commented on a diff in pull request #36165: [SPARK-36620][SHUFFLE] Add Push Based Shuffle client side read metrics
Posted by GitBox <gi...@apache.org>.
mridulm commented on code in PR #36165:
URL: https://github.com/apache/spark/pull/36165#discussion_r1038913313
##########
core/src/main/scala/org/apache/spark/status/AppStatusStore.scala:
##########
@@ -319,7 +319,18 @@ private[spark] class AppStatusStore(
toValues(_.shuffleFetchWaitTime),
toValues(_.shuffleRemoteBytesRead),
toValues(_.shuffleRemoteBytesReadToDisk),
- toValues(_.shuffleTotalBlocksFetched)),
+ toValues(_.shuffleTotalBlocksFetched),
+ toValues(_.shuffleRemoteReqsDuration),
+ new v1.ShufflePushReadMetricDistributions(
+ toValues(_.shuffleCorruptMergedBlockChunks),
+ toValues(_.shuffleFallbackCount),
Review Comment:
`shuffleFallbackCount` and `ShufflePushReadMetricDistributions.fallbackCount` needs to be renamed as well to be consistent.
Please do make sure the other renames we did are consistent throughout as well - I will go over it once more later, but would be great if you can take a pass as well @thejdeep . Thx
##########
core/src/main/scala/org/apache/spark/status/storeTypes.scala:
##########
@@ -233,6 +243,38 @@ private[spark] class TaskDataWrapper(
val shuffleLocalBytesRead: Long,
@KVIndexParam(value = TaskIndexNames.SHUFFLE_READ_RECORDS, parent = TaskIndexNames.STAGE)
val shuffleRecordsRead: Long,
+ @KVIndexParam(
+ value = TaskIndexNames.PUSH_BASED_SHUFFLE_CORRUPT_MERGED_BLOCK_CHUNKS,
+ parent = TaskIndexNames.STAGE)
+ val shuffleCorruptMergedBlockChunks: Long,
+ @KVIndexParam(value = TaskIndexNames.PUSH_BASED_SHUFFLE_FALLBACK_COUNT,
+ parent = TaskIndexNames.STAGE)
+ val shuffleFallbackCount: Long,
+ @KVIndexParam(
+ value = TaskIndexNames.PUSH_BASED_SHUFFLE_MERGED_REMOTE_BLOCKS, parent = TaskIndexNames.STAGE)
+ val shuffleMergedRemoteBlocksFetched: Long,
+ @KVIndexParam(
+ value = TaskIndexNames.PUSH_BASED_SHUFFLE_MERGED_LOCAL_BLOCKS, parent = TaskIndexNames.STAGE)
+ val shuffleMergedLocalBlocksFetched: Long,
+ @KVIndexParam(
+ value = TaskIndexNames.PUSH_BASED_SHUFFLE_MERGED_REMOTE_CHUNKS, parent = TaskIndexNames.STAGE)
+ val shuffleMergedRemoteChunksFetched: Long,
+ @KVIndexParam(
+ value = TaskIndexNames.PUSH_BASED_SHUFFLE_MERGED_LOCAL_CHUNKS, parent = TaskIndexNames.STAGE)
+ val shuffleMergedLocalChunksFetched: Long,
+ @KVIndexParam(
+ value = TaskIndexNames.PUSH_BASED_SHUFFLE_MERGED_REMOTE_READS, parent = TaskIndexNames.STAGE)
+ val shuffleMergedRemoteBlocksBytesRead: Long,
Review Comment:
`Blocks` is not required for `BytesRead` - change here and in all other similar cases.
For ex: `shuffleMergedRemoteBlocksBytesRead` -> `shuffleMergedRemoteBytesRead`, `shuffleMergedLocalBlocksBytesRead` -> `shuffleMergedLocalBytesRead`
Same for `ShufflePushReadMetrics`, etc.
##########
core/src/main/scala/org/apache/spark/executor/Executor.scala:
##########
@@ -791,6 +770,53 @@ private[spark] class Executor(
}
}
+ private def incrementShuffleMetrics(
+ executorSource: ExecutorSource,
+ metrics: TaskMetrics
+ ): Unit = {
+ executorSource.METRIC_SHUFFLE_FETCH_WAIT_TIME
+ .inc(metrics.shuffleReadMetrics.fetchWaitTime)
+ executorSource.METRIC_SHUFFLE_WRITE_TIME.inc(metrics.shuffleWriteMetrics.writeTime)
+ executorSource.METRIC_SHUFFLE_TOTAL_BYTES_READ
+ .inc(metrics.shuffleReadMetrics.totalBytesRead)
+ executorSource.METRIC_SHUFFLE_REMOTE_BYTES_READ
+ .inc(metrics.shuffleReadMetrics.remoteBytesRead)
+ executorSource.METRIC_SHUFFLE_REMOTE_BYTES_READ_TO_DISK
+ .inc(metrics.shuffleReadMetrics.remoteBytesReadToDisk)
+ executorSource.METRIC_SHUFFLE_LOCAL_BYTES_READ
+ .inc(metrics.shuffleReadMetrics.localBytesRead)
+ executorSource.METRIC_SHUFFLE_RECORDS_READ
+ .inc(metrics.shuffleReadMetrics.recordsRead)
+ executorSource.METRIC_SHUFFLE_REMOTE_BLOCKS_FETCHED
+ .inc(metrics.shuffleReadMetrics.remoteBlocksFetched)
+ executorSource.METRIC_SHUFFLE_LOCAL_BLOCKS_FETCHED
+ .inc(metrics.shuffleReadMetrics.localBlocksFetched)
+ executorSource.METRIC_SHUFFLE_REMOTE_REQS_DURATION
+ .inc(metrics.shuffleReadMetrics.remoteReqsDuration)
+ executorSource.METRIC_SHUFFLE_BYTES_WRITTEN
+ .inc(metrics.shuffleWriteMetrics.bytesWritten)
+ executorSource.METRIC_SHUFFLE_RECORDS_WRITTEN
+ .inc(metrics.shuffleWriteMetrics.recordsWritten)
+ executorSource.METRIC_PUSH_BASED_SHUFFLE_CORRUPT_MERGED_BLOCK_CHUNKS
+ .inc(metrics.shuffleReadMetrics.corruptMergedBlockChunks)
+ executorSource.METRIC_PUSH_BASED_SHUFFLE_MERGED_FETCH_FALLBACK_COUNT
+ .inc(metrics.shuffleReadMetrics.fallbackCount)
Review Comment:
Rename `fallbackCount` (and corresponding `_fallbackCount`) to match the metric name ?
`mergedFetchFallbackCount` like in other places ?
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/metric/SQLShuffleMetricsReporter.scala:
##########
@@ -44,6 +44,26 @@ class SQLShuffleReadMetricsReporter(
metrics(SQLShuffleReadMetricsReporter.FETCH_WAIT_TIME)
private[this] val _recordsRead =
metrics(SQLShuffleReadMetricsReporter.RECORDS_READ)
Review Comment:
Keep metrics here consistent with the other renames as well.
--
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
[GitHub] [spark] mridulm commented on a diff in pull request #36165: [SPARK-36620][SHUFFLE] Add Push Based Shuffle client side metrics
Posted by GitBox <gi...@apache.org>.
mridulm commented on code in PR #36165:
URL: https://github.com/apache/spark/pull/36165#discussion_r1029927503
##########
core/src/main/scala/org/apache/spark/storage/ShuffleBlockFetcherIterator.scala:
##########
@@ -282,6 +280,17 @@ final class ShuffleBlockFetcherIterator(
}
}
+ @inline def updateMergedReqsDuration(wasReqForMergedChunks: Boolean = false): Unit = {
+ if (remainingBlocks.isEmpty) {
+ val durationMs = TimeUnit.NANOSECONDS.toMillis(clock.nanoTime() - requestStartTime)
+ if (wasReqForMergedChunks) {
+ shuffleMetrics.incRemoteMergedReqsDuration(durationMs)
+ } else {
+ shuffleMetrics.incRemoteReqsDuration(durationMs)
+ }
Review Comment:
`shuffleLocalMetricsUpdate` and `shuffleRemoteMetricsUpdate` is treating merged fetches as a subset of fetches for remote bytes - while here we are treating them separately.
Any particular reason for this difference ?
--
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
[GitHub] [spark] mridulm commented on a diff in pull request #36165: [SPARK-36620][SHUFFLE] Add Push Based Shuffle client side metrics
Posted by GitBox <gi...@apache.org>.
mridulm commented on code in PR #36165:
URL: https://github.com/apache/spark/pull/36165#discussion_r1029927503
##########
core/src/main/scala/org/apache/spark/storage/ShuffleBlockFetcherIterator.scala:
##########
@@ -282,6 +280,17 @@ final class ShuffleBlockFetcherIterator(
}
}
+ @inline def updateMergedReqsDuration(wasReqForMergedChunks: Boolean = false): Unit = {
+ if (remainingBlocks.isEmpty) {
+ val durationMs = TimeUnit.NANOSECONDS.toMillis(clock.nanoTime() - requestStartTime)
+ if (wasReqForMergedChunks) {
+ shuffleMetrics.incRemoteMergedReqsDuration(durationMs)
+ } else {
+ shuffleMetrics.incRemoteReqsDuration(durationMs)
+ }
Review Comment:
`shuffleLocalMetricsUpdate` and `shuffleRemoteMetricsUpdate` is treating merged fetches as a subset of all fetches - while here we are treating them seperately.
Any particular reason for this difference ?
--
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
[GitHub] [spark] mridulm commented on a diff in pull request #36165: [SPARK-36620][SHUFFLE] Add Push Based Shuffle client side read metrics
Posted by GitBox <gi...@apache.org>.
mridulm commented on code in PR #36165:
URL: https://github.com/apache/spark/pull/36165#discussion_r1038913697
##########
core/src/test/scala/org/apache/spark/storage/ShuffleBlockFetcherIteratorSuite.scala:
##########
@@ -1219,7 +1219,12 @@ class ShuffleBlockFetcherIteratorSuite extends SparkFunSuite with PrivateMethodT
new RuntimeException("forced error"))
}
Review Comment:
We can do this as a follow up, given the size of this PR.
Can you please file a jira for this ? 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: 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
[GitHub] [spark] thejdeep commented on a diff in pull request #36165: [SPARK-36620][SHUFFLE] Add Push Based Shuffle client side read metrics
Posted by GitBox <gi...@apache.org>.
thejdeep commented on code in PR #36165:
URL: https://github.com/apache/spark/pull/36165#discussion_r1058019046
##########
core/src/main/protobuf/org/apache/spark/status/protobuf/store_types.proto:
##########
@@ -100,11 +100,21 @@ message TaskDataWrapper {
int64 shuffle_remote_bytes_read_to_disk = 34;
int64 shuffle_local_bytes_read = 35;
int64 shuffle_records_read = 36;
- int64 shuffle_bytes_written = 37;
- int64 shuffle_write_time = 38;
- int64 shuffle_records_written = 39;
- int64 stage_id = 40;
- int32 stage_attempt_id = 41;
+ int64 shuffle_corrupt_merged_block_chunks = 37;
+ int64 shuffle_merged_fetch_fallback_count = 38;
+ int64 shuffle_merged_remote_blocks_fetched = 39;
+ int64 shuffle_merged_local_blocks_fetched = 40;
+ int64 shuffle_merged_remote_chunks_fetched = 41;
+ int64 shuffle_merged_local_chunks_fetched = 42;
+ int64 shuffle_merged_remote_bytes_read = 43;
+ int64 shuffle_merged_local_bytes_read = 44;
+ int64 shuffle_remote_reqs_duration = 45;
+ int64 shuffle_merged_remote_req_duration = 46;
+ int64 shuffle_bytes_written = 47;
+ int64 shuffle_write_time = 48;
+ int64 shuffle_records_written = 49;
+ int64 stage_id = 50;
+ int32 stage_attempt_id = 51;
Review Comment:
@mridulm Do you suggest we keep the attribute `id`s same as before ? This would mean having to change the function definitions and usages across most of the diff in this PR
--
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
[GitHub] [spark] mridulm commented on a diff in pull request #36165: [SPARK-36620][SHUFFLE] Add Push Based Shuffle client side read metrics
Posted by GitBox <gi...@apache.org>.
mridulm commented on code in PR #36165:
URL: https://github.com/apache/spark/pull/36165#discussion_r1058701274
##########
core/src/main/protobuf/org/apache/spark/status/protobuf/store_types.proto:
##########
@@ -100,11 +100,21 @@ message TaskDataWrapper {
int64 shuffle_remote_bytes_read_to_disk = 34;
int64 shuffle_local_bytes_read = 35;
int64 shuffle_records_read = 36;
- int64 shuffle_bytes_written = 37;
- int64 shuffle_write_time = 38;
- int64 shuffle_records_written = 39;
- int64 stage_id = 40;
- int32 stage_attempt_id = 41;
+ int64 shuffle_corrupt_merged_block_chunks = 37;
+ int64 shuffle_merged_fetch_fallback_count = 38;
+ int64 shuffle_merged_remote_blocks_fetched = 39;
+ int64 shuffle_merged_local_blocks_fetched = 40;
+ int64 shuffle_merged_remote_chunks_fetched = 41;
+ int64 shuffle_merged_local_chunks_fetched = 42;
+ int64 shuffle_merged_remote_bytes_read = 43;
+ int64 shuffle_merged_local_bytes_read = 44;
+ int64 shuffle_remote_reqs_duration = 45;
+ int64 shuffle_merged_remote_req_duration = 46;
+ int64 shuffle_bytes_written = 47;
+ int64 shuffle_write_time = 48;
+ int64 shuffle_records_written = 49;
+ int64 stage_id = 50;
+ int32 stage_attempt_id = 51;
Review Comment:
Protobuf id's should not be changed.
--
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
[GitHub] [spark] mridulm commented on a diff in pull request #36165: [SPARK-36620][SHUFFLE] Add Push Based Shuffle client side read metrics
Posted by GitBox <gi...@apache.org>.
mridulm commented on code in PR #36165:
URL: https://github.com/apache/spark/pull/36165#discussion_r1034238236
##########
core/src/main/scala/org/apache/spark/executor/Executor.scala:
##########
@@ -791,6 +770,53 @@ private[spark] class Executor(
}
}
+ private def incrementShuffleMetrics(
+ executorSource: ExecutorSource,
+ metrics: TaskMetrics
+ ): Unit = {
+ executorSource.METRIC_SHUFFLE_FETCH_WAIT_TIME
+ .inc(metrics.shuffleReadMetrics.fetchWaitTime)
+ executorSource.METRIC_SHUFFLE_WRITE_TIME.inc(metrics.shuffleWriteMetrics.writeTime)
+ executorSource.METRIC_SHUFFLE_TOTAL_BYTES_READ
+ .inc(metrics.shuffleReadMetrics.totalBytesRead)
+ executorSource.METRIC_SHUFFLE_REMOTE_BYTES_READ
+ .inc(metrics.shuffleReadMetrics.remoteBytesRead)
+ executorSource.METRIC_SHUFFLE_REMOTE_BYTES_READ_TO_DISK
+ .inc(metrics.shuffleReadMetrics.remoteBytesReadToDisk)
+ executorSource.METRIC_SHUFFLE_LOCAL_BYTES_READ
+ .inc(metrics.shuffleReadMetrics.localBytesRead)
+ executorSource.METRIC_SHUFFLE_RECORDS_READ
+ .inc(metrics.shuffleReadMetrics.recordsRead)
+ executorSource.METRIC_SHUFFLE_REMOTE_BLOCKS_FETCHED
+ .inc(metrics.shuffleReadMetrics.remoteBlocksFetched)
+ executorSource.METRIC_SHUFFLE_LOCAL_BLOCKS_FETCHED
+ .inc(metrics.shuffleReadMetrics.localBlocksFetched)
+ executorSource.METRIC_SHUFFLE_BYTES_WRITTEN
+ .inc(metrics.shuffleWriteMetrics.bytesWritten)
+ executorSource.METRIC_SHUFFLE_RECORDS_WRITTEN
+ .inc(metrics.shuffleWriteMetrics.recordsWritten)
+ executorSource.METRIC_PUSH_BASED_SHUFFLE_CORRUPT_MERGED_BLOCK_CHUNKS
+ .inc(metrics.shuffleReadMetrics.corruptMergedBlockChunks)
+ executorSource.METRIC_PUSH_BASED_SHUFFLE_FALLBACK_COUNT
+ .inc(metrics.shuffleReadMetrics.fallbackCount)
+ executorSource.METRIC_PUSH_BASED_SHUFFLE_MERGED_REMOTE_BLOCKS_FETCHED
+ .inc(metrics.shuffleReadMetrics.remoteMergedBlocksFetched)
+ executorSource.METRIC_PUSH_BASED_SHUFFLE_MERGED_LOCAL_BLOCKS_FETCHED
+ .inc(metrics.shuffleReadMetrics.localMergedBlocksFetched)
+ executorSource.METRIC_PUSH_BASED_SHUFFLE_MERGED_REMOTE_CHUNKS_FETCHED
+ .inc(metrics.shuffleReadMetrics.remoteMergedChunksFetched)
+ executorSource.METRIC_PUSH_BASED_SHUFFLE_MERGED_LOCAL_CHUNKS_FETCHED
+ .inc(metrics.shuffleReadMetrics.localMergedChunksFetched)
+ executorSource.METRIC_PUSH_BASED_SHUFFLE_MERGED_REMOTE_BYTES_READ
+ .inc(metrics.shuffleReadMetrics.remoteMergedBlocksBytesRead)
+ executorSource.METRIC_PUSH_BASED_SHUFFLE_MERGED_LOCAL_BLOCKS_FETCHED
+ .inc(metrics.shuffleReadMetrics.localMergedBlocksBytesRead)
+ executorSource.METRIC_PUSH_BASED_SHUFFLE_REMOTE_REQS_DURATION
+ .inc(metrics.shuffleReadMetrics.remoteReqsDuration)
Review Comment:
This is not specific to push based shuffle, right ? Why is the metric prefixed with that namespace ? (Here and in other places where this metric is referenced)
##########
core/src/main/scala/org/apache/spark/InternalAccumulator.scala:
##########
@@ -54,6 +55,16 @@ 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 FALLBACK_COUNT = SHUFFLE_PUSH_READ_METRICS_PREFIX + "fallbackCount"
Review Comment:
`fallbackCount` is fairly generic - and does not necessarily indicate it is relevant to push based shuffle - rename to make it more specific ? Something like `mergedFetchFallbackCount` or some such.
(Rename the `ExecutorSource`, other references suitably as well)
+CC @zhouyejoe, @otterc
##########
core/src/main/scala/org/apache/spark/status/storeTypes.scala:
##########
@@ -138,6 +138,16 @@ private[spark] object TaskIndexNames {
final val SHUFFLE_WRITE_RECORDS = "swr"
final val SHUFFLE_WRITE_SIZE = "sws"
final val SHUFFLE_WRITE_TIME = "swt"
+ final val PUSH_BASED_SHUFFLE_CORRUPT_MERGED_BLOCK_CHUNKS = "pbscmbc"
+ final val PUSH_BASED_SHUFFLE_FALLBACK_COUNT = "pbsfc"
+ final val PUSH_BASED_SHUFFLE_MERGED_REMOTE_BLOCKS = "pbsmrb"
+ final val PUSH_BASED_SHUFFLE_MERGED_LOCAL_BLOCKS = "pbsmlb"
+ final val PUSH_BASED_SHUFFLE_MERGED_REMOTE_CHUNKS = "pbsmrc"
+ final val PUSH_BASED_SHUFFLE_MERGED_LOCAL_CHUNKS = "pbsmlc"
+ final val PUSH_BASED_SHUFFLE_MERGED_REMOTE_READS = "pbsmrr"
+ final val PUSH_BASED_SHUFFLE_MERGED_LOCAL_READS = "pbsmlr"
+ final val PUSH_BASED_SHUFFLE_REMOTE_REQS_DURATION = "pbsrrd"
Review Comment:
This is not specific to push based shuffle - `srrd` instead ?
##########
core/src/main/scala/org/apache/spark/storage/ShuffleBlockFetcherIterator.scala:
##########
@@ -866,6 +930,7 @@ final class ShuffleBlockFetcherIterator(
}
if (blockId.isShuffleChunk) {
+ shuffleMetrics.incCorruptMergedBlockChunks(1)
Review Comment:
We need to add this for the invocation of `initiateFallbackFetchForPushMergedBlock` [above](https://github.com/apache/spark/blob/72cce5c39da8a52efa0a7cad4d65dac4eb389ff6/core/src/main/scala/org/apache/spark/storage/ShuffleBlockFetcherIterator.scala#L806) as well - the zero byte read is unexpected, and is essentially a corrupt read
##########
core/src/main/scala/org/apache/spark/storage/ShuffleBlockFetcherIterator.scala:
##########
@@ -726,6 +736,65 @@ final class ShuffleBlockFetcherIterator(
}
}
+ // Number of map blocks in a ShuffleBlockChunk
+ private def getShuffleChunkCardinality(blockId: ShuffleBlockChunkId): Int = {
+ val chunkTracker = pushBasedFetchHelper.getRoaringBitMap(blockId)
+ chunkTracker match {
+ case Some(bitmap) => bitmap.getCardinality
+ case None => 0
+ }
Review Comment:
```suggestion
pushBasedFetchHelper.getRoaringBitMap(blockId).map(_.getCardinality).getOrElse(0)
```
--
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
[GitHub] [spark] mridulm commented on a diff in pull request #36165: [SPARK-36620][SHUFFLE] Add Push Based Shuffle client side read metrics
Posted by GitBox <gi...@apache.org>.
mridulm commented on code in PR #36165:
URL: https://github.com/apache/spark/pull/36165#discussion_r1061138371
##########
core/src/main/protobuf/org/apache/spark/status/protobuf/store_types.proto:
##########
@@ -652,6 +699,20 @@ message ShuffleReadMetricDistributions {
repeated double remote_bytes_read = 6;
repeated double remote_bytes_read_to_disk = 7;
repeated double total_blocks_fetched = 8;
+ repeated double remote_reqs_duration = 9;
+ ShufflePushReadMetricDistributions shuffle_push_read_metrics_dist = 10;
+}
+
+message ShufflePushReadMetricDistributions {
+ repeated double corrupt_merged_block_chunks = 1;
+ repeated double merged_fetch_fallback_count = 2;
+ repeated double remote_merged_blocks_fetched = 3;
+ repeated double local_merged_blocks_fetched = 4;
+ repeated double remote_merged_chunks_fetched = 5;
+ repeated double local_merged_chunks_fetched = 6;
+ repeated double remote_merged_bytes_read = 7;
+ repeated double local_merged_bytes_read = 8;
+ repeated double remote_merged_reqs_duration = 9;
}
Review Comment:
+CC @gengliangwang
Can you please take a look specifically at the protobuf definitions and make sure they are aligned with your mental model of how this should be modeled ?
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: 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
[GitHub] [spark] mridulm commented on a diff in pull request #36165: [SPARK-36620][SHUFFLE] Add Push Based Shuffle client side read metrics
Posted by GitBox <gi...@apache.org>.
mridulm commented on code in PR #36165:
URL: https://github.com/apache/spark/pull/36165#discussion_r1062884659
##########
core/src/main/protobuf/org/apache/spark/status/protobuf/store_types.proto:
##########
@@ -652,6 +699,20 @@ message ShuffleReadMetricDistributions {
repeated double remote_bytes_read = 6;
repeated double remote_bytes_read_to_disk = 7;
repeated double total_blocks_fetched = 8;
+ repeated double remote_reqs_duration = 9;
+ ShufflePushReadMetricDistributions shuffle_push_read_metrics_dist = 10;
+}
+
+message ShufflePushReadMetricDistributions {
+ repeated double corrupt_merged_block_chunks = 1;
+ repeated double merged_fetch_fallback_count = 2;
+ repeated double remote_merged_blocks_fetched = 3;
+ repeated double local_merged_blocks_fetched = 4;
+ repeated double remote_merged_chunks_fetched = 5;
+ repeated double local_merged_chunks_fetched = 6;
+ repeated double remote_merged_bytes_read = 7;
+ repeated double local_merged_bytes_read = 8;
+ repeated double remote_merged_reqs_duration = 9;
}
Review Comment:
Thanks @gengliangwang !
--
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
[GitHub] [spark] thejdeep commented on pull request #36165: [SPARK-36620][SHUFFLE] Add Push Based Shuffle client side metrics
Posted by GitBox <gi...@apache.org>.
thejdeep commented on PR #36165:
URL: https://github.com/apache/spark/pull/36165#issuecomment-1208332949
@zhouyejoe @otterc Updated the PR with the latest changes, please take a look. Thank you!
--
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
[GitHub] [spark] otterc commented on a diff in pull request #36165: [SPARK-36620][SHUFFLE] Add Push Based Shuffle client side metrics
Posted by GitBox <gi...@apache.org>.
otterc commented on code in PR #36165:
URL: https://github.com/apache/spark/pull/36165#discussion_r1030746682
##########
core/src/main/scala/org/apache/spark/storage/ShuffleBlockFetcherIterator.scala:
##########
@@ -282,6 +280,17 @@ final class ShuffleBlockFetcherIterator(
}
}
+ @inline def updateMergedReqsDuration(wasReqForMergedChunks: Boolean = false): Unit = {
+ if (remainingBlocks.isEmpty) {
+ val durationMs = TimeUnit.NANOSECONDS.toMillis(clock.nanoTime() - requestStartTime)
+ if (wasReqForMergedChunks) {
+ shuffleMetrics.incRemoteMergedReqsDuration(durationMs)
+ } else {
+ shuffleMetrics.incRemoteReqsDuration(durationMs)
+ }
Review Comment:
We added these metrics internally so we can easily figure out what is taking longer- the fetch of the merge chunks or the few unmerged blocks. If we include the duration of merged chunks in the total then we have to derive the value for un-merged blocks which is fine. It makes sense to be consistent to avoid confusion.
--
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
[GitHub] [spark] mridulm commented on a diff in pull request #36165: [SPARK-36620][SHUFFLE] Add Push Based Shuffle client side metrics
Posted by GitBox <gi...@apache.org>.
mridulm commented on code in PR #36165:
URL: https://github.com/apache/spark/pull/36165#discussion_r1029927503
##########
core/src/main/scala/org/apache/spark/storage/ShuffleBlockFetcherIterator.scala:
##########
@@ -282,6 +280,17 @@ final class ShuffleBlockFetcherIterator(
}
}
+ @inline def updateMergedReqsDuration(wasReqForMergedChunks: Boolean = false): Unit = {
+ if (remainingBlocks.isEmpty) {
+ val durationMs = TimeUnit.NANOSECONDS.toMillis(clock.nanoTime() - requestStartTime)
+ if (wasReqForMergedChunks) {
+ shuffleMetrics.incRemoteMergedReqsDuration(durationMs)
+ } else {
+ shuffleMetrics.incRemoteReqsDuration(durationMs)
+ }
Review Comment:
`shuffleLocalMetricsUpdate` and `shuffleRemoteMetricsUpdate` is treating merged fetches as a subset of fetches for remote bytes - while here we are treating them separately.
Any particular reason for this difference ? (A throughput computation will need to take this into account due to this difference)
--
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
[GitHub] [spark] mridulm commented on a diff in pull request #36165: [SPARK-36620][SHUFFLE] Add Push Based Shuffle client side metrics
Posted by GitBox <gi...@apache.org>.
mridulm commented on code in PR #36165:
URL: https://github.com/apache/spark/pull/36165#discussion_r1008795619
##########
core/src/test/scala/org/apache/spark/ui/StagePageSuite.scala:
##########
@@ -78,6 +78,16 @@ class StagePageSuite extends SparkFunSuite with LocalSparkContext {
shuffleLocalBytesRead = 1L,
shuffleReadBytes = 1L,
shuffleReadRecords = 1L,
+ shuffleCorruptMergedBlockChunks = 2L,
+ shuffleFallbackCount = 2L,
+ shuffleMergedRemoteBlocksFetched = 1L,
+ shuffleMergedLocalBlocksFetched = 1L,
+ shuffleMergedRemoteChunksFetched = 1L,
+ shuffleMergedLocalChunksFetched = 1L,
+ shuffleMergedRemoteBytesRead = 1L,
+ shuffleMergedLocalBytesRead = 1L,
+ shuffleRemoteReqsDuration = 1L,
+ shuffleMergedRemoteReqsDuration = 1L,
Review Comment:
nit: given inputBytes > 0, no likely :-) But that is not relevant to this 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: 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
[GitHub] [spark] thejdeep commented on pull request #36165: [SPARK-36620][SHUFFLE] Add Push Based Shuffle client side read metrics
Posted by GitBox <gi...@apache.org>.
thejdeep commented on PR #36165:
URL: https://github.com/apache/spark/pull/36165#issuecomment-1363632144
@otterc @mridulm The test failure is from a rebased commit. I had to rebase locally and force push, this rewrote commit history.
--
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
[GitHub] [spark] mridulm commented on a diff in pull request #36165: [SPARK-36620][SHUFFLE] Add Push Based Shuffle client side read metrics
Posted by GitBox <gi...@apache.org>.
mridulm commented on code in PR #36165:
URL: https://github.com/apache/spark/pull/36165#discussion_r1063009392
##########
core/src/main/scala/org/apache/spark/status/protobuf/StageDataWrapperSerializer.scala:
##########
@@ -496,7 +555,52 @@ class StageDataWrapperSerializer extends ProtobufSerDe {
remoteBytesRead = binary.getRemoteBytesReadList.asScala.map(_.toDouble).toIndexedSeq,
remoteBytesReadToDisk =
binary.getRemoteBytesReadToDiskList.asScala.map(_.toDouble).toIndexedSeq,
- totalBlocksFetched = binary.getTotalBlocksFetchedList.asScala.map(_.toDouble).toIndexedSeq
+ totalBlocksFetched = binary.getTotalBlocksFetchedList.asScala.map(_.toDouble).toIndexedSeq,
+ remoteReqsDuration = binary.getRemoteReqsDurationList.asScala.map(_.toDouble).toIndexedSeq,
+ shufflePushReadMetricsDist =
+ deserializeShufflePushReadMetricsDistributions(binary.getShufflePushReadMetricsDist)
+ )
+ }
+
+ private def deserializeShufflePushReadMetricsDistributions(
+ binary: StoreTypes.ShuffleReadMetricDistributions): ShuffleReadMetricDistributions = {
Review Comment:
Remove this method ? It appears unused (and incorrectly named if it is required)
--
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
[GitHub] [spark] thejdeep commented on a diff in pull request #36165: [SPARK-36620][SHUFFLE] Add Push Based Shuffle client side read metrics
Posted by GitBox <gi...@apache.org>.
thejdeep commented on code in PR #36165:
URL: https://github.com/apache/spark/pull/36165#discussion_r1060720199
##########
core/src/main/protobuf/org/apache/spark/status/protobuf/store_types.proto:
##########
@@ -100,11 +100,21 @@ message TaskDataWrapper {
int64 shuffle_remote_bytes_read_to_disk = 34;
int64 shuffle_local_bytes_read = 35;
int64 shuffle_records_read = 36;
- int64 shuffle_bytes_written = 37;
- int64 shuffle_write_time = 38;
- int64 shuffle_records_written = 39;
- int64 stage_id = 40;
- int32 stage_attempt_id = 41;
+ int64 shuffle_corrupt_merged_block_chunks = 37;
+ int64 shuffle_merged_fetch_fallback_count = 38;
+ int64 shuffle_merged_remote_blocks_fetched = 39;
+ int64 shuffle_merged_local_blocks_fetched = 40;
+ int64 shuffle_merged_remote_chunks_fetched = 41;
+ int64 shuffle_merged_local_chunks_fetched = 42;
+ int64 shuffle_merged_remote_bytes_read = 43;
+ int64 shuffle_merged_local_bytes_read = 44;
+ int64 shuffle_remote_reqs_duration = 45;
+ int64 shuffle_merged_remote_req_duration = 46;
+ int64 shuffle_bytes_written = 47;
+ int64 shuffle_write_time = 48;
+ int64 shuffle_records_written = 49;
+ int64 stage_id = 50;
+ int32 stage_attempt_id = 51;
Review Comment:
@mridulm Thanks for reviewing, changed it to keep the existing protobuf ids
--
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
[GitHub] [spark] thejdeep commented on a diff in pull request #36165: [SPARK-36620][SHUFFLE] Add Push Based Shuffle client side read metrics
Posted by GitBox <gi...@apache.org>.
thejdeep commented on code in PR #36165:
URL: https://github.com/apache/spark/pull/36165#discussion_r1039691827
##########
core/src/main/scala/org/apache/spark/executor/Executor.scala:
##########
@@ -791,6 +770,53 @@ private[spark] class Executor(
}
}
+ private def incrementShuffleMetrics(
+ executorSource: ExecutorSource,
+ metrics: TaskMetrics
+ ): Unit = {
+ executorSource.METRIC_SHUFFLE_FETCH_WAIT_TIME
+ .inc(metrics.shuffleReadMetrics.fetchWaitTime)
+ executorSource.METRIC_SHUFFLE_WRITE_TIME.inc(metrics.shuffleWriteMetrics.writeTime)
+ executorSource.METRIC_SHUFFLE_TOTAL_BYTES_READ
+ .inc(metrics.shuffleReadMetrics.totalBytesRead)
+ executorSource.METRIC_SHUFFLE_REMOTE_BYTES_READ
+ .inc(metrics.shuffleReadMetrics.remoteBytesRead)
+ executorSource.METRIC_SHUFFLE_REMOTE_BYTES_READ_TO_DISK
+ .inc(metrics.shuffleReadMetrics.remoteBytesReadToDisk)
+ executorSource.METRIC_SHUFFLE_LOCAL_BYTES_READ
+ .inc(metrics.shuffleReadMetrics.localBytesRead)
+ executorSource.METRIC_SHUFFLE_RECORDS_READ
+ .inc(metrics.shuffleReadMetrics.recordsRead)
+ executorSource.METRIC_SHUFFLE_REMOTE_BLOCKS_FETCHED
+ .inc(metrics.shuffleReadMetrics.remoteBlocksFetched)
+ executorSource.METRIC_SHUFFLE_LOCAL_BLOCKS_FETCHED
+ .inc(metrics.shuffleReadMetrics.localBlocksFetched)
+ executorSource.METRIC_SHUFFLE_REMOTE_REQS_DURATION
+ .inc(metrics.shuffleReadMetrics.remoteReqsDuration)
+ executorSource.METRIC_SHUFFLE_BYTES_WRITTEN
+ .inc(metrics.shuffleWriteMetrics.bytesWritten)
+ executorSource.METRIC_SHUFFLE_RECORDS_WRITTEN
+ .inc(metrics.shuffleWriteMetrics.recordsWritten)
+ executorSource.METRIC_PUSH_BASED_SHUFFLE_CORRUPT_MERGED_BLOCK_CHUNKS
+ .inc(metrics.shuffleReadMetrics.corruptMergedBlockChunks)
+ executorSource.METRIC_PUSH_BASED_SHUFFLE_MERGED_FETCH_FALLBACK_COUNT
+ .inc(metrics.shuffleReadMetrics.fallbackCount)
Review Comment:
Thanks, updated to reflect this.
--
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
[GitHub] [spark] zhouyejoe commented on pull request #36165: [SPARK-36620][SHUFFLE] Add Push Based Shuffle client side metrics
Posted by GitBox <gi...@apache.org>.
zhouyejoe commented on PR #36165:
URL: https://github.com/apache/spark/pull/36165#issuecomment-1208631043
@thejdeep Thanks for updating the PR. Yet to review it, but several UTs within HistoryServerSuite failed, can you help double check?
--
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
[GitHub] [spark] thejdeep commented on a diff in pull request #36165: [SPARK-36620][SHUFFLE] Add Push Based Shuffle client side read metrics
Posted by GitBox <gi...@apache.org>.
thejdeep commented on code in PR #36165:
URL: https://github.com/apache/spark/pull/36165#discussion_r1035142645
##########
core/src/main/scala/org/apache/spark/InternalAccumulator.scala:
##########
@@ -54,6 +55,16 @@ 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 FALLBACK_COUNT = SHUFFLE_PUSH_READ_METRICS_PREFIX + "fallbackCount"
Review Comment:
renamed in all sources to `mergedFetchFallbackCount `
--
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
[GitHub] [spark] thejdeep commented on a diff in pull request #36165: [SPARK-36620][SHUFFLE] Add Push Based Shuffle client side read metrics
Posted by GitBox <gi...@apache.org>.
thejdeep commented on code in PR #36165:
URL: https://github.com/apache/spark/pull/36165#discussion_r1035141286
##########
core/src/main/scala/org/apache/spark/storage/ShuffleBlockFetcherIterator.scala:
##########
@@ -866,6 +930,7 @@ final class ShuffleBlockFetcherIterator(
}
if (blockId.isShuffleChunk) {
+ shuffleMetrics.incCorruptMergedBlockChunks(1)
Review Comment:
Thanks for calling this out, added 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: 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
[GitHub] [spark] otterc commented on pull request #36165: [SPARK-36620][SHUFFLE] Add Push Based Shuffle client side read metrics
Posted by GitBox <gi...@apache.org>.
otterc commented on PR #36165:
URL: https://github.com/apache/spark/pull/36165#issuecomment-1363226694
@thejdeep The test failures are related to this change. Please fix them.
```
/home/runner/work/spark/spark/core/src/main/scala/org/apache/spark/status/protobuf/TaskDataWrapperSerializer.scala:92:5: not enough arguments for constructor TaskDataWrapper: (taskId: Long, index: Int, attempt: Int, partitionId: Int, launchTime: Long, resultFetchStart: Long, duration: Long, executorId: String, host: String, status: String, taskLocality: String, speculative: Boolean, accumulatorUpdates: Seq[org.apache.spark.status.api.v1.AccumulableInfo], errorMessage: Option[String], hasMetrics: Boolean, executorDeserializeTime: Long, executorDeserializeCpuTime: Long, executorRunTime: Long, executorCpuTime: Long, resultSize: Long, jvmGcTime: Long, resultSerializationTime: Long, memoryBytesSpilled: Long, diskBytesSpilled: Long, peakExecutionMemory: Long, inputBytesRead: Long, inputRecordsRead: Long, outputBytesWritten: Long, outputRecordsWritten: Long, shuffleRemoteBlocksFetched: Long, shuffleLocalBlocksFetched: Long, shuffleFetchWaitTime: Long, shuffleRemoteBytesRead: Long, shuffl
eRemoteBytesReadToDisk: Long, shuffleLocalBytesRead: Long, shuffleRecordsRead: Long, shuffleCorruptMergedBlockChunks: Long, shuffleMergedFetchFallbackCount: Long, shuffleMergedRemoteBlocksFetched: Long, shuffleMergedLocalBlocksFetched: Long, shuffleMergedRemoteChunksFetched: Long, shuffleMergedLocalChunksFetched: Long, shuffleMergedRemoteBytesRead: Long, shuffleMergedLocalBytesRead: Long, shuffleRemoteReqsDuration: Long, shuffleMergedRemoteReqDuration: Long, shuffleBytesWritten: Long, shuffleWriteTime: Long, shuffleRecordsWritten: Long, stageId: Int, stageAttemptId: Int)org.apache.spark.status.TaskDataWrapper.
[error] Unspecified value parameters shuffleCorruptMergedBlockChunks, shuffleMergedFetchFallbackCount, shuffleMergedRemoteBlocksFetched...
[error] new TaskDataWrapper(
[error] ^
```
--
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
[GitHub] [spark] thejdeep commented on pull request #36165: [SPARK-36620][SHUFFLE] Add Push Based Shuffle client side read metrics
Posted by GitBox <gi...@apache.org>.
thejdeep commented on PR #36165:
URL: https://github.com/apache/spark/pull/36165#issuecomment-1361297035
Addressed comments, please take another look @otterc @mridulm @zhouyejoe
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: 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
[GitHub] [spark] thejdeep commented on a diff in pull request #36165: [SPARK-36620][SHUFFLE] Add Push Based Shuffle client side read metrics
Posted by GitBox <gi...@apache.org>.
thejdeep commented on code in PR #36165:
URL: https://github.com/apache/spark/pull/36165#discussion_r1046139711
##########
core/src/main/scala/org/apache/spark/storage/ShuffleBlockFetcherIterator.scala:
##########
@@ -726,6 +736,61 @@ final class ShuffleBlockFetcherIterator(
}
}
+ // Number of map blocks in a ShuffleBlockChunk
+ private def getShuffleChunkCardinality(blockId: ShuffleBlockChunkId): Int = {
+ pushBasedFetchHelper.getRoaringBitMap(blockId).map(_.getCardinality).getOrElse(0)
+ }
+
+ // Check if the merged block is local to the host or not
+ private def isLocalMergedBlockAddress(address: BlockManagerId): Boolean = {
+ address.executorId.isEmpty && address.host == blockManager.blockManagerId.host
Review Comment:
Reused method from `PushBasedFetchHelper`. There was a class instance available for use, so I used that instead of making the function static.
--
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
[GitHub] [spark] otterc commented on a diff in pull request #36165: [SPARK-36620][SHUFFLE] Add Push Based Shuffle client side read metrics
Posted by GitBox <gi...@apache.org>.
otterc commented on code in PR #36165:
URL: https://github.com/apache/spark/pull/36165#discussion_r1046299412
##########
core/src/main/scala/org/apache/spark/status/LiveEntity.scala:
##########
@@ -843,6 +917,25 @@ private[spark] object LiveEntityHelpers {
m2.shuffleReadMetrics.remoteBytesReadToDisk * mult,
m1.shuffleReadMetrics.localBytesRead + m2.shuffleReadMetrics.localBytesRead * mult,
m1.shuffleReadMetrics.recordsRead + m2.shuffleReadMetrics.recordsRead * mult,
+ m1.shuffleReadMetrics.shufflePushReadMetrics.corruptMergedBlockChunks +
+ m2.shuffleReadMetrics.shufflePushReadMetrics.corruptMergedBlockChunks * mult,
+ m1.shuffleReadMetrics.shufflePushReadMetrics.mergedFetchFallbackCount +
+ m2.shuffleReadMetrics.shufflePushReadMetrics.mergedFetchFallbackCount * mult,
+ m1.shuffleReadMetrics.shufflePushReadMetrics.remoteMergedBlocksFetched +
+ m2.shuffleReadMetrics.shufflePushReadMetrics.remoteMergedBlocksFetched * mult,
+ m1.shuffleReadMetrics.shufflePushReadMetrics.localMergedBlocksFetched +
+ m2.shuffleReadMetrics.shufflePushReadMetrics.localMergedBlocksFetched * mult,
+ m1.shuffleReadMetrics.shufflePushReadMetrics.remoteMergedChunksFetched +
+ m2.shuffleReadMetrics.shufflePushReadMetrics.remoteMergedChunksFetched * mult,
+ m1.shuffleReadMetrics.shufflePushReadMetrics.localMergedChunksFetched +
+ m2.shuffleReadMetrics.shufflePushReadMetrics.localMergedChunksFetched * mult,
+ m1.shuffleReadMetrics.shufflePushReadMetrics.remoteMergedBytesRead +
+ m2.shuffleReadMetrics.shufflePushReadMetrics.remoteMergedBytesRead * mult,
+ m1.shuffleReadMetrics.shufflePushReadMetrics.localMergedBytesRead +
+ m2.shuffleReadMetrics.shufflePushReadMetrics.localMergedBytesRead * mult,
+ m1.shuffleReadMetrics.remoteReqsDuration + m2.shuffleReadMetrics.remoteReqsDuration * mult,
+ m1.shuffleReadMetrics.shufflePushReadMetrics.remoteMergedReqsDuration +
+ m2.shuffleReadMetrics.shufflePushReadMetrics.remoteMergedChunksFetched * mult,
Review Comment:
This should be `remoteMergedReqsDuration` instead of `remoteMergedChunksFetched`
--
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
[GitHub] [spark] mridulm commented on pull request #36165: [SPARK-36620][SHUFFLE] Add Push Based Shuffle client side read metrics
Posted by GitBox <gi...@apache.org>.
mridulm commented on PR #36165:
URL: https://github.com/apache/spark/pull/36165#issuecomment-1370491779
Can you take a look at the test failures @thejdeep ? They look relevant.
--
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
[GitHub] [spark] thejdeep commented on a diff in pull request #36165: [SPARK-36620][SHUFFLE] Add Push Based Shuffle client side metrics
Posted by GitBox <gi...@apache.org>.
thejdeep commented on code in PR #36165:
URL: https://github.com/apache/spark/pull/36165#discussion_r1008470301
##########
core/src/main/scala/org/apache/spark/status/storeTypes.scala:
##########
@@ -233,6 +243,38 @@ private[spark] class TaskDataWrapper(
val shuffleLocalBytesRead: Long,
@KVIndexParam(value = TaskIndexNames.SHUFFLE_READ_RECORDS, parent = TaskIndexNames.STAGE)
val shuffleRecordsRead: Long,
+ @KVIndexParam(
+ value = TaskIndexNames.PUSH_BASED_SHUFFLE_CORRUPT_MERGED_BLOCK_CHUNKS,
+ parent = TaskIndexNames.STAGE)
+ val shuffleCorruptMergedBlockChunks: Long,
+ @KVIndexParam(value = TaskIndexNames.PUSH_BASED_SHUFFLE_FALLBACK_COUNT,
+ parent = TaskIndexNames.STAGE)
+ val shuffleFallbackCount: Long,
+ @KVIndexParam(
+ value = TaskIndexNames.PUSH_BASED_SHUFFLE_MERGED_REMOTE_BLOCKS, parent = TaskIndexNames.STAGE)
+ val shuffleMergedRemoteBlocksFetched: Long,
+ @KVIndexParam(
+ value = TaskIndexNames.PUSH_BASED_SHUFFLE_MERGED_LOCAL_BLOCKS, parent = TaskIndexNames.STAGE)
+ val shuffleMergedLocalBlocksFetched: Long,
+ @KVIndexParam(
+ value = TaskIndexNames.PUSH_BASED_SHUFFLE_MERGED_REMOTE_CHUNKS, parent = TaskIndexNames.STAGE)
+ val shuffleMergedRemoteChunksFetched: Long,
+ @KVIndexParam(
+ value = TaskIndexNames.PUSH_BASED_SHUFFLE_MERGED_LOCAL_CHUNKS, parent = TaskIndexNames.STAGE)
+ val shuffleMergedLocalChunksFetched: Long,
+ @KVIndexParam(
+ value = TaskIndexNames.PUSH_BASED_SHUFFLE_MERGED_REMOTE_READS, parent = TaskIndexNames.STAGE)
+ val shuffleMergedRemoteBlocksBytesRead: Long,
+ @KVIndexParam(
+ value = TaskIndexNames.PUSH_BASED_SHUFFLE_MERGED_LOCAL_READS, parent = TaskIndexNames.STAGE)
+ val shuffleMergedLocalBlocksBytesRead: Long,
+ @KVIndexParam(
+ value = TaskIndexNames.PUSH_BASED_SHUFFLE_REMOTE_REQS_DURATION, parent = TaskIndexNames.STAGE)
+ val shuffleRemoteReqsDuration: Long,
+ @KVIndexParam(
+ value = TaskIndexNames.PUSH_BASED_SHUFFLE_MERGED_REMOTE_REQS_DURATION,
+ parent = TaskIndexNames.STAGE)
Review Comment:
Indexes for these fields is a requirement when we are computing quantiles and reading `TaskDataWrapper`s.
--
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
[GitHub] [spark] thejdeep commented on a diff in pull request #36165: [SPARK-36620][SHUFFLE] Add Push Based Shuffle client side read metrics
Posted by GitBox <gi...@apache.org>.
thejdeep commented on code in PR #36165:
URL: https://github.com/apache/spark/pull/36165#discussion_r1046140712
##########
core/src/test/resources/HistoryServerExpectations/failed_stage_list_json_expectation.json:
##########
@@ -27,13 +27,23 @@
"outputBytes" : 0,
"outputRecords" : 0,
"shuffleRemoteBlocksFetched" : 0,
- "shuffleLocalBlocksFetched" : 64,
- "shuffleFetchWaitTime" : 1,
+ "shuffleLocalBlocksFetched" : 0,
+ "shuffleFetchWaitTime" : 0,
Review Comment:
There was a logical bug in merging the read metrics, thanks for catching this! Updated the expectation JSON.
--
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
[GitHub] [spark] mridulm commented on a diff in pull request #36165: [SPARK-36620][SHUFFLE] Add Push Based Shuffle client side read metrics
Posted by GitBox <gi...@apache.org>.
mridulm commented on code in PR #36165:
URL: https://github.com/apache/spark/pull/36165#discussion_r1057888476
##########
core/src/main/protobuf/org/apache/spark/status/protobuf/store_types.proto:
##########
@@ -100,11 +100,21 @@ message TaskDataWrapper {
int64 shuffle_remote_bytes_read_to_disk = 34;
int64 shuffle_local_bytes_read = 35;
int64 shuffle_records_read = 36;
- int64 shuffle_bytes_written = 37;
- int64 shuffle_write_time = 38;
- int64 shuffle_records_written = 39;
- int64 stage_id = 40;
- int32 stage_attempt_id = 41;
+ int64 shuffle_corrupt_merged_block_chunks = 37;
+ int64 shuffle_merged_fetch_fallback_count = 38;
+ int64 shuffle_merged_remote_blocks_fetched = 39;
+ int64 shuffle_merged_local_blocks_fetched = 40;
+ int64 shuffle_merged_remote_chunks_fetched = 41;
+ int64 shuffle_merged_local_chunks_fetched = 42;
+ int64 shuffle_merged_remote_bytes_read = 43;
+ int64 shuffle_merged_local_bytes_read = 44;
+ int64 shuffle_remote_reqs_duration = 45;
+ int64 shuffle_merged_remote_req_duration = 46;
+ int64 shuffle_bytes_written = 47;
+ int64 shuffle_write_time = 48;
+ int64 shuffle_records_written = 49;
+ int64 stage_id = 50;
+ int32 stage_attempt_id = 51;
Review Comment:
We should not change attribute `id`'s (here and elsewhere)
In this specific case, given this is not yet released, it is not as bad - but it might still interfere with some ongoing dev/test work.
--
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
[GitHub] [spark] gengliangwang commented on a diff in pull request #36165: [SPARK-36620][SHUFFLE] Add Push Based Shuffle client side read metrics
Posted by GitBox <gi...@apache.org>.
gengliangwang commented on code in PR #36165:
URL: https://github.com/apache/spark/pull/36165#discussion_r1061199799
##########
core/src/main/protobuf/org/apache/spark/status/protobuf/store_types.proto:
##########
@@ -652,6 +699,20 @@ message ShuffleReadMetricDistributions {
repeated double remote_bytes_read = 6;
repeated double remote_bytes_read_to_disk = 7;
repeated double total_blocks_fetched = 8;
+ repeated double remote_reqs_duration = 9;
+ ShufflePushReadMetricDistributions shuffle_push_read_metrics_dist = 10;
+}
+
+message ShufflePushReadMetricDistributions {
+ repeated double corrupt_merged_block_chunks = 1;
+ repeated double merged_fetch_fallback_count = 2;
+ repeated double remote_merged_blocks_fetched = 3;
+ repeated double local_merged_blocks_fetched = 4;
+ repeated double remote_merged_chunks_fetched = 5;
+ repeated double local_merged_chunks_fetched = 6;
+ repeated double remote_merged_bytes_read = 7;
+ repeated double local_merged_bytes_read = 8;
+ repeated double remote_merged_reqs_duration = 9;
}
Review Comment:
Yes, the protobuf definitions LGTM
--
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
[GitHub] [spark] mridulm commented on a diff in pull request #36165: [SPARK-36620][SHUFFLE] Add Push Based Shuffle client side read metrics
Posted by GitBox <gi...@apache.org>.
mridulm commented on code in PR #36165:
URL: https://github.com/apache/spark/pull/36165#discussion_r1034209167
##########
core/src/main/scala/org/apache/spark/storage/ShuffleBlockFetcherIterator.scala:
##########
@@ -722,6 +718,63 @@ final class ShuffleBlockFetcherIterator(
}
}
+ // Number of map blocks in a ShuffleBlockChunk
+ private def getShuffleChunkCardinality(blockId: ShuffleBlockChunkId): Int = {
+ val chunkTracker = pushBasedFetchHelper.getRoaringBitMap(blockId)
+ chunkTracker match {
+ case Some(bitmap) => bitmap.getCardinality
+ case None => 0
+ }
+ }
+
+ // Check if the merged block is local to the host or not
+ private def isLocalMergedBlockAddress(address: BlockManagerId): Boolean = {
+ address.executorId.isEmpty && address.host == blockManager.blockManagerId.host
+ }
+
+ private def shuffleMetricsUpdate(
+ blockId: BlockId,
+ buf: ManagedBuffer,
+ isLocal: Boolean): Unit = {
+ if (isLocal) {
+ shuffleLocalMetricsUpdate(blockId, buf)
+ } else {
+ shuffleRemoteMetricsUpdate(blockId, buf)
+ }
+ }
+
+ private def shuffleLocalMetricsUpdate(blockId: BlockId, buf: ManagedBuffer): Unit = {
+ // Check if the block is within the host-local blocks set, or if it is a merged local
+ // block. In either case, it is local read and we need to increase the local
+ // shuffle read metrics.
+ blockId match {
+ case chunkId: ShuffleBlockChunkId =>
+ shuffleMetrics.incLocalMergedChunksFetched(1)
+ shuffleMetrics.incLocalMergedBlocksFetched(getShuffleChunkCardinality(chunkId))
+ shuffleMetrics.incLocalMergedBlocksBytesRead(buf.size)
+ shuffleMetrics.incLocalBlocksFetched(getShuffleChunkCardinality(chunkId))
Review Comment:
Take a look at the snippet above - it is used twice :-)
Same in `shuffleRemoteMetricsUpdate`
--
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
[GitHub] [spark] zhouyejoe commented on pull request #36165: [SPARK-36620][SHUFFLE] Add Push Based Shuffle client side metrics
Posted by GitBox <gi...@apache.org>.
zhouyejoe commented on PR #36165:
URL: https://github.com/apache/spark/pull/36165#issuecomment-1329460191
Left some minor comments last week, others mostly LGTM. @thejdeep
--
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
[GitHub] [spark] mridulm closed pull request #36165: [SPARK-36620][SHUFFLE] Add Push Based Shuffle client side read metrics
Posted by GitBox <gi...@apache.org>.
mridulm closed pull request #36165: [SPARK-36620][SHUFFLE] Add Push Based Shuffle client side read metrics
URL: https://github.com/apache/spark/pull/36165
--
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
[GitHub] [spark] mridulm commented on pull request #36165: [SPARK-36620][SHUFFLE] Add Push Based Shuffle client side read metrics
Posted by GitBox <gi...@apache.org>.
mridulm commented on PR #36165:
URL: https://github.com/apache/spark/pull/36165#issuecomment-1372970457
I am not sure what has changed since last review due to the force pushes - please do not force push, particularly for large PR's.
--
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
[GitHub] [spark] thejdeep commented on a diff in pull request #36165: [SPARK-36620][SHUFFLE] Add Push Based Shuffle client side read metrics
Posted by GitBox <gi...@apache.org>.
thejdeep commented on code in PR #36165:
URL: https://github.com/apache/spark/pull/36165#discussion_r1063026190
##########
core/src/main/scala/org/apache/spark/status/protobuf/StageDataWrapperSerializer.scala:
##########
@@ -496,7 +555,52 @@ class StageDataWrapperSerializer extends ProtobufSerDe {
remoteBytesRead = binary.getRemoteBytesReadList.asScala.map(_.toDouble).toIndexedSeq,
remoteBytesReadToDisk =
binary.getRemoteBytesReadToDiskList.asScala.map(_.toDouble).toIndexedSeq,
- totalBlocksFetched = binary.getTotalBlocksFetchedList.asScala.map(_.toDouble).toIndexedSeq
+ totalBlocksFetched = binary.getTotalBlocksFetchedList.asScala.map(_.toDouble).toIndexedSeq,
+ remoteReqsDuration = binary.getRemoteReqsDurationList.asScala.map(_.toDouble).toIndexedSeq,
+ shufflePushReadMetricsDist =
+ deserializeShufflePushReadMetricsDistributions(binary.getShufflePushReadMetricsDist)
+ )
+ }
+
+ private def deserializeShufflePushReadMetricsDistributions(
+ binary: StoreTypes.ShuffleReadMetricDistributions): ShuffleReadMetricDistributions = {
Review Comment:
Removed the duplicate method, 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: 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
[GitHub] [spark] thejdeep commented on a diff in pull request #36165: [SPARK-36620][SHUFFLE] Add Push Based Shuffle client side read metrics
Posted by GitBox <gi...@apache.org>.
thejdeep commented on code in PR #36165:
URL: https://github.com/apache/spark/pull/36165#discussion_r1046138205
##########
core/src/main/scala/org/apache/spark/status/api/v1/api.scala:
##########
@@ -302,7 +312,9 @@ class StageData private[spark](
@JsonDeserialize(using = classOf[ExecutorMetricsJsonDeserializer])
val peakExecutorMetrics: Option[ExecutorMetrics],
val taskMetricsDistributions: Option[TaskMetricDistributions],
- val executorMetricsDistributions: Option[ExecutorMetricsDistributions])
+ val executorMetricsDistributions: Option[ExecutorMetricsDistributions],
+ val isPushBasedShuffleEnabled: Boolean,
Review Comment:
Removed `Based` from user facing 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: 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
[GitHub] [spark] thejdeep commented on a diff in pull request #36165: [SPARK-36620][SHUFFLE] Add Push Based Shuffle client side read metrics
Posted by GitBox <gi...@apache.org>.
thejdeep commented on code in PR #36165:
URL: https://github.com/apache/spark/pull/36165#discussion_r1040478699
##########
core/src/test/resources/HistoryServerExpectations/excludeOnFailure_node_for_stage_expectation.json:
##########
@@ -81,7 +93,19 @@
"remoteBytesRead" : 0,
"remoteBytesReadToDisk" : 0,
"localBytesRead" : 0,
- "recordsRead" : 0
+ "recordsRead" : 0,
+ "pushBased": {
Review Comment:
This is the API response that serialized based on the [case classes in api.scala here](https://github.com/apache/spark/blob/7259a8c0147d9da70848ef2e7499a1712a4c4741/core/src/main/scala/org/apache/spark/status/api/v1/api.scala#L384). We only write "Push Based Shuffle" when logging to the event file
--
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
[GitHub] [spark] thejdeep commented on pull request #36165: [SPARK-36620][SHUFFLE] Add Push Based Shuffle client side metrics
Posted by GitBox <gi...@apache.org>.
thejdeep commented on PR #36165:
URL: https://github.com/apache/spark/pull/36165#issuecomment-1238489304
@otterc @zhouyejoe Fixed the failing tests, should be good for another review now. 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: 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
[GitHub] [spark] thejdeep commented on a diff in pull request #36165: [SPARK-36620][SHUFFLE] Add Push Based Shuffle client side read metrics
Posted by GitBox <gi...@apache.org>.
thejdeep commented on code in PR #36165:
URL: https://github.com/apache/spark/pull/36165#discussion_r1040884126
##########
core/src/test/resources/HistoryServerExpectations/excludeOnFailure_node_for_stage_expectation.json:
##########
@@ -81,7 +93,19 @@
"remoteBytesRead" : 0,
"remoteBytesReadToDisk" : 0,
"localBytesRead" : 0,
- "recordsRead" : 0
+ "recordsRead" : 0,
+ "pushBased": {
Review Comment:
Do you recommend changing this to "pushBasedShuffle" ? to note: it is inside the context of shuffle read 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: 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
[GitHub] [spark] thejdeep commented on pull request #36165: [SPARK-36620][SHUFFLE] Add Push Based Shuffle client side read metrics
Posted by GitBox <gi...@apache.org>.
thejdeep commented on PR #36165:
URL: https://github.com/apache/spark/pull/36165#issuecomment-1371213103
Fixed failing tests and updated commit messages to reflect the overall changes. 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: 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
[GitHub] [spark] thejdeep commented on pull request #36165: [SPARK-36620][SHUFFLE] Add Push Based Shuffle client side read metrics
Posted by GitBox <gi...@apache.org>.
thejdeep commented on PR #36165:
URL: https://github.com/apache/spark/pull/36165#issuecomment-1370219026
@mridulm `StageDataWrapper` was also added to protobuf serialization in #39192 (merged a few days ago) causing this build to fail. I have updated this PR to make more changes to the protobuf store types and also added serialization for push shuffle metrics. The build should be passing now. 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: 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
[GitHub] [spark] mridulm commented on pull request #36165: [SPARK-36620][SHUFFLE] Add Push Based Shuffle client side metrics
Posted by GitBox <gi...@apache.org>.
mridulm commented on PR #36165:
URL: https://github.com/apache/spark/pull/36165#issuecomment-1299640509
+CC @zhouyejoe
--
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
[GitHub] [spark] zhouyejoe commented on a diff in pull request #36165: [SPARK-36620][SHUFFLE] Add Push Based Shuffle client side metrics
Posted by GitBox <gi...@apache.org>.
zhouyejoe commented on code in PR #36165:
URL: https://github.com/apache/spark/pull/36165#discussion_r1012326931
##########
core/src/main/scala/org/apache/spark/executor/TaskMetrics.scala:
##########
@@ -227,6 +227,16 @@ 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.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_BYTES_READ -> shuffleReadMetrics._remoteMergedBlocksBytesRead,
+ shuffleRead.LOCAL_MERGED_BYTES_READ -> shuffleReadMetrics._localMergedBlocksBytesRead,
+ shuffleRead.REMOTE_REQS_DURATION -> shuffleReadMetrics._remoteReqsDuration,
+ shuffleRead.REMOTE_MERGED_REQS_DURATION -> shuffleReadMetrics._remoteMergedReqsDuration,
shuffleWrite.BYTES_WRITTEN -> shuffleWriteMetrics._bytesWritten,
shuffleWrite.RECORDS_WRITTEN -> shuffleWriteMetrics._recordsWritten,
shuffleWrite.WRITE_TIME -> shuffleWriteMetrics._writeTime,
Review Comment:
Chatted offline. Ignore my comments on this one since it is planned to do it in a separate PR.
--
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
[GitHub] [spark] zhouyejoe commented on a diff in pull request #36165: [SPARK-36620][SHUFFLE] Add Push Based Shuffle client side metrics
Posted by GitBox <gi...@apache.org>.
zhouyejoe commented on code in PR #36165:
URL: https://github.com/apache/spark/pull/36165#discussion_r1030723363
##########
core/src/main/scala/org/apache/spark/storage/ShuffleBlockFetcherIterator.scala:
##########
@@ -282,6 +280,17 @@ final class ShuffleBlockFetcherIterator(
}
}
+ @inline def updateMergedReqsDuration(wasReqForMergedChunks: Boolean = false): Unit = {
+ if (remainingBlocks.isEmpty) {
+ val durationMs = TimeUnit.NANOSECONDS.toMillis(clock.nanoTime() - requestStartTime)
+ if (wasReqForMergedChunks) {
+ shuffleMetrics.incRemoteMergedReqsDuration(durationMs)
+ } else {
+ shuffleMetrics.incRemoteReqsDuration(durationMs)
+ }
Review Comment:
+1 on this point, the merged remote request duration should be counted into the total remote request duration, following the same logic for the local/remote bytes read.
@otterc WDTY?
--
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
[GitHub] [spark] zhouyejoe commented on a diff in pull request #36165: [SPARK-36620][SHUFFLE] Add Push Based Shuffle client side metrics
Posted by GitBox <gi...@apache.org>.
zhouyejoe commented on code in PR #36165:
URL: https://github.com/apache/spark/pull/36165#discussion_r1030959574
##########
core/src/main/scala/org/apache/spark/util/JsonProtocol.scala:
##########
@@ -407,6 +407,8 @@ private[spark] object JsonProtocol {
g.writeFieldName("Accumulables")
accumulablesToJson(stageInfo.accumulables.values, g)
g.writeNumberField("Resource Profile Id", stageInfo.resourceProfileId)
+ g.writeBooleanField("Push based shuffle enabled", stageInfo.isPushBasedShuffleEnabled)
+ g.writeNumberField("Shuffle push mergers count", stageInfo.shuffleMergerCount)
Review Comment:
Nit: Capitalization for the first char of each word, following the same style as others shown above
##########
project/MimaExcludes.scala:
##########
@@ -87,6 +87,11 @@ object MimaExcludes {
// [SPARK-36511][MINOR][SQL] Remove ColumnIOUtil
ProblemFilters.exclude[MissingClassProblem]("org.apache.parquet.io.ColumnIOUtil"),
+ // [SPARK-36620] [SHUFFLE] Expose push based shuffle metrics
+ ProblemFilters.exclude[DirectMissingMethodProblem]("org.apache.spark.status.api.v1.ShuffleReadMetricDistributions.this"),
+ ProblemFilters.exclude[DirectMissingMethodProblem]("org.apache.spark.status.api.v1.ShuffleReadMetrics.this"),
+ ProblemFilters.exclude[MissingMethodProblem]("org.apache.spark.status.api.v1.StageData.this"),
+
// [SPARK-40324][SQL] Provide query context in AnalysisException
Review Comment:
Just curious: What specific changes lead to this exclusion? Can we avoid this exclusion?
##########
core/src/main/scala/org/apache/spark/storage/ShuffleBlockFetcherIterator.scala:
##########
@@ -722,6 +718,63 @@ final class ShuffleBlockFetcherIterator(
}
}
+ // Number of map blocks in a ShuffleBlockChunk
+ private def getShuffleChunkCardinality(blockId: ShuffleBlockChunkId): Int = {
+ val chunkTracker = pushBasedFetchHelper.getRoaringBitMap(blockId)
+ chunkTracker match {
+ case Some(bitmap) => bitmap.getCardinality
+ case None => 0
+ }
+ }
+
+ // Check if the merged block is local to the host or not
+ private def isLocalMergedBlockAddress(address: BlockManagerId): Boolean = {
+ address.executorId.isEmpty && address.host == blockManager.blockManagerId.host
+ }
+
+ private def shuffleMetricsUpdate(
+ blockId: BlockId,
+ buf: ManagedBuffer,
+ isLocal: Boolean): Unit = {
+ if (isLocal) {
+ shuffleLocalMetricsUpdate(blockId, buf)
+ } else {
+ shuffleRemoteMetricsUpdate(blockId, buf)
+ }
+ }
+
+ private def shuffleLocalMetricsUpdate(blockId: BlockId, buf: ManagedBuffer): Unit = {
+ // Check if the block is within the host-local blocks set, or if it is a merged local
+ // block. In either case, it is local read and we need to increase the local
+ // shuffle read metrics.
+ blockId match {
+ case chunkId: ShuffleBlockChunkId =>
+ shuffleMetrics.incLocalMergedChunksFetched(1)
+ shuffleMetrics.incLocalMergedBlocksFetched(getShuffleChunkCardinality(chunkId))
+ shuffleMetrics.incLocalMergedBlocksBytesRead(buf.size)
+ shuffleMetrics.incLocalBlocksFetched(getShuffleChunkCardinality(chunkId))
Review Comment:
getShuffleChunkCardinality will only get called once for each blockId in either shuffleLocalMetricsUpdate or shuffleRemoteMetricsUpdate, I think it won't get reused.
##########
core/src/main/scala/org/apache/spark/util/JsonProtocol.scala:
##########
@@ -1105,11 +1132,17 @@ private[spark] object JsonProtocol {
case None => Seq.empty[AccumulableInfo]
}
}
+ val isPushBasedShuffleEnabled =
+ jsonOption(json.get("Push based shuffle enabled")).map(_.extractBoolean).getOrElse(false)
+ val shufflePushMergersCount =
Review Comment:
Nit: Same here. Capitalization
##########
core/src/test/resources/HistoryServerExpectations/one_stage_json_with_details_expectation.json:
##########
@@ -32,7 +32,19 @@
"shuffleRemoteBytesReadToDisk" : 0,
"shuffleLocalBytesRead" : 0,
"shuffleReadBytes" : 0,
+ "shuffleCorruptMergedBlockChunks" : 0,
+ "shuffleFallbackCount" : 0,
+ "shuffleMergedRemoteBlocksFetched" : 0,
+ "shuffleMergedLocalBlocksFetched" : 0,
+ "shuffleMergedRemoteChunksFetched" : 0,
+ "shuffleMergedLocalChunksFetched" : 0,
+ "shuffleMergedRemoteBytesRead" : 0,
+ "shuffleMergedLocalBytesRead" : 0,
+ "shuffleRemoteReqsDuration" : 0,
+ "shuffleMergedRemoteReqsDuration" : 0,
"shuffleReadRecords" : 0,
+ "isPushBasedShuffleEnabled" : false,
Review Comment:
Nit: Line 45 should be moved to line 35.
##########
core/src/main/scala/org/apache/spark/util/JsonProtocol.scala:
##########
@@ -1226,6 +1259,29 @@ private[spark] object JsonProtocol {
readMetrics.incRecordsRead(
jsonOption(readJson.get("Total Records Read")).map(_.extractLong).getOrElse(0L))
metrics.mergeShuffleReadMetrics()
+ readMetrics.incRemoteReqsDuration(jsonOption(readJson.get("Remote Requests Duration"))
+ .map(_.extractLong).getOrElse(0L))
+ jsonOption(readJson.get("Push Based")).foreach { pushBasedShuffleReadJson =>
+ readMetrics.incCorruptMergedBlockChunks(jsonOption(
+ pushBasedShuffleReadJson.get("Corrupt Merged Block Chunks"))
+ .map(_.extractLong).getOrElse(0L))
Review Comment:
Nit: Looks like indent is off
##########
core/src/main/scala/org/apache/spark/storage/ShuffleBlockFetcherIterator.scala:
##########
@@ -726,6 +736,63 @@ final class ShuffleBlockFetcherIterator(
}
}
+ // Number of map blocks in a ShuffleBlockChunk
+ private def getShuffleChunkCardinality(blockId: ShuffleBlockChunkId): Int = {
+ val chunkTracker = pushBasedFetchHelper.getRoaringBitMap(blockId)
+ chunkTracker match {
+ case Some(bitmap) => bitmap.getCardinality
+ case None => 0
+ }
+ }
+
+ // Check if the merged block is local to the host or not
+ private def isLocalMergedBlockAddress(address: BlockManagerId): Boolean = {
+ address.executorId.isEmpty && address.host == blockManager.blockManagerId.host
+ }
+
+ private def shuffleMetricsUpdate(
+ blockId: BlockId,
+ buf: ManagedBuffer,
+ local: Boolean): Unit = {
Review Comment:
Nit: Indent is 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
[GitHub] [spark] otterc commented on a diff in pull request #36165: [SPARK-36620][SHUFFLE] Add Push Based Shuffle client side metrics
Posted by GitBox <gi...@apache.org>.
otterc commented on code in PR #36165:
URL: https://github.com/apache/spark/pull/36165#discussion_r1008462544
##########
core/src/test/scala/org/apache/spark/ui/StagePageSuite.scala:
##########
@@ -78,6 +78,16 @@ class StagePageSuite extends SparkFunSuite with LocalSparkContext {
shuffleLocalBytesRead = 1L,
shuffleReadBytes = 1L,
shuffleReadRecords = 1L,
+ shuffleCorruptMergedBlockChunks = 2L,
+ shuffleFallbackCount = 2L,
+ shuffleMergedRemoteBlocksFetched = 1L,
+ shuffleMergedLocalBlocksFetched = 1L,
+ shuffleMergedRemoteChunksFetched = 1L,
+ shuffleMergedLocalChunksFetched = 1L,
+ shuffleMergedRemoteBytesRead = 1L,
+ shuffleMergedLocalBytesRead = 1L,
+ shuffleRemoteReqsDuration = 1L,
+ shuffleMergedRemoteReqsDuration = 1L,
Review Comment:
We didn't put much thought into it. But `shuffleMergerCount` can be 0 because this stage may not push but this stage can still read shuffle data generated by the previous stage.
--
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
[GitHub] [spark] mridulm commented on a diff in pull request #36165: [SPARK-36620][SHUFFLE] Add Push Based Shuffle client side metrics
Posted by GitBox <gi...@apache.org>.
mridulm commented on code in PR #36165:
URL: https://github.com/apache/spark/pull/36165#discussion_r1012357823
##########
core/src/main/scala/org/apache/spark/executor/ShuffleReadMetrics.scala:
##########
@@ -146,6 +268,16 @@ private[spark] class TempShuffleReadMetrics extends ShuffleReadMetricsReporter {
override def incLocalBytesRead(v: Long): Unit = _localBytesRead += v
override def incFetchWaitTime(v: Long): Unit = _fetchWaitTime += v
override def incRecordsRead(v: Long): Unit = _recordsRead += v
+ override def incCorruptMergedBlockChunks(v: Long): Unit = _corruptMergedBlockChunks += v
+ override def incFallbackCount(v: Long): Unit = _fallbackCount += v
+ override def incRemoteMergedBlocksFetched(v: Long): Unit = _remoteMergedBlocksFetched += v
+ override def incLocalMergedBlocksFetched(v: Long): Unit = _localMergedBlocksFetched += v
+ override def incRemoteMergedChunksFetched(v: Long): Unit = _remoteMergedChunksFetched += v
+ override def incLocalMergedChunksFetched(v: Long): Unit = _localMergedChunksFetched += v
+ override def incRemoteMergedBlocksBytesRead(v: Long): Unit = _remoteMergedBlocksBytesRead += v
+ override def incLocalMergedBlocksBytesRead(v: Long): Unit = _localMergedBlocksBytesRead += v
+ override def incRemoteReqsDuration(v: Long): Unit = _remoteReqsDuration += v
+ override def incRemoteMergedReqsDuration(v: Long): Unit = _remoteMergedReqsDuration += v
Review Comment:
I called out the same [above](https://github.com/apache/spark/pull/36165#discussion_r978229580)
I am fine with pulling those two metrics out and introducing them in a different PR
--
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
[GitHub] [spark] zhouyejoe commented on a diff in pull request #36165: [SPARK-36620][SHUFFLE] Add Push Based Shuffle client side read metrics
Posted by GitBox <gi...@apache.org>.
zhouyejoe commented on code in PR #36165:
URL: https://github.com/apache/spark/pull/36165#discussion_r1036608522
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/metric/SQLShuffleMetricsReporter.scala:
##########
@@ -44,6 +44,26 @@ class SQLShuffleReadMetricsReporter(
metrics(SQLShuffleReadMetricsReporter.FETCH_WAIT_TIME)
private[this] val _recordsRead =
metrics(SQLShuffleReadMetricsReporter.RECORDS_READ)
+ private[this] val _corruptMergedBlockChunks =
+ metrics(SQLShuffleReadMetricsReporter.CORRUPT_MERGED_BLOCK_CHUNKS)
+ private[this] val _fallbackCount =
+ metrics(SQLShuffleReadMetricsReporter.FALLBACK_COUNT)
Review Comment:
Build failed on this line. Need rename it to MERGED_FETCH_FALLBACK_COUNT.
--
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
[GitHub] [spark] thejdeep commented on pull request #36165: [SPARK-36620][SHUFFLE] Add Push Based Shuffle client side metrics
Posted by GitBox <gi...@apache.org>.
thejdeep commented on PR #36165:
URL: https://github.com/apache/spark/pull/36165#issuecomment-1329713626
Thanks for reviewing @zhouyejoe @mridulm, addressed feedback.
--
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
[GitHub] [spark] mridulm commented on a diff in pull request #36165: [SPARK-36620][SHUFFLE] Add Push Based Shuffle client side read metrics
Posted by GitBox <gi...@apache.org>.
mridulm commented on code in PR #36165:
URL: https://github.com/apache/spark/pull/36165#discussion_r1038913666
##########
core/src/test/scala/org/apache/spark/util/JsonProtocolSuite.scala:
##########
@@ -1780,7 +1802,19 @@ private[spark] object JsonProtocolSuite extends Assertions {
| "Remote Bytes Read" : 0,
| "Remote Bytes Read To Disk" : 0,
| "Local Bytes Read" : 0,
- | "Total Records Read" : 0
+ | "Total Records Read" : 0,
+ | "Remote Requests Duration": 0,
+ | "Push Based": {
Review Comment:
Let us do this rename as well. Thoughts ?
--
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
[GitHub] [spark] thejdeep commented on a diff in pull request #36165: [SPARK-36620][SHUFFLE] Add Push Based Shuffle client side read metrics
Posted by GitBox <gi...@apache.org>.
thejdeep commented on code in PR #36165:
URL: https://github.com/apache/spark/pull/36165#discussion_r1054375191
##########
core/src/test/scala/org/apache/spark/util/JsonProtocolSuite.scala:
##########
@@ -1653,7 +1663,19 @@ private[spark] object JsonProtocolSuite extends Assertions {
| "Remote Bytes Read": 1000,
| "Remote Bytes Read To Disk": 400,
| "Local Bytes Read": 1100,
- | "Total Records Read": 10
+ | "Total Records Read": 10,
+ | "Remote Requests Duration": 900,
+ | "Push Based Shuffle": {
+ | "Corrupt Merged Block Chunks" : 100,
+ | "Merged Fetch Fallback Count" : 100,
+ | "Merged Remote Blocks Fetched" : 0,
+ | "Merged Local Blocks Fetched" : 0,
+ | "Merged Remote Chunks Fetched" : 0,
+ | "Merged Local Chunks Fetched" : 0,
+ | "Merged Remote Bytes Read" : 0,
+ | "Merged Local Bytes Read" : 0,
+ | "Merged Remote Requests Duration": 900
Review Comment:
This is [currently being set](https://github.com/apache/spark/blob/e46031651617393bb182e44ae04920d6c89592f6/core/src/test/scala/org/apache/spark/util/JsonProtocolSuite.scala#L1315) to `executorDeserializeTime` + `jvmGCTime` while making the test task metrics, do you think that assumption is incorrect ?
--
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
[GitHub] [spark] otterc commented on a diff in pull request #36165: [SPARK-36620][SHUFFLE] Add Push Based Shuffle client side read metrics
Posted by GitBox <gi...@apache.org>.
otterc commented on code in PR #36165:
URL: https://github.com/apache/spark/pull/36165#discussion_r1054863925
##########
core/src/test/scala/org/apache/spark/util/JsonProtocolSuite.scala:
##########
@@ -1653,7 +1663,19 @@ private[spark] object JsonProtocolSuite extends Assertions {
| "Remote Bytes Read": 1000,
| "Remote Bytes Read To Disk": 400,
| "Local Bytes Read": 1100,
- | "Total Records Read": 10
+ | "Total Records Read": 10,
+ | "Remote Requests Duration": 900,
+ | "Push Based Shuffle": {
+ | "Corrupt Merged Block Chunks" : 100,
+ | "Merged Fetch Fallback Count" : 100,
+ | "Merged Remote Blocks Fetched" : 0,
+ | "Merged Local Blocks Fetched" : 0,
+ | "Merged Remote Chunks Fetched" : 0,
+ | "Merged Local Chunks Fetched" : 0,
+ | "Merged Remote Bytes Read" : 0,
+ | "Merged Local Bytes Read" : 0,
+ | "Merged Remote Requests Duration": 900
Review Comment:
merge remote request duration is not at all related to `executorDeserializeTime` and `jvmGCTime`. I guess we didn't pay attention to these tests internally.
--
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
[GitHub] [spark] otterc commented on a diff in pull request #36165: [SPARK-36620][SHUFFLE] Add Push Based Shuffle client side read metrics
Posted by GitBox <gi...@apache.org>.
otterc commented on code in PR #36165:
URL: https://github.com/apache/spark/pull/36165#discussion_r1046319200
##########
core/src/main/scala/org/apache/spark/util/JsonProtocol.scala:
##########
@@ -1105,11 +1133,17 @@ private[spark] object JsonProtocol {
case None => Seq.empty[AccumulableInfo]
}
}
+ val isPushBasedShuffleEnabled =
Review Comment:
Nit: rename to `isShufflePushEnabled`?
##########
core/src/test/resources/HistoryServerExpectations/stage_with_summaries_expectation.json:
##########
@@ -1094,5 +1308,7 @@
"MajorGCTime" : [ 0.0, 0.0, 339.0, 339.0, 339.0 ],
"TotalGCTime" : [ 0.0, 0.0, 0.0, 0.0, 0.0 ]
}
- }
-}
+ },
+ "isShufflePushEnabled" : false,
+ "shuffleMergersCount" : 0
+}
Review Comment:
Missing new line at the end of file
##########
core/src/test/resources/HistoryServerExpectations/stage_list_with_peak_metrics_expectation.json:
##########
@@ -203,5 +237,7 @@
"MajorGCCount" : 3,
"MajorGCTime" : 110,
"TotalGCTime" : 0
- }
-} ]
+ },
+ "isShufflePushEnabled" : false,
+ "shuffleMergersCount" : 0
+} ]
Review Comment:
Missing new line at the end of file
##########
project/MimaExcludes.scala:
##########
@@ -87,6 +87,11 @@ object MimaExcludes {
// [SPARK-36511][MINOR][SQL] Remove ColumnIOUtil
ProblemFilters.exclude[MissingClassProblem]("org.apache.parquet.io.ColumnIOUtil"),
+ // [SPARK-36620] [SHUFFLE] Expose push based shuffle metrics
Review Comment:
Reword to `Expose shuffle push metrics`
##########
core/src/test/scala/org/apache/spark/util/JsonProtocolSuite.scala:
##########
@@ -1653,7 +1663,19 @@ private[spark] object JsonProtocolSuite extends Assertions {
| "Remote Bytes Read": 1000,
| "Remote Bytes Read To Disk": 400,
| "Local Bytes Read": 1100,
- | "Total Records Read": 10
+ | "Total Records Read": 10,
+ | "Remote Requests Duration": 900,
+ | "Push Based Shuffle": {
+ | "Corrupt Merged Block Chunks" : 100,
+ | "Merged Fetch Fallback Count" : 100,
+ | "Merged Remote Blocks Fetched" : 0,
+ | "Merged Local Blocks Fetched" : 0,
+ | "Merged Remote Chunks Fetched" : 0,
+ | "Merged Local Chunks Fetched" : 0,
+ | "Merged Remote Bytes Read" : 0,
+ | "Merged Local Bytes Read" : 0,
+ | "Merged Remote Requests Duration": 900
Review Comment:
If merged remote blocks are 0 then this can't be > 0
##########
core/src/main/scala/org/apache/spark/util/JsonProtocol.scala:
##########
@@ -1225,6 +1259,29 @@ private[spark] object JsonProtocol {
readMetrics.incFetchWaitTime(readJson.get("Fetch Wait Time").extractLong)
readMetrics.incRecordsRead(
jsonOption(readJson.get("Total Records Read")).map(_.extractLong).getOrElse(0L))
+ readMetrics.incRemoteReqsDuration(jsonOption(readJson.get("Remote Requests Duration"))
+ .map(_.extractLong).getOrElse(0L))
+ jsonOption(readJson.get("Shuffle Push Read Metrics")).foreach { pushBasedShuffleReadJson =>
Review Comment:
Nit: Rename to shufflePushReadJson?
##########
core/src/test/resources/HistoryServerExpectations/stage_with_summaries_expectation.json:
##########
@@ -33,6 +33,16 @@
"shuffleLocalBytesRead" : 0,
"shuffleReadBytes" : 0,
"shuffleReadRecords" : 0,
+ "shuffleCorruptMergedBlockChunks" : 160,
+ "shuffleMergedFetchFallbackCount" : 32,
+ "shuffleMergedRemoteBlocksFetched" : 160,
+ "shuffleMergedLocalBlocksFetched" : 240,
+ "shuffleMergedRemoteChunksFetched" : 192,
+ "shuffleMergedLocalChunksFetched" : 160,
+ "shuffleMergedRemoteBytesRead" : 1600,
+ "shuffleMergedLocalBytesRead" : 1600,
+ "shuffleRemoteReqsDuration" : 0,
Review Comment:
Same here
##########
core/src/test/resources/HistoryServerExpectations/stage_with_peak_metrics_expectation.json:
##########
@@ -33,6 +33,16 @@
"shuffleLocalBytesRead" : 0,
"shuffleReadBytes" : 0,
"shuffleReadRecords" : 0,
+ "shuffleCorruptMergedBlockChunks" : 160,
+ "shuffleMergedFetchFallbackCount" : 32,
+ "shuffleMergedRemoteBlocksFetched" : 160,
+ "shuffleMergedLocalBlocksFetched" : 240,
+ "shuffleMergedRemoteChunksFetched" : 192,
+ "shuffleMergedLocalChunksFetched" : 160,
+ "shuffleMergedRemoteBytesRead" : 1600,
+ "shuffleMergedLocalBytesRead" : 1600,
+ "shuffleRemoteReqsDuration" : 0,
Review Comment:
if mergedRemoteReqDuration > 0, then shuffleRemoteReqsDuration should be greater than mergedRemoteReqDuration. Can we update this?
The existing values for shuffleLocalBytesRead, and shuffleReadBytes also don't make sense now because if we are saying that there were merged local blocks, then these value should be > 0.
We can either change or ignore. I am good with either.
##########
core/src/test/resources/HistoryServerExpectations/stage_list_json_expectation.json:
##########
@@ -270,5 +316,7 @@
"MajorGCCount" : 0,
"MajorGCTime" : 0,
"TotalGCTime" : 0
- }
-} ]
+ },
+ "isShufflePushEnabled" : false,
+ "shuffleMergersCount" : 0
+} ]
Review Comment:
Missing newline at the end of file
##########
core/src/test/resources/spark-events/app-20200706201101-0003:
##########
@@ -100,25 +100,25 @@
{"Event":"SparkListenerTaskStart","Stage ID":2,"Stage Attempt ID":0,"Task Info":{"Task ID":45,"Index":13,"Attempt":0,"Launch Time":1594091481129,"Executor ID":"0","Host":"127.0.0.1","Locality":"PROCESS_LOCAL","Speculative":false,"Getting Result Time":0,"Finish Time":0,"Failed":false,"Killed":false,"Accumulables":[]}}
{"Event":"SparkListenerTaskStart","Stage ID":2,"Stage Attempt ID":0,"Task Info":{"Task ID":46,"Index":14,"Attempt":0,"Launch Time":1594091481132,"Executor ID":"0","Host":"127.0.0.1","Locality":"PROCESS_LOCAL","Speculative":false,"Getting Result Time":0,"Finish Time":0,"Failed":false,"Killed":false,"Accumulables":[]}}
{"Event":"SparkListenerTaskStart","Stage ID":2,"Stage Attempt ID":0,"Task Info":{"Task ID":47,"Index":15,"Attempt":0,"Launch Time":1594091481136,"Executor ID":"0","Host":"127.0.0.1","Locality":"PROCESS_LOCAL","Speculative":false,"Getting Result Time":0,"Finish Time":0,"Failed":false,"Killed":false,"Accumulables":[]}}
-{"Event":"SparkListenerTaskEnd","Stage ID":2,"Stage Attempt ID":0,"Task Type":"ResultTask","Task End Reason":{"Reason":"Success"},"Task Info":{"Task ID":38,"Index":6,"Attempt":0,"Launch Time":1594091481104,"Executor ID":"0","Host":"127.0.0.1","Locality":"PROCESS_LOCAL","Speculative":false,"Getting Result Time":0,"Finish Time":1594091482939,"Failed":false,"Killed":false,"Accumulables":[{"ID":0,"Name":"duration","Update":"1467","Value":"1467","Internal":true,"Count Failed Values":true,"Metadata":"sql"},{"ID":1,"Name":"number of output rows","Update":"375000","Value":"375000","Internal":true,"Count Failed Values":true,"Metadata":"sql"},{"ID":14,"Name":"number of output rows","Update":"6250","Value":"6250","Internal":true,"Count Failed Values":true,"Metadata":"sql"},{"ID":74,"Name":"internal.metrics.peakExecutionMemory","Update":24040,"Value":24040,"Internal":true,"Count Failed Values":true},{"ID":71,"Name":"internal.metrics.resultSerializationTime","Update":11,"Value":11,"Internal":tru
e,"Count Failed Values":true},{"ID":70,"Name":"internal.metrics.jvmGCTime","Update":62,"Value":62,"Internal":true,"Count Failed Values":true},{"ID":69,"Name":"internal.metrics.resultSize","Update":2158,"Value":2158,"Internal":true,"Count Failed Values":true},{"ID":68,"Name":"internal.metrics.executorCpuTime","Update":414110000,"Value":414110000,"Internal":true,"Count Failed Values":true},{"ID":67,"Name":"internal.metrics.executorRunTime","Update":1498,"Value":1498,"Internal":true,"Count Failed Values":true},{"ID":66,"Name":"internal.metrics.executorDeserializeCpuTime","Update":60358000,"Value":60358000,"Internal":true,"Count Failed Values":true},{"ID":65,"Name":"internal.metrics.executorDeserializeTime","Update":255,"Value":255,"Internal":true,"Count Failed Values":true}]},"Task Executor Metrics":{"JVMHeapMemory":0,"JVMOffHeapMemory":0,"OnHeapExecutionMemory":0,"OffHeapExecutionMemory":0,"OnHeapStorageMemory":0,"OffHeapStorageMemory":0,"OnHeapUnifiedMemory":0,"OffHeapUnifiedMemory":
0,"DirectPoolMemory":0,"MappedPoolMemory":0,"ProcessTreeJVMVMemory":0,"ProcessTreeJVMRSSMemory":0,"ProcessTreePythonVMemory":0,"ProcessTreePythonRSSMemory":0,"ProcessTreeOtherVMemory":0,"ProcessTreeOtherRSSMemory":0,"MinorGCCount":0,"MinorGCTime":0,"MajorGCCount":0,"MajorGCTime":0},"Task Metrics":{"Executor Deserialize Time":255,"Executor Deserialize CPU Time":60358000,"Executor Run Time":1498,"Executor CPU Time":414110000,"Peak Execution Memory":24040,"Result Size":2158,"JVM GC Time":62,"Result Serialization Time":11,"Memory Bytes Spilled":0,"Disk Bytes Spilled":0,"Shuffle Read Metrics":{"Remote Blocks Fetched":0,"Local Blocks Fetched":0,"Fetch Wait Time":0,"Remote Bytes Read":0,"Remote Bytes Read To Disk":0,"Local Bytes Read":0,"Total Records Read":0},"Shuffle Write Metrics":{"Shuffle Bytes Written":0,"Shuffle Write Time":0,"Shuffle Records Written":0},"Input Metrics":{"Bytes Read":0,"Records Read":0},"Output Metrics":{"Bytes Written":0,"Records Written":0},"Updated Blocks":[]}}
-{"Event":"SparkListenerTaskEnd","Stage ID":2,"Stage Attempt ID":0,"Task Type":"ResultTask","Task End Reason":{"Reason":"Success"},"Task Info":{"Task ID":47,"Index":15,"Attempt":0,"Launch Time":1594091481136,"Executor ID":"0","Host":"127.0.0.1","Locality":"PROCESS_LOCAL","Speculative":false,"Getting Result Time":0,"Finish Time":1594091483014,"Failed":false,"Killed":false,"Accumulables":[{"ID":0,"Name":"duration","Update":"1563","Value":"3030","Internal":true,"Count Failed Values":true,"Metadata":"sql"},{"ID":1,"Name":"number of output rows","Update":"375000","Value":"750000","Internal":true,"Count Failed Values":true,"Metadata":"sql"},{"ID":14,"Name":"number of output rows","Update":"6250","Value":"12500","Internal":true,"Count Failed Values":true,"Metadata":"sql"},{"ID":74,"Name":"internal.metrics.peakExecutionMemory","Update":24040,"Value":48080,"Internal":true,"Count Failed Values":true},{"ID":70,"Name":"internal.metrics.jvmGCTime","Update":52,"Value":114,"Internal":true,"Count Fa
iled Values":true},{"ID":69,"Name":"internal.metrics.resultSize","Update":2115,"Value":4273,"Internal":true,"Count Failed Values":true},{"ID":68,"Name":"internal.metrics.executorCpuTime","Update":1324251000,"Value":1738361000,"Internal":true,"Count Failed Values":true},{"ID":67,"Name":"internal.metrics.executorRunTime","Update":1594,"Value":3092,"Internal":true,"Count Failed Values":true},{"ID":66,"Name":"internal.metrics.executorDeserializeCpuTime","Update":47496000,"Value":107854000,"Internal":true,"Count Failed Values":true},{"ID":65,"Name":"internal.metrics.executorDeserializeTime","Update":213,"Value":468,"Internal":true,"Count Failed Values":true}]},"Task Executor Metrics":{"JVMHeapMemory":0,"JVMOffHeapMemory":0,"OnHeapExecutionMemory":0,"OffHeapExecutionMemory":0,"OnHeapStorageMemory":0,"OffHeapStorageMemory":0,"OnHeapUnifiedMemory":0,"OffHeapUnifiedMemory":0,"DirectPoolMemory":0,"MappedPoolMemory":0,"ProcessTreeJVMVMemory":0,"ProcessTreeJVMRSSMemory":0,"ProcessTreePythonVMem
ory":0,"ProcessTreePythonRSSMemory":0,"ProcessTreeOtherVMemory":0,"ProcessTreeOtherRSSMemory":0,"MinorGCCount":0,"MinorGCTime":0,"MajorGCCount":0,"MajorGCTime":0},"Task Metrics":{"Executor Deserialize Time":213,"Executor Deserialize CPU Time":47496000,"Executor Run Time":1594,"Executor CPU Time":1324251000,"Peak Execution Memory":24040,"Result Size":2115,"JVM GC Time":52,"Result Serialization Time":0,"Memory Bytes Spilled":0,"Disk Bytes Spilled":0,"Shuffle Read Metrics":{"Remote Blocks Fetched":0,"Local Blocks Fetched":0,"Fetch Wait Time":0,"Remote Bytes Read":0,"Remote Bytes Read To Disk":0,"Local Bytes Read":0,"Total Records Read":0},"Shuffle Write Metrics":{"Shuffle Bytes Written":0,"Shuffle Write Time":0,"Shuffle Records Written":0},"Input Metrics":{"Bytes Read":0,"Records Read":0},"Output Metrics":{"Bytes Written":0,"Records Written":0},"Updated Blocks":[]}}
-{"Event":"SparkListenerTaskEnd","Stage ID":2,"Stage Attempt ID":0,"Task Type":"ResultTask","Task End Reason":{"Reason":"Success"},"Task Info":{"Task ID":37,"Index":5,"Attempt":0,"Launch Time":1594091481100,"Executor ID":"0","Host":"127.0.0.1","Locality":"PROCESS_LOCAL","Speculative":false,"Getting Result Time":0,"Finish Time":1594091483015,"Failed":false,"Killed":false,"Accumulables":[{"ID":0,"Name":"duration","Update":"1563","Value":"4593","Internal":true,"Count Failed Values":true,"Metadata":"sql"},{"ID":1,"Name":"number of output rows","Update":"375000","Value":"1125000","Internal":true,"Count Failed Values":true,"Metadata":"sql"},{"ID":14,"Name":"number of output rows","Update":"6250","Value":"18750","Internal":true,"Count Failed Values":true,"Metadata":"sql"},{"ID":74,"Name":"internal.metrics.peakExecutionMemory","Update":24040,"Value":72120,"Internal":true,"Count Failed Values":true},{"ID":70,"Name":"internal.metrics.jvmGCTime","Update":62,"Value":176,"Internal":true,"Count Fa
iled Values":true},{"ID":69,"Name":"internal.metrics.resultSize","Update":2115,"Value":6388,"Internal":true,"Count Failed Values":true},{"ID":68,"Name":"internal.metrics.executorCpuTime","Update":507192000,"Value":2245553000,"Internal":true,"Count Failed Values":true},{"ID":67,"Name":"internal.metrics.executorRunTime","Update":1596,"Value":4688,"Internal":true,"Count Failed Values":true},{"ID":66,"Name":"internal.metrics.executorDeserializeCpuTime","Update":60890000,"Value":168744000,"Internal":true,"Count Failed Values":true},{"ID":65,"Name":"internal.metrics.executorDeserializeTime","Update":256,"Value":724,"Internal":true,"Count Failed Values":true}]},"Task Executor Metrics":{"JVMHeapMemory":0,"JVMOffHeapMemory":0,"OnHeapExecutionMemory":0,"OffHeapExecutionMemory":0,"OnHeapStorageMemory":0,"OffHeapStorageMemory":0,"OnHeapUnifiedMemory":0,"OffHeapUnifiedMemory":0,"DirectPoolMemory":0,"MappedPoolMemory":0,"ProcessTreeJVMVMemory":0,"ProcessTreeJVMRSSMemory":0,"ProcessTreePythonVMemo
ry":0,"ProcessTreePythonRSSMemory":0,"ProcessTreeOtherVMemory":0,"ProcessTreeOtherRSSMemory":0,"MinorGCCount":0,"MinorGCTime":0,"MajorGCCount":0,"MajorGCTime":0},"Task Metrics":{"Executor Deserialize Time":256,"Executor Deserialize CPU Time":60890000,"Executor Run Time":1596,"Executor CPU Time":507192000,"Peak Execution Memory":24040,"Result Size":2115,"JVM GC Time":62,"Result Serialization Time":0,"Memory Bytes Spilled":0,"Disk Bytes Spilled":0,"Shuffle Read Metrics":{"Remote Blocks Fetched":0,"Local Blocks Fetched":0,"Fetch Wait Time":0,"Remote Bytes Read":0,"Remote Bytes Read To Disk":0,"Local Bytes Read":0,"Total Records Read":0},"Shuffle Write Metrics":{"Shuffle Bytes Written":0,"Shuffle Write Time":0,"Shuffle Records Written":0},"Input Metrics":{"Bytes Read":0,"Records Read":0},"Output Metrics":{"Bytes Written":0,"Records Written":0},"Updated Blocks":[]}}
-{"Event":"SparkListenerTaskEnd","Stage ID":2,"Stage Attempt ID":0,"Task Type":"ResultTask","Task End Reason":{"Reason":"Success"},"Task Info":{"Task ID":36,"Index":4,"Attempt":0,"Launch Time":1594091481095,"Executor ID":"0","Host":"127.0.0.1","Locality":"PROCESS_LOCAL","Speculative":false,"Getting Result Time":0,"Finish Time":1594091483015,"Failed":false,"Killed":false,"Accumulables":[{"ID":0,"Name":"duration","Update":"1564","Value":"6157","Internal":true,"Count Failed Values":true,"Metadata":"sql"},{"ID":1,"Name":"number of output rows","Update":"375000","Value":"1500000","Internal":true,"Count Failed Values":true,"Metadata":"sql"},{"ID":14,"Name":"number of output rows","Update":"6250","Value":"25000","Internal":true,"Count Failed Values":true,"Metadata":"sql"},{"ID":74,"Name":"internal.metrics.peakExecutionMemory","Update":24040,"Value":96160,"Internal":true,"Count Failed Values":true},{"ID":70,"Name":"internal.metrics.jvmGCTime","Update":62,"Value":238,"Internal":true,"Count Fa
iled Values":true},{"ID":69,"Name":"internal.metrics.resultSize","Update":2115,"Value":8503,"Internal":true,"Count Failed Values":true},{"ID":68,"Name":"internal.metrics.executorCpuTime","Update":503010000,"Value":2748563000,"Internal":true,"Count Failed Values":true},{"ID":67,"Name":"internal.metrics.executorRunTime","Update":1596,"Value":6284,"Internal":true,"Count Failed Values":true},{"ID":66,"Name":"internal.metrics.executorDeserializeCpuTime","Update":112849000,"Value":281593000,"Internal":true,"Count Failed Values":true},{"ID":65,"Name":"internal.metrics.executorDeserializeTime","Update":260,"Value":984,"Internal":true,"Count Failed Values":true}]},"Task Executor Metrics":{"JVMHeapMemory":0,"JVMOffHeapMemory":0,"OnHeapExecutionMemory":0,"OffHeapExecutionMemory":0,"OnHeapStorageMemory":0,"OffHeapStorageMemory":0,"OnHeapUnifiedMemory":0,"OffHeapUnifiedMemory":0,"DirectPoolMemory":0,"MappedPoolMemory":0,"ProcessTreeJVMVMemory":0,"ProcessTreeJVMRSSMemory":0,"ProcessTreePythonVMem
ory":0,"ProcessTreePythonRSSMemory":0,"ProcessTreeOtherVMemory":0,"ProcessTreeOtherRSSMemory":0,"MinorGCCount":0,"MinorGCTime":0,"MajorGCCount":0,"MajorGCTime":0},"Task Metrics":{"Executor Deserialize Time":260,"Executor Deserialize CPU Time":112849000,"Executor Run Time":1596,"Executor CPU Time":503010000,"Peak Execution Memory":24040,"Result Size":2115,"JVM GC Time":62,"Result Serialization Time":0,"Memory Bytes Spilled":0,"Disk Bytes Spilled":0,"Shuffle Read Metrics":{"Remote Blocks Fetched":0,"Local Blocks Fetched":0,"Fetch Wait Time":0,"Remote Bytes Read":0,"Remote Bytes Read To Disk":0,"Local Bytes Read":0,"Total Records Read":0},"Shuffle Write Metrics":{"Shuffle Bytes Written":0,"Shuffle Write Time":0,"Shuffle Records Written":0},"Input Metrics":{"Bytes Read":0,"Records Read":0},"Output Metrics":{"Bytes Written":0,"Records Written":0},"Updated Blocks":[]}}
-{"Event":"SparkListenerTaskEnd","Stage ID":2,"Stage Attempt ID":0,"Task Type":"ResultTask","Task End Reason":{"Reason":"Success"},"Task Info":{"Task ID":35,"Index":3,"Attempt":0,"Launch Time":1594091481091,"Executor ID":"0","Host":"127.0.0.1","Locality":"PROCESS_LOCAL","Speculative":false,"Getting Result Time":0,"Finish Time":1594091483016,"Failed":false,"Killed":false,"Accumulables":[{"ID":0,"Name":"duration","Update":"1566","Value":"7723","Internal":true,"Count Failed Values":true,"Metadata":"sql"},{"ID":1,"Name":"number of output rows","Update":"375000","Value":"1875000","Internal":true,"Count Failed Values":true,"Metadata":"sql"},{"ID":14,"Name":"number of output rows","Update":"6250","Value":"31250","Internal":true,"Count Failed Values":true,"Metadata":"sql"},{"ID":74,"Name":"internal.metrics.peakExecutionMemory","Update":24040,"Value":120200,"Internal":true,"Count Failed Values":true},{"ID":70,"Name":"internal.metrics.jvmGCTime","Update":62,"Value":300,"Internal":true,"Count F
ailed Values":true},{"ID":69,"Name":"internal.metrics.resultSize","Update":2115,"Value":10618,"Internal":true,"Count Failed Values":true},{"ID":68,"Name":"internal.metrics.executorCpuTime","Update":502908000,"Value":3251471000,"Internal":true,"Count Failed Values":true},{"ID":67,"Name":"internal.metrics.executorRunTime","Update":1598,"Value":7882,"Internal":true,"Count Failed Values":true},{"ID":66,"Name":"internal.metrics.executorDeserializeCpuTime","Update":62944000,"Value":344537000,"Internal":true,"Count Failed Values":true},{"ID":65,"Name":"internal.metrics.executorDeserializeTime","Update":263,"Value":1247,"Internal":true,"Count Failed Values":true}]},"Task Executor Metrics":{"JVMHeapMemory":0,"JVMOffHeapMemory":0,"OnHeapExecutionMemory":0,"OffHeapExecutionMemory":0,"OnHeapStorageMemory":0,"OffHeapStorageMemory":0,"OnHeapUnifiedMemory":0,"OffHeapUnifiedMemory":0,"DirectPoolMemory":0,"MappedPoolMemory":0,"ProcessTreeJVMVMemory":0,"ProcessTreeJVMRSSMemory":0,"ProcessTreePythonVM
emory":0,"ProcessTreePythonRSSMemory":0,"ProcessTreeOtherVMemory":0,"ProcessTreeOtherRSSMemory":0,"MinorGCCount":0,"MinorGCTime":0,"MajorGCCount":0,"MajorGCTime":0},"Task Metrics":{"Executor Deserialize Time":263,"Executor Deserialize CPU Time":62944000,"Executor Run Time":1598,"Executor CPU Time":502908000,"Peak Execution Memory":24040,"Result Size":2115,"JVM GC Time":62,"Result Serialization Time":0,"Memory Bytes Spilled":0,"Disk Bytes Spilled":0,"Shuffle Read Metrics":{"Remote Blocks Fetched":0,"Local Blocks Fetched":0,"Fetch Wait Time":0,"Remote Bytes Read":0,"Remote Bytes Read To Disk":0,"Local Bytes Read":0,"Total Records Read":0},"Shuffle Write Metrics":{"Shuffle Bytes Written":0,"Shuffle Write Time":0,"Shuffle Records Written":0},"Input Metrics":{"Bytes Read":0,"Records Read":0},"Output Metrics":{"Bytes Written":0,"Records Written":0},"Updated Blocks":[]}}
-{"Event":"SparkListenerTaskEnd","Stage ID":2,"Stage Attempt ID":0,"Task Type":"ResultTask","Task End Reason":{"Reason":"Success"},"Task Info":{"Task ID":40,"Index":8,"Attempt":0,"Launch Time":1594091481112,"Executor ID":"0","Host":"127.0.0.1","Locality":"PROCESS_LOCAL","Speculative":false,"Getting Result Time":0,"Finish Time":1594091483016,"Failed":false,"Killed":false,"Accumulables":[{"ID":0,"Name":"duration","Update":"1562","Value":"9285","Internal":true,"Count Failed Values":true,"Metadata":"sql"},{"ID":1,"Name":"number of output rows","Update":"375000","Value":"2250000","Internal":true,"Count Failed Values":true,"Metadata":"sql"},{"ID":14,"Name":"number of output rows","Update":"6250","Value":"37500","Internal":true,"Count Failed Values":true,"Metadata":"sql"},{"ID":74,"Name":"internal.metrics.peakExecutionMemory","Update":24040,"Value":144240,"Internal":true,"Count Failed Values":true},{"ID":70,"Name":"internal.metrics.jvmGCTime","Update":62,"Value":362,"Internal":true,"Count F
ailed Values":true},{"ID":69,"Name":"internal.metrics.resultSize","Update":2115,"Value":12733,"Internal":true,"Count Failed Values":true},{"ID":68,"Name":"internal.metrics.executorCpuTime","Update":510597000,"Value":3762068000,"Internal":true,"Count Failed Values":true},{"ID":67,"Name":"internal.metrics.executorRunTime","Update":1595,"Value":9477,"Internal":true,"Count Failed Values":true},{"ID":66,"Name":"internal.metrics.executorDeserializeCpuTime","Update":69760000,"Value":414297000,"Internal":true,"Count Failed Values":true},{"ID":65,"Name":"internal.metrics.executorDeserializeTime","Update":246,"Value":1493,"Internal":true,"Count Failed Values":true}]},"Task Executor Metrics":{"JVMHeapMemory":0,"JVMOffHeapMemory":0,"OnHeapExecutionMemory":0,"OffHeapExecutionMemory":0,"OnHeapStorageMemory":0,"OffHeapStorageMemory":0,"OnHeapUnifiedMemory":0,"OffHeapUnifiedMemory":0,"DirectPoolMemory":0,"MappedPoolMemory":0,"ProcessTreeJVMVMemory":0,"ProcessTreeJVMRSSMemory":0,"ProcessTreePythonVM
emory":0,"ProcessTreePythonRSSMemory":0,"ProcessTreeOtherVMemory":0,"ProcessTreeOtherRSSMemory":0,"MinorGCCount":0,"MinorGCTime":0,"MajorGCCount":0,"MajorGCTime":0},"Task Metrics":{"Executor Deserialize Time":246,"Executor Deserialize CPU Time":69760000,"Executor Run Time":1595,"Executor CPU Time":510597000,"Peak Execution Memory":24040,"Result Size":2115,"JVM GC Time":62,"Result Serialization Time":0,"Memory Bytes Spilled":0,"Disk Bytes Spilled":0,"Shuffle Read Metrics":{"Remote Blocks Fetched":0,"Local Blocks Fetched":0,"Fetch Wait Time":0,"Remote Bytes Read":0,"Remote Bytes Read To Disk":0,"Local Bytes Read":0,"Total Records Read":0},"Shuffle Write Metrics":{"Shuffle Bytes Written":0,"Shuffle Write Time":0,"Shuffle Records Written":0},"Input Metrics":{"Bytes Read":0,"Records Read":0},"Output Metrics":{"Bytes Written":0,"Records Written":0},"Updated Blocks":[]}}
-{"Event":"SparkListenerTaskEnd","Stage ID":2,"Stage Attempt ID":0,"Task Type":"ResultTask","Task End Reason":{"Reason":"Success"},"Task Info":{"Task ID":45,"Index":13,"Attempt":0,"Launch Time":1594091481129,"Executor ID":"0","Host":"127.0.0.1","Locality":"PROCESS_LOCAL","Speculative":false,"Getting Result Time":0,"Finish Time":1594091483024,"Failed":false,"Killed":false,"Accumulables":[{"ID":0,"Name":"duration","Update":"1564","Value":"10849","Internal":true,"Count Failed Values":true,"Metadata":"sql"},{"ID":1,"Name":"number of output rows","Update":"375000","Value":"2625000","Internal":true,"Count Failed Values":true,"Metadata":"sql"},{"ID":14,"Name":"number of output rows","Update":"6250","Value":"43750","Internal":true,"Count Failed Values":true,"Metadata":"sql"},{"ID":74,"Name":"internal.metrics.peakExecutionMemory","Update":24040,"Value":168280,"Internal":true,"Count Failed Values":true},{"ID":70,"Name":"internal.metrics.jvmGCTime","Update":62,"Value":424,"Internal":true,"Count
Failed Values":true},{"ID":69,"Name":"internal.metrics.resultSize","Update":2115,"Value":14848,"Internal":true,"Count Failed Values":true},{"ID":68,"Name":"internal.metrics.executorCpuTime","Update":495138000,"Value":4257206000,"Internal":true,"Count Failed Values":true},{"ID":67,"Name":"internal.metrics.executorRunTime","Update":1595,"Value":11072,"Internal":true,"Count Failed Values":true},{"ID":66,"Name":"internal.metrics.executorDeserializeCpuTime","Update":54222000,"Value":468519000,"Internal":true,"Count Failed Values":true},{"ID":65,"Name":"internal.metrics.executorDeserializeTime","Update":221,"Value":1714,"Internal":true,"Count Failed Values":true}]},"Task Executor Metrics":{"JVMHeapMemory":0,"JVMOffHeapMemory":0,"OnHeapExecutionMemory":0,"OffHeapExecutionMemory":0,"OnHeapStorageMemory":0,"OffHeapStorageMemory":0,"OnHeapUnifiedMemory":0,"OffHeapUnifiedMemory":0,"DirectPoolMemory":0,"MappedPoolMemory":0,"ProcessTreeJVMVMemory":0,"ProcessTreeJVMRSSMemory":0,"ProcessTreePytho
nVMemory":0,"ProcessTreePythonRSSMemory":0,"ProcessTreeOtherVMemory":0,"ProcessTreeOtherRSSMemory":0,"MinorGCCount":0,"MinorGCTime":0,"MajorGCCount":0,"MajorGCTime":0},"Task Metrics":{"Executor Deserialize Time":221,"Executor Deserialize CPU Time":54222000,"Executor Run Time":1595,"Executor CPU Time":495138000,"Peak Execution Memory":24040,"Result Size":2115,"JVM GC Time":62,"Result Serialization Time":0,"Memory Bytes Spilled":0,"Disk Bytes Spilled":0,"Shuffle Read Metrics":{"Remote Blocks Fetched":0,"Local Blocks Fetched":0,"Fetch Wait Time":0,"Remote Bytes Read":0,"Remote Bytes Read To Disk":0,"Local Bytes Read":0,"Total Records Read":0},"Shuffle Write Metrics":{"Shuffle Bytes Written":0,"Shuffle Write Time":0,"Shuffle Records Written":0},"Input Metrics":{"Bytes Read":0,"Records Read":0},"Output Metrics":{"Bytes Written":0,"Records Written":0},"Updated Blocks":[]}}
-{"Event":"SparkListenerTaskEnd","Stage ID":2,"Stage Attempt ID":0,"Task Type":"ResultTask","Task End Reason":{"Reason":"Success"},"Task Info":{"Task ID":39,"Index":7,"Attempt":0,"Launch Time":1594091481109,"Executor ID":"0","Host":"127.0.0.1","Locality":"PROCESS_LOCAL","Speculative":false,"Getting Result Time":0,"Finish Time":1594091483024,"Failed":false,"Killed":false,"Accumulables":[{"ID":0,"Name":"duration","Update":"1566","Value":"12415","Internal":true,"Count Failed Values":true,"Metadata":"sql"},{"ID":1,"Name":"number of output rows","Update":"375000","Value":"3000000","Internal":true,"Count Failed Values":true,"Metadata":"sql"},{"ID":14,"Name":"number of output rows","Update":"6250","Value":"50000","Internal":true,"Count Failed Values":true,"Metadata":"sql"},{"ID":74,"Name":"internal.metrics.peakExecutionMemory","Update":24040,"Value":192320,"Internal":true,"Count Failed Values":true},{"ID":70,"Name":"internal.metrics.jvmGCTime","Update":62,"Value":486,"Internal":true,"Count
Failed Values":true},{"ID":69,"Name":"internal.metrics.resultSize","Update":2115,"Value":16963,"Internal":true,"Count Failed Values":true},{"ID":68,"Name":"internal.metrics.executorCpuTime","Update":539451000,"Value":4796657000,"Internal":true,"Count Failed Values":true},{"ID":67,"Name":"internal.metrics.executorRunTime","Update":1596,"Value":12668,"Internal":true,"Count Failed Values":true},{"ID":66,"Name":"internal.metrics.executorDeserializeCpuTime","Update":64380000,"Value":532899000,"Internal":true,"Count Failed Values":true},{"ID":65,"Name":"internal.metrics.executorDeserializeTime","Update":254,"Value":1968,"Internal":true,"Count Failed Values":true}]},"Task Executor Metrics":{"JVMHeapMemory":0,"JVMOffHeapMemory":0,"OnHeapExecutionMemory":0,"OffHeapExecutionMemory":0,"OnHeapStorageMemory":0,"OffHeapStorageMemory":0,"OnHeapUnifiedMemory":0,"OffHeapUnifiedMemory":0,"DirectPoolMemory":0,"MappedPoolMemory":0,"ProcessTreeJVMVMemory":0,"ProcessTreeJVMRSSMemory":0,"ProcessTreePython
VMemory":0,"ProcessTreePythonRSSMemory":0,"ProcessTreeOtherVMemory":0,"ProcessTreeOtherRSSMemory":0,"MinorGCCount":0,"MinorGCTime":0,"MajorGCCount":0,"MajorGCTime":0},"Task Metrics":{"Executor Deserialize Time":254,"Executor Deserialize CPU Time":64380000,"Executor Run Time":1596,"Executor CPU Time":539451000,"Peak Execution Memory":24040,"Result Size":2115,"JVM GC Time":62,"Result Serialization Time":0,"Memory Bytes Spilled":0,"Disk Bytes Spilled":0,"Shuffle Read Metrics":{"Remote Blocks Fetched":0,"Local Blocks Fetched":0,"Fetch Wait Time":0,"Remote Bytes Read":0,"Remote Bytes Read To Disk":0,"Local Bytes Read":0,"Total Records Read":0},"Shuffle Write Metrics":{"Shuffle Bytes Written":0,"Shuffle Write Time":0,"Shuffle Records Written":0},"Input Metrics":{"Bytes Read":0,"Records Read":0},"Output Metrics":{"Bytes Written":0,"Records Written":0},"Updated Blocks":[]}}
-{"Event":"SparkListenerTaskEnd","Stage ID":2,"Stage Attempt ID":0,"Task Type":"ResultTask","Task End Reason":{"Reason":"Success"},"Task Info":{"Task ID":33,"Index":1,"Attempt":0,"Launch Time":1594091481082,"Executor ID":"0","Host":"127.0.0.1","Locality":"PROCESS_LOCAL","Speculative":false,"Getting Result Time":0,"Finish Time":1594091483025,"Failed":false,"Killed":false,"Accumulables":[{"ID":0,"Name":"duration","Update":"1566","Value":"13981","Internal":true,"Count Failed Values":true,"Metadata":"sql"},{"ID":1,"Name":"number of output rows","Update":"375000","Value":"3375000","Internal":true,"Count Failed Values":true,"Metadata":"sql"},{"ID":14,"Name":"number of output rows","Update":"6250","Value":"56250","Internal":true,"Count Failed Values":true,"Metadata":"sql"},{"ID":74,"Name":"internal.metrics.peakExecutionMemory","Update":24040,"Value":216360,"Internal":true,"Count Failed Values":true},{"ID":70,"Name":"internal.metrics.jvmGCTime","Update":62,"Value":548,"Internal":true,"Count
Failed Values":true},{"ID":69,"Name":"internal.metrics.resultSize","Update":2115,"Value":19078,"Internal":true,"Count Failed Values":true},{"ID":68,"Name":"internal.metrics.executorCpuTime","Update":519178000,"Value":5315835000,"Internal":true,"Count Failed Values":true},{"ID":67,"Name":"internal.metrics.executorRunTime","Update":1597,"Value":14265,"Internal":true,"Count Failed Values":true},{"ID":66,"Name":"internal.metrics.executorDeserializeCpuTime","Update":54442000,"Value":587341000,"Internal":true,"Count Failed Values":true},{"ID":65,"Name":"internal.metrics.executorDeserializeTime","Update":267,"Value":2235,"Internal":true,"Count Failed Values":true}]},"Task Executor Metrics":{"JVMHeapMemory":0,"JVMOffHeapMemory":0,"OnHeapExecutionMemory":0,"OffHeapExecutionMemory":0,"OnHeapStorageMemory":0,"OffHeapStorageMemory":0,"OnHeapUnifiedMemory":0,"OffHeapUnifiedMemory":0,"DirectPoolMemory":0,"MappedPoolMemory":0,"ProcessTreeJVMVMemory":0,"ProcessTreeJVMRSSMemory":0,"ProcessTreePython
VMemory":0,"ProcessTreePythonRSSMemory":0,"ProcessTreeOtherVMemory":0,"ProcessTreeOtherRSSMemory":0,"MinorGCCount":0,"MinorGCTime":0,"MajorGCCount":0,"MajorGCTime":0},"Task Metrics":{"Executor Deserialize Time":267,"Executor Deserialize CPU Time":54442000,"Executor Run Time":1597,"Executor CPU Time":519178000,"Peak Execution Memory":24040,"Result Size":2115,"JVM GC Time":62,"Result Serialization Time":0,"Memory Bytes Spilled":0,"Disk Bytes Spilled":0,"Shuffle Read Metrics":{"Remote Blocks Fetched":0,"Local Blocks Fetched":0,"Fetch Wait Time":0,"Remote Bytes Read":0,"Remote Bytes Read To Disk":0,"Local Bytes Read":0,"Total Records Read":0},"Shuffle Write Metrics":{"Shuffle Bytes Written":0,"Shuffle Write Time":0,"Shuffle Records Written":0},"Input Metrics":{"Bytes Read":0,"Records Read":0},"Output Metrics":{"Bytes Written":0,"Records Written":0},"Updated Blocks":[]}}
-{"Event":"SparkListenerTaskEnd","Stage ID":2,"Stage Attempt ID":0,"Task Type":"ResultTask","Task End Reason":{"Reason":"Success"},"Task Info":{"Task ID":34,"Index":2,"Attempt":0,"Launch Time":1594091481087,"Executor ID":"0","Host":"127.0.0.1","Locality":"PROCESS_LOCAL","Speculative":false,"Getting Result Time":0,"Finish Time":1594091483026,"Failed":false,"Killed":false,"Accumulables":[{"ID":0,"Name":"duration","Update":"1574","Value":"15555","Internal":true,"Count Failed Values":true,"Metadata":"sql"},{"ID":1,"Name":"number of output rows","Update":"375000","Value":"3750000","Internal":true,"Count Failed Values":true,"Metadata":"sql"},{"ID":14,"Name":"number of output rows","Update":"6250","Value":"62500","Internal":true,"Count Failed Values":true,"Metadata":"sql"},{"ID":74,"Name":"internal.metrics.peakExecutionMemory","Update":24040,"Value":240400,"Internal":true,"Count Failed Values":true},{"ID":70,"Name":"internal.metrics.jvmGCTime","Update":66,"Value":614,"Internal":true,"Count
Failed Values":true},{"ID":69,"Name":"internal.metrics.resultSize","Update":2115,"Value":21193,"Internal":true,"Count Failed Values":true},{"ID":68,"Name":"internal.metrics.executorCpuTime","Update":508433000,"Value":5824268000,"Internal":true,"Count Failed Values":true},{"ID":67,"Name":"internal.metrics.executorRunTime","Update":1606,"Value":15871,"Internal":true,"Count Failed Values":true},{"ID":66,"Name":"internal.metrics.executorDeserializeCpuTime","Update":69492000,"Value":656833000,"Internal":true,"Count Failed Values":true},{"ID":65,"Name":"internal.metrics.executorDeserializeTime","Update":265,"Value":2500,"Internal":true,"Count Failed Values":true}]},"Task Executor Metrics":{"JVMHeapMemory":0,"JVMOffHeapMemory":0,"OnHeapExecutionMemory":0,"OffHeapExecutionMemory":0,"OnHeapStorageMemory":0,"OffHeapStorageMemory":0,"OnHeapUnifiedMemory":0,"OffHeapUnifiedMemory":0,"DirectPoolMemory":0,"MappedPoolMemory":0,"ProcessTreeJVMVMemory":0,"ProcessTreeJVMRSSMemory":0,"ProcessTreePython
VMemory":0,"ProcessTreePythonRSSMemory":0,"ProcessTreeOtherVMemory":0,"ProcessTreeOtherRSSMemory":0,"MinorGCCount":0,"MinorGCTime":0,"MajorGCCount":0,"MajorGCTime":0},"Task Metrics":{"Executor Deserialize Time":265,"Executor Deserialize CPU Time":69492000,"Executor Run Time":1606,"Executor CPU Time":508433000,"Peak Execution Memory":24040,"Result Size":2115,"JVM GC Time":66,"Result Serialization Time":0,"Memory Bytes Spilled":0,"Disk Bytes Spilled":0,"Shuffle Read Metrics":{"Remote Blocks Fetched":0,"Local Blocks Fetched":0,"Fetch Wait Time":0,"Remote Bytes Read":0,"Remote Bytes Read To Disk":0,"Local Bytes Read":0,"Total Records Read":0},"Shuffle Write Metrics":{"Shuffle Bytes Written":0,"Shuffle Write Time":0,"Shuffle Records Written":0},"Input Metrics":{"Bytes Read":0,"Records Read":0},"Output Metrics":{"Bytes Written":0,"Records Written":0},"Updated Blocks":[]}}
-{"Event":"SparkListenerTaskEnd","Stage ID":2,"Stage Attempt ID":0,"Task Type":"ResultTask","Task End Reason":{"Reason":"Success"},"Task Info":{"Task ID":43,"Index":11,"Attempt":0,"Launch Time":1594091481123,"Executor ID":"0","Host":"127.0.0.1","Locality":"PROCESS_LOCAL","Speculative":false,"Getting Result Time":0,"Finish Time":1594091483029,"Failed":false,"Killed":false,"Accumulables":[{"ID":0,"Name":"duration","Update":"1578","Value":"17133","Internal":true,"Count Failed Values":true,"Metadata":"sql"},{"ID":1,"Name":"number of output rows","Update":"375000","Value":"4125000","Internal":true,"Count Failed Values":true,"Metadata":"sql"},{"ID":14,"Name":"number of output rows","Update":"6250","Value":"68750","Internal":true,"Count Failed Values":true,"Metadata":"sql"},{"ID":74,"Name":"internal.metrics.peakExecutionMemory","Update":24040,"Value":264440,"Internal":true,"Count Failed Values":true},{"ID":70,"Name":"internal.metrics.jvmGCTime","Update":66,"Value":680,"Internal":true,"Count
Failed Values":true},{"ID":69,"Name":"internal.metrics.resultSize","Update":2115,"Value":23308,"Internal":true,"Count Failed Values":true},{"ID":68,"Name":"internal.metrics.executorCpuTime","Update":502120000,"Value":6326388000,"Internal":true,"Count Failed Values":true},{"ID":67,"Name":"internal.metrics.executorRunTime","Update":1609,"Value":17480,"Internal":true,"Count Failed Values":true},{"ID":66,"Name":"internal.metrics.executorDeserializeCpuTime","Update":48849000,"Value":705682000,"Internal":true,"Count Failed Values":true},{"ID":65,"Name":"internal.metrics.executorDeserializeTime","Update":225,"Value":2725,"Internal":true,"Count Failed Values":true}]},"Task Executor Metrics":{"JVMHeapMemory":0,"JVMOffHeapMemory":0,"OnHeapExecutionMemory":0,"OffHeapExecutionMemory":0,"OnHeapStorageMemory":0,"OffHeapStorageMemory":0,"OnHeapUnifiedMemory":0,"OffHeapUnifiedMemory":0,"DirectPoolMemory":0,"MappedPoolMemory":0,"ProcessTreeJVMVMemory":0,"ProcessTreeJVMRSSMemory":0,"ProcessTreePytho
nVMemory":0,"ProcessTreePythonRSSMemory":0,"ProcessTreeOtherVMemory":0,"ProcessTreeOtherRSSMemory":0,"MinorGCCount":0,"MinorGCTime":0,"MajorGCCount":0,"MajorGCTime":0},"Task Metrics":{"Executor Deserialize Time":225,"Executor Deserialize CPU Time":48849000,"Executor Run Time":1609,"Executor CPU Time":502120000,"Peak Execution Memory":24040,"Result Size":2115,"JVM GC Time":66,"Result Serialization Time":0,"Memory Bytes Spilled":0,"Disk Bytes Spilled":0,"Shuffle Read Metrics":{"Remote Blocks Fetched":0,"Local Blocks Fetched":0,"Fetch Wait Time":0,"Remote Bytes Read":0,"Remote Bytes Read To Disk":0,"Local Bytes Read":0,"Total Records Read":0},"Shuffle Write Metrics":{"Shuffle Bytes Written":0,"Shuffle Write Time":0,"Shuffle Records Written":0},"Input Metrics":{"Bytes Read":0,"Records Read":0},"Output Metrics":{"Bytes Written":0,"Records Written":0},"Updated Blocks":[]}}
-{"Event":"SparkListenerTaskEnd","Stage ID":2,"Stage Attempt ID":0,"Task Type":"ResultTask","Task End Reason":{"Reason":"Success"},"Task Info":{"Task ID":41,"Index":9,"Attempt":0,"Launch Time":1594091481116,"Executor ID":"0","Host":"127.0.0.1","Locality":"PROCESS_LOCAL","Speculative":false,"Getting Result Time":0,"Finish Time":1594091483032,"Failed":false,"Killed":false,"Accumulables":[{"ID":0,"Name":"duration","Update":"1583","Value":"18716","Internal":true,"Count Failed Values":true,"Metadata":"sql"},{"ID":1,"Name":"number of output rows","Update":"375000","Value":"4500000","Internal":true,"Count Failed Values":true,"Metadata":"sql"},{"ID":14,"Name":"number of output rows","Update":"6250","Value":"75000","Internal":true,"Count Failed Values":true,"Metadata":"sql"},{"ID":74,"Name":"internal.metrics.peakExecutionMemory","Update":24040,"Value":288480,"Internal":true,"Count Failed Values":true},{"ID":70,"Name":"internal.metrics.jvmGCTime","Update":66,"Value":746,"Internal":true,"Count
Failed Values":true},{"ID":69,"Name":"internal.metrics.resultSize","Update":2115,"Value":25423,"Internal":true,"Count Failed Values":true},{"ID":68,"Name":"internal.metrics.executorCpuTime","Update":489923000,"Value":6816311000,"Internal":true,"Count Failed Values":true},{"ID":67,"Name":"internal.metrics.executorRunTime","Update":1614,"Value":19094,"Internal":true,"Count Failed Values":true},{"ID":66,"Name":"internal.metrics.executorDeserializeCpuTime","Update":55787000,"Value":761469000,"Internal":true,"Count Failed Values":true},{"ID":65,"Name":"internal.metrics.executorDeserializeTime","Update":240,"Value":2965,"Internal":true,"Count Failed Values":true}]},"Task Executor Metrics":{"JVMHeapMemory":0,"JVMOffHeapMemory":0,"OnHeapExecutionMemory":0,"OffHeapExecutionMemory":0,"OnHeapStorageMemory":0,"OffHeapStorageMemory":0,"OnHeapUnifiedMemory":0,"OffHeapUnifiedMemory":0,"DirectPoolMemory":0,"MappedPoolMemory":0,"ProcessTreeJVMVMemory":0,"ProcessTreeJVMRSSMemory":0,"ProcessTreePython
VMemory":0,"ProcessTreePythonRSSMemory":0,"ProcessTreeOtherVMemory":0,"ProcessTreeOtherRSSMemory":0,"MinorGCCount":0,"MinorGCTime":0,"MajorGCCount":0,"MajorGCTime":0},"Task Metrics":{"Executor Deserialize Time":240,"Executor Deserialize CPU Time":55787000,"Executor Run Time":1614,"Executor CPU Time":489923000,"Peak Execution Memory":24040,"Result Size":2115,"JVM GC Time":66,"Result Serialization Time":0,"Memory Bytes Spilled":0,"Disk Bytes Spilled":0,"Shuffle Read Metrics":{"Remote Blocks Fetched":0,"Local Blocks Fetched":0,"Fetch Wait Time":0,"Remote Bytes Read":0,"Remote Bytes Read To Disk":0,"Local Bytes Read":0,"Total Records Read":0},"Shuffle Write Metrics":{"Shuffle Bytes Written":0,"Shuffle Write Time":0,"Shuffle Records Written":0},"Input Metrics":{"Bytes Read":0,"Records Read":0},"Output Metrics":{"Bytes Written":0,"Records Written":0},"Updated Blocks":[]}}
-{"Event":"SparkListenerTaskEnd","Stage ID":2,"Stage Attempt ID":0,"Task Type":"ResultTask","Task End Reason":{"Reason":"Success"},"Task Info":{"Task ID":46,"Index":14,"Attempt":0,"Launch Time":1594091481132,"Executor ID":"0","Host":"127.0.0.1","Locality":"PROCESS_LOCAL","Speculative":false,"Getting Result Time":0,"Finish Time":1594091483037,"Failed":false,"Killed":false,"Accumulables":[{"ID":0,"Name":"duration","Update":"1587","Value":"20303","Internal":true,"Count Failed Values":true,"Metadata":"sql"},{"ID":1,"Name":"number of output rows","Update":"375000","Value":"4875000","Internal":true,"Count Failed Values":true,"Metadata":"sql"},{"ID":14,"Name":"number of output rows","Update":"6250","Value":"81250","Internal":true,"Count Failed Values":true,"Metadata":"sql"},{"ID":74,"Name":"internal.metrics.peakExecutionMemory","Update":24040,"Value":312520,"Internal":true,"Count Failed Values":true},{"ID":70,"Name":"internal.metrics.jvmGCTime","Update":66,"Value":812,"Internal":true,"Count
Failed Values":true},{"ID":69,"Name":"internal.metrics.resultSize","Update":2115,"Value":27538,"Internal":true,"Count Failed Values":true},{"ID":68,"Name":"internal.metrics.executorCpuTime","Update":490927000,"Value":7307238000,"Internal":true,"Count Failed Values":true},{"ID":67,"Name":"internal.metrics.executorRunTime","Update":1618,"Value":20712,"Internal":true,"Count Failed Values":true},{"ID":66,"Name":"internal.metrics.executorDeserializeCpuTime","Update":51464000,"Value":812933000,"Internal":true,"Count Failed Values":true},{"ID":65,"Name":"internal.metrics.executorDeserializeTime","Update":218,"Value":3183,"Internal":true,"Count Failed Values":true}]},"Task Executor Metrics":{"JVMHeapMemory":0,"JVMOffHeapMemory":0,"OnHeapExecutionMemory":0,"OffHeapExecutionMemory":0,"OnHeapStorageMemory":0,"OffHeapStorageMemory":0,"OnHeapUnifiedMemory":0,"OffHeapUnifiedMemory":0,"DirectPoolMemory":0,"MappedPoolMemory":0,"ProcessTreeJVMVMemory":0,"ProcessTreeJVMRSSMemory":0,"ProcessTreePytho
nVMemory":0,"ProcessTreePythonRSSMemory":0,"ProcessTreeOtherVMemory":0,"ProcessTreeOtherRSSMemory":0,"MinorGCCount":0,"MinorGCTime":0,"MajorGCCount":0,"MajorGCTime":0},"Task Metrics":{"Executor Deserialize Time":218,"Executor Deserialize CPU Time":51464000,"Executor Run Time":1618,"Executor CPU Time":490927000,"Peak Execution Memory":24040,"Result Size":2115,"JVM GC Time":66,"Result Serialization Time":0,"Memory Bytes Spilled":0,"Disk Bytes Spilled":0,"Shuffle Read Metrics":{"Remote Blocks Fetched":0,"Local Blocks Fetched":0,"Fetch Wait Time":0,"Remote Bytes Read":0,"Remote Bytes Read To Disk":0,"Local Bytes Read":0,"Total Records Read":0},"Shuffle Write Metrics":{"Shuffle Bytes Written":0,"Shuffle Write Time":0,"Shuffle Records Written":0},"Input Metrics":{"Bytes Read":0,"Records Read":0},"Output Metrics":{"Bytes Written":0,"Records Written":0},"Updated Blocks":[]}}
-{"Event":"SparkListenerTaskEnd","Stage ID":2,"Stage Attempt ID":0,"Task Type":"ResultTask","Task End Reason":{"Reason":"Success"},"Task Info":{"Task ID":32,"Index":0,"Attempt":0,"Launch Time":1594091481077,"Executor ID":"0","Host":"127.0.0.1","Locality":"PROCESS_LOCAL","Speculative":false,"Getting Result Time":0,"Finish Time":1594091483037,"Failed":false,"Killed":false,"Accumulables":[{"ID":0,"Name":"duration","Update":"1587","Value":"21890","Internal":true,"Count Failed Values":true,"Metadata":"sql"},{"ID":1,"Name":"number of output rows","Update":"375000","Value":"5250000","Internal":true,"Count Failed Values":true,"Metadata":"sql"},{"ID":14,"Name":"number of output rows","Update":"6250","Value":"87500","Internal":true,"Count Failed Values":true,"Metadata":"sql"},{"ID":74,"Name":"internal.metrics.peakExecutionMemory","Update":24040,"Value":336560,"Internal":true,"Count Failed Values":true},{"ID":70,"Name":"internal.metrics.jvmGCTime","Update":66,"Value":878,"Internal":true,"Count
Failed Values":true},{"ID":69,"Name":"internal.metrics.resultSize","Update":2115,"Value":29653,"Internal":true,"Count Failed Values":true},{"ID":68,"Name":"internal.metrics.executorCpuTime","Update":496683000,"Value":7803921000,"Internal":true,"Count Failed Values":true},{"ID":67,"Name":"internal.metrics.executorRunTime","Update":1619,"Value":22331,"Internal":true,"Count Failed Values":true},{"ID":66,"Name":"internal.metrics.executorDeserializeCpuTime","Update":56827000,"Value":869760000,"Internal":true,"Count Failed Values":true},{"ID":65,"Name":"internal.metrics.executorDeserializeTime","Update":271,"Value":3454,"Internal":true,"Count Failed Values":true}]},"Task Executor Metrics":{"JVMHeapMemory":0,"JVMOffHeapMemory":0,"OnHeapExecutionMemory":0,"OffHeapExecutionMemory":0,"OnHeapStorageMemory":0,"OffHeapStorageMemory":0,"OnHeapUnifiedMemory":0,"OffHeapUnifiedMemory":0,"DirectPoolMemory":0,"MappedPoolMemory":0,"ProcessTreeJVMVMemory":0,"ProcessTreeJVMRSSMemory":0,"ProcessTreePython
VMemory":0,"ProcessTreePythonRSSMemory":0,"ProcessTreeOtherVMemory":0,"ProcessTreeOtherRSSMemory":0,"MinorGCCount":0,"MinorGCTime":0,"MajorGCCount":0,"MajorGCTime":0},"Task Metrics":{"Executor Deserialize Time":271,"Executor Deserialize CPU Time":56827000,"Executor Run Time":1619,"Executor CPU Time":496683000,"Peak Execution Memory":24040,"Result Size":2115,"JVM GC Time":66,"Result Serialization Time":0,"Memory Bytes Spilled":0,"Disk Bytes Spilled":0,"Shuffle Read Metrics":{"Remote Blocks Fetched":0,"Local Blocks Fetched":0,"Fetch Wait Time":0,"Remote Bytes Read":0,"Remote Bytes Read To Disk":0,"Local Bytes Read":0,"Total Records Read":0},"Shuffle Write Metrics":{"Shuffle Bytes Written":0,"Shuffle Write Time":0,"Shuffle Records Written":0},"Input Metrics":{"Bytes Read":0,"Records Read":0},"Output Metrics":{"Bytes Written":0,"Records Written":0},"Updated Blocks":[]}}
-{"Event":"SparkListenerTaskEnd","Stage ID":2,"Stage Attempt ID":0,"Task Type":"ResultTask","Task End Reason":{"Reason":"Success"},"Task Info":{"Task ID":42,"Index":10,"Attempt":0,"Launch Time":1594091481120,"Executor ID":"0","Host":"127.0.0.1","Locality":"PROCESS_LOCAL","Speculative":false,"Getting Result Time":0,"Finish Time":1594091483043,"Failed":false,"Killed":false,"Accumulables":[{"ID":0,"Name":"duration","Update":"1593","Value":"23483","Internal":true,"Count Failed Values":true,"Metadata":"sql"},{"ID":1,"Name":"number of output rows","Update":"375000","Value":"5625000","Internal":true,"Count Failed Values":true,"Metadata":"sql"},{"ID":14,"Name":"number of output rows","Update":"6250","Value":"93750","Internal":true,"Count Failed Values":true,"Metadata":"sql"},{"ID":74,"Name":"internal.metrics.peakExecutionMemory","Update":24040,"Value":360600,"Internal":true,"Count Failed Values":true},{"ID":70,"Name":"internal.metrics.jvmGCTime","Update":66,"Value":944,"Internal":true,"Count
Failed Values":true},{"ID":69,"Name":"internal.metrics.resultSize","Update":2115,"Value":31768,"Internal":true,"Count Failed Values":true},{"ID":68,"Name":"internal.metrics.executorCpuTime","Update":508230000,"Value":8312151000,"Internal":true,"Count Failed Values":true},{"ID":67,"Name":"internal.metrics.executorRunTime","Update":1624,"Value":23955,"Internal":true,"Count Failed Values":true},{"ID":66,"Name":"internal.metrics.executorDeserializeCpuTime","Update":58152000,"Value":927912000,"Internal":true,"Count Failed Values":true},{"ID":65,"Name":"internal.metrics.executorDeserializeTime","Update":229,"Value":3683,"Internal":true,"Count Failed Values":true}]},"Task Executor Metrics":{"JVMHeapMemory":0,"JVMOffHeapMemory":0,"OnHeapExecutionMemory":0,"OffHeapExecutionMemory":0,"OnHeapStorageMemory":0,"OffHeapStorageMemory":0,"OnHeapUnifiedMemory":0,"OffHeapUnifiedMemory":0,"DirectPoolMemory":0,"MappedPoolMemory":0,"ProcessTreeJVMVMemory":0,"ProcessTreeJVMRSSMemory":0,"ProcessTreePytho
nVMemory":0,"ProcessTreePythonRSSMemory":0,"ProcessTreeOtherVMemory":0,"ProcessTreeOtherRSSMemory":0,"MinorGCCount":0,"MinorGCTime":0,"MajorGCCount":0,"MajorGCTime":0},"Task Metrics":{"Executor Deserialize Time":229,"Executor Deserialize CPU Time":58152000,"Executor Run Time":1624,"Executor CPU Time":508230000,"Peak Execution Memory":24040,"Result Size":2115,"JVM GC Time":66,"Result Serialization Time":0,"Memory Bytes Spilled":0,"Disk Bytes Spilled":0,"Shuffle Read Metrics":{"Remote Blocks Fetched":0,"Local Blocks Fetched":0,"Fetch Wait Time":0,"Remote Bytes Read":0,"Remote Bytes Read To Disk":0,"Local Bytes Read":0,"Total Records Read":0},"Shuffle Write Metrics":{"Shuffle Bytes Written":0,"Shuffle Write Time":0,"Shuffle Records Written":0},"Input Metrics":{"Bytes Read":0,"Records Read":0},"Output Metrics":{"Bytes Written":0,"Records Written":0},"Updated Blocks":[]}}
-{"Event":"SparkListenerTaskEnd","Stage ID":2,"Stage Attempt ID":0,"Task Type":"ResultTask","Task End Reason":{"Reason":"Success"},"Task Info":{"Task ID":44,"Index":12,"Attempt":0,"Launch Time":1594091481126,"Executor ID":"0","Host":"127.0.0.1","Locality":"PROCESS_LOCAL","Speculative":false,"Getting Result Time":0,"Finish Time":1594091483043,"Failed":false,"Killed":false,"Accumulables":[{"ID":0,"Name":"duration","Update":"1593","Value":"25076","Internal":true,"Count Failed Values":true,"Metadata":"sql"},{"ID":1,"Name":"number of output rows","Update":"375000","Value":"6000000","Internal":true,"Count Failed Values":true,"Metadata":"sql"},{"ID":14,"Name":"number of output rows","Update":"6250","Value":"100000","Internal":true,"Count Failed Values":true,"Metadata":"sql"},{"ID":74,"Name":"internal.metrics.peakExecutionMemory","Update":24040,"Value":384640,"Internal":true,"Count Failed Values":true},{"ID":70,"Name":"internal.metrics.jvmGCTime","Update":66,"Value":1010,"Internal":true,"Cou
nt Failed Values":true},{"ID":69,"Name":"internal.metrics.resultSize","Update":2115,"Value":33883,"Internal":true,"Count Failed Values":true},{"ID":68,"Name":"internal.metrics.executorCpuTime","Update":498187000,"Value":8810338000,"Internal":true,"Count Failed Values":true},{"ID":67,"Name":"internal.metrics.executorRunTime","Update":1624,"Value":25579,"Internal":true,"Count Failed Values":true},{"ID":66,"Name":"internal.metrics.executorDeserializeCpuTime","Update":51988000,"Value":979900000,"Internal":true,"Count Failed Values":true},{"ID":65,"Name":"internal.metrics.executorDeserializeTime","Update":222,"Value":3905,"Internal":true,"Count Failed Values":true}]},"Task Executor Metrics":{"JVMHeapMemory":0,"JVMOffHeapMemory":0,"OnHeapExecutionMemory":0,"OffHeapExecutionMemory":0,"OnHeapStorageMemory":0,"OffHeapStorageMemory":0,"OnHeapUnifiedMemory":0,"OffHeapUnifiedMemory":0,"DirectPoolMemory":0,"MappedPoolMemory":0,"ProcessTreeJVMVMemory":0,"ProcessTreeJVMRSSMemory":0,"ProcessTreePyt
honVMemory":0,"ProcessTreePythonRSSMemory":0,"ProcessTreeOtherVMemory":0,"ProcessTreeOtherRSSMemory":0,"MinorGCCount":0,"MinorGCTime":0,"MajorGCCount":0,"MajorGCTime":0},"Task Metrics":{"Executor Deserialize Time":222,"Executor Deserialize CPU Time":51988000,"Executor Run Time":1624,"Executor CPU Time":498187000,"Peak Execution Memory":24040,"Result Size":2115,"JVM GC Time":66,"Result Serialization Time":0,"Memory Bytes Spilled":0,"Disk Bytes Spilled":0,"Shuffle Read Metrics":{"Remote Blocks Fetched":0,"Local Blocks Fetched":0,"Fetch Wait Time":0,"Remote Bytes Read":0,"Remote Bytes Read To Disk":0,"Local Bytes Read":0,"Total Records Read":0},"Shuffle Write Metrics":{"Shuffle Bytes Written":0,"Shuffle Write Time":0,"Shuffle Records Written":0},"Input Metrics":{"Bytes Read":0,"Records Read":0},"Output Metrics":{"Bytes Written":0,"Records Written":0},"Updated Blocks":[]}}
+{"Event":"SparkListenerTaskEnd","Stage ID":2,"Stage Attempt ID":0,"Task Type":"ResultTask","Task End Reason":{"Reason":"Success"},"Task Info":{"Task ID":38,"Index":6,"Attempt":0,"Launch Time":1594091481104,"Executor ID":"0","Host":"127.0.0.1","Locality":"PROCESS_LOCAL","Speculative":false,"Getting Result Time":0,"Finish Time":1594091482939,"Failed":false,"Killed":false,"Accumulables":[{"ID":0,"Name":"duration","Update":"1467","Value":"1467","Internal":true,"Count Failed Values":true,"Metadata":"sql"},{"ID":1,"Name":"number of output rows","Update":"375000","Value":"375000","Internal":true,"Count Failed Values":true,"Metadata":"sql"},{"ID":14,"Name":"number of output rows","Update":"6250","Value":"6250","Internal":true,"Count Failed Values":true,"Metadata":"sql"},{"ID":74,"Name":"internal.metrics.peakExecutionMemory","Update":24040,"Value":24040,"Internal":true,"Count Failed Values":true},{"ID":71,"Name":"internal.metrics.resultSerializationTime","Update":11,"Value":11,"Internal":tru
e,"Count Failed Values":true},{"ID":70,"Name":"internal.metrics.jvmGCTime","Update":62,"Value":62,"Internal":true,"Count Failed Values":true},{"ID":69,"Name":"internal.metrics.resultSize","Update":2158,"Value":2158,"Internal":true,"Count Failed Values":true},{"ID":68,"Name":"internal.metrics.executorCpuTime","Update":414110000,"Value":414110000,"Internal":true,"Count Failed Values":true},{"ID":67,"Name":"internal.metrics.executorRunTime","Update":1498,"Value":1498,"Internal":true,"Count Failed Values":true},{"ID":66,"Name":"internal.metrics.executorDeserializeCpuTime","Update":60358000,"Value":60358000,"Internal":true,"Count Failed Values":true},{"ID":65,"Name":"internal.metrics.executorDeserializeTime","Update":255,"Value":255,"Internal":true,"Count Failed Values":true}]},"Task Executor Metrics":{"JVMHeapMemory":0,"JVMOffHeapMemory":0,"OnHeapExecutionMemory":0,"OffHeapExecutionMemory":0,"OnHeapStorageMemory":0,"OffHeapStorageMemory":0,"OnHeapUnifiedMemory":0,"OffHeapUnifiedMemory":
0,"DirectPoolMemory":0,"MappedPoolMemory":0,"ProcessTreeJVMVMemory":0,"ProcessTreeJVMRSSMemory":0,"ProcessTreePythonVMemory":0,"ProcessTreePythonRSSMemory":0,"ProcessTreeOtherVMemory":0,"ProcessTreeOtherRSSMemory":0,"MinorGCCount":0,"MinorGCTime":0,"MajorGCCount":0,"MajorGCTime":0},"Task Metrics":{"Executor Deserialize Time":255,"Executor Deserialize CPU Time":60358000,"Executor Run Time":1498,"Executor CPU Time":414110000,"Peak Execution Memory":24040,"Result Size":2158,"JVM GC Time":62,"Result Serialization Time":11,"Memory Bytes Spilled":0,"Disk Bytes Spilled":0,"Shuffle Read Metrics":{"Remote Blocks Fetched":0,"Local Blocks Fetched":0,"Fetch Wait Time":0,"Remote Bytes Read":0,"Remote Bytes Read To Disk":0,"Local Bytes Read":0,"Total Records Read":0,"Shuffle Push Read Metrics":{"Corrupt Merged Block Chunks":10,"Merged Fallback Count":2,"Merged Remote Blocks Fetched":10,"Merged Local Blocks Fetched":15,"Merged Remote Chunks Fetched":12,"Merged Local Chunks Fetched":10,"Merged Remo
te Bytes Read":100,"Merged Local Bytes Read":100}},"Shuffle Write Metrics":{"Shuffle Bytes Written":0,"Shuffle Write Time":0,"Shuffle Records Written":0},"Input Metrics":{"Bytes Read":0,"Records Read":0},"Output Metrics":{"Bytes Written":0,"Records Written":0},"Updated Blocks":[]}}
+{"Event":"SparkListenerTaskEnd","Stage ID":2,"Stage Attempt ID":0,"Task Type":"ResultTask","Task End Reason":{"Reason":"Success"},"Task Info":{"Task ID":47,"Index":15,"Attempt":0,"Launch Time":1594091481136,"Executor ID":"0","Host":"127.0.0.1","Locality":"PROCESS_LOCAL","Speculative":false,"Getting Result Time":0,"Finish Time":1594091483014,"Failed":false,"Killed":false,"Accumulables":[{"ID":0,"Name":"duration","Update":"1563","Value":"3030","Internal":true,"Count Failed Values":true,"Metadata":"sql"},{"ID":1,"Name":"number of output rows","Update":"375000","Value":"750000","Internal":true,"Count Failed Values":true,"Metadata":"sql"},{"ID":14,"Name":"number of output rows","Update":"6250","Value":"12500","Internal":true,"Count Failed Values":true,"Metadata":"sql"},{"ID":74,"Name":"internal.metrics.peakExecutionMemory","Update":24040,"Value":48080,"Internal":true,"Count Failed Values":true},{"ID":70,"Name":"internal.metrics.jvmGCTime","Update":52,"Value":114,"Internal":true,"Count Fa
iled Values":true},{"ID":69,"Name":"internal.metrics.resultSize","Update":2115,"Value":4273,"Internal":true,"Count Failed Values":true},{"ID":68,"Name":"internal.metrics.executorCpuTime","Update":1324251000,"Value":1738361000,"Internal":true,"Count Failed Values":true},{"ID":67,"Name":"internal.metrics.executorRunTime","Update":1594,"Value":3092,"Internal":true,"Count Failed Values":true},{"ID":66,"Name":"internal.metrics.executorDeserializeCpuTime","Update":47496000,"Value":107854000,"Internal":true,"Count Failed Values":true},{"ID":65,"Name":"internal.metrics.executorDeserializeTime","Update":213,"Value":468,"Internal":true,"Count Failed Values":true}]},"Task Executor Metrics":{"JVMHeapMemory":0,"JVMOffHeapMemory":0,"OnHeapExecutionMemory":0,"OffHeapExecutionMemory":0,"OnHeapStorageMemory":0,"OffHeapStorageMemory":0,"OnHeapUnifiedMemory":0,"OffHeapUnifiedMemory":0,"DirectPoolMemory":0,"MappedPoolMemory":0,"ProcessTreeJVMVMemory":0,"ProcessTreeJVMRSSMemory":0,"ProcessTreePythonVMem
ory":0,"ProcessTreePythonRSSMemory":0,"ProcessTreeOtherVMemory":0,"ProcessTreeOtherRSSMemory":0,"MinorGCCount":0,"MinorGCTime":0,"MajorGCCount":0,"MajorGCTime":0},"Task Metrics":{"Executor Deserialize Time":213,"Executor Deserialize CPU Time":47496000,"Executor Run Time":1594,"Executor CPU Time":1324251000,"Peak Execution Memory":24040,"Result Size":2115,"JVM GC Time":52,"Result Serialization Time":0,"Memory Bytes Spilled":0,"Disk Bytes Spilled":0,"Shuffle Read Metrics":{"Remote Blocks Fetched":0,"Local Blocks Fetched":0,"Fetch Wait Time":0,"Remote Bytes Read":0,"Remote Bytes Read To Disk":0,"Local Bytes Read":0,"Total Records Read":0,"Shuffle Push Read Metrics":{"Corrupt Merged Block Chunks":10,"Merged Fallback Count":2,"Merged Remote Blocks Fetched":10,"Merged Local Blocks Fetched":15,"Merged Remote Chunks Fetched":12,"Merged Local Chunks Fetched":10,"Merged Remote Bytes Read":100,"Merged Local Bytes Read":100}},"Shuffle Write Metrics":{"Shuffle Bytes Written":0,"Shuffle Write Tim
e":0,"Shuffle Records Written":0},"Input Metrics":{"Bytes Read":0,"Records Read":0},"Output Metrics":{"Bytes Written":0,"Records Written":0},"Updated Blocks":[]}}
+{"Event":"SparkListenerTaskEnd","Stage ID":2,"Stage Attempt ID":0,"Task Type":"ResultTask","Task End Reason":{"Reason":"Success"},"Task Info":{"Task ID":37,"Index":5,"Attempt":0,"Launch Time":1594091481100,"Executor ID":"0","Host":"127.0.0.1","Locality":"PROCESS_LOCAL","Speculative":false,"Getting Result Time":0,"Finish Time":1594091483015,"Failed":false,"Killed":false,"Accumulables":[{"ID":0,"Name":"duration","Update":"1563","Value":"4593","Internal":true,"Count Failed Values":true,"Metadata":"sql"},{"ID":1,"Name":"number of output rows","Update":"375000","Value":"1125000","Internal":true,"Count Failed Values":true,"Metadata":"sql"},{"ID":14,"Name":"number of output rows","Update":"6250","Value":"18750","Internal":true,"Count Failed Values":true,"Metadata":"sql"},{"ID":74,"Name":"internal.metrics.peakExecutionMemory","Update":24040,"Value":72120,"Internal":true,"Count Failed Values":true},{"ID":70,"Name":"internal.metrics.jvmGCTime","Update":62,"Value":176,"Internal":true,"Count Fa
iled Values":true},{"ID":69,"Name":"internal.metrics.resultSize","Update":2115,"Value":6388,"Internal":true,"Count Failed Values":true},{"ID":68,"Name":"internal.metrics.executorCpuTime","Update":507192000,"Value":2245553000,"Internal":true,"Count Failed Values":true},{"ID":67,"Name":"internal.metrics.executorRunTime","Update":1596,"Value":4688,"Internal":true,"Count Failed Values":true},{"ID":66,"Name":"internal.metrics.executorDeserializeCpuTime","Update":60890000,"Value":168744000,"Internal":true,"Count Failed Values":true},{"ID":65,"Name":"internal.metrics.executorDeserializeTime","Update":256,"Value":724,"Internal":true,"Count Failed Values":true}]},"Task Executor Metrics":{"JVMHeapMemory":0,"JVMOffHeapMemory":0,"OnHeapExecutionMemory":0,"OffHeapExecutionMemory":0,"OnHeapStorageMemory":0,"OffHeapStorageMemory":0,"OnHeapUnifiedMemory":0,"OffHeapUnifiedMemory":0,"DirectPoolMemory":0,"MappedPoolMemory":0,"ProcessTreeJVMVMemory":0,"ProcessTreeJVMRSSMemory":0,"ProcessTreePythonVMemo
ry":0,"ProcessTreePythonRSSMemory":0,"ProcessTreeOtherVMemory":0,"ProcessTreeOtherRSSMemory":0,"MinorGCCount":0,"MinorGCTime":0,"MajorGCCount":0,"MajorGCTime":0},"Task Metrics":{"Executor Deserialize Time":256,"Executor Deserialize CPU Time":60890000,"Executor Run Time":1596,"Executor CPU Time":507192000,"Peak Execution Memory":24040,"Result Size":2115,"JVM GC Time":62,"Result Serialization Time":0,"Memory Bytes Spilled":0,"Disk Bytes Spilled":0,"Shuffle Read Metrics":{"Remote Blocks Fetched":0,"Local Blocks Fetched":0,"Fetch Wait Time":0,"Remote Bytes Read":0,"Remote Bytes Read To Disk":0,"Local Bytes Read":0,"Total Records Read":0,"Shuffle Push Read Metrics":{"Corrupt Merged Block Chunks":10,"Merged Fallback Count":2,"Merged Remote Blocks Fetched":10,"Merged Local Blocks Fetched":15,"Merged Remote Chunks Fetched":12,"Merged Local Chunks Fetched":10,"Merged Remote Bytes Read":100,"Merged Local Bytes Read":100}},"Shuffle Write Metrics":{"Shuffle Bytes Written":0,"Shuffle Write Time"
:0,"Shuffle Records Written":0},"Input Metrics":{"Bytes Read":0,"Records Read":0},"Output Metrics":{"Bytes Written":0,"Records Written":0},"Updated Blocks":[]}}
+{"Event":"SparkListenerTaskEnd","Stage ID":2,"Stage Attempt ID":0,"Task Type":"ResultTask","Task End Reason":{"Reason":"Success"},"Task Info":{"Task ID":36,"Index":4,"Attempt":0,"Launch Time":1594091481095,"Executor ID":"0","Host":"127.0.0.1","Locality":"PROCESS_LOCAL","Speculative":false,"Getting Result Time":0,"Finish Time":1594091483015,"Failed":false,"Killed":false,"Accumulables":[{"ID":0,"Name":"duration","Update":"1564","Value":"6157","Internal":true,"Count Failed Values":true,"Metadata":"sql"},{"ID":1,"Name":"number of output rows","Update":"375000","Value":"1500000","Internal":true,"Count Failed Values":true,"Metadata":"sql"},{"ID":14,"Name":"number of output rows","Update":"6250","Value":"25000","Internal":true,"Count Failed Values":true,"Metadata":"sql"},{"ID":74,"Name":"internal.metrics.peakExecutionMemory","Update":24040,"Value":96160,"Internal":true,"Count Failed Values":true},{"ID":70,"Name":"internal.metrics.jvmGCTime","Update":62,"Value":238,"Internal":true,"Count Fa
iled Values":true},{"ID":69,"Name":"internal.metrics.resultSize","Update":2115,"Value":8503,"Internal":true,"Count Failed Values":true},{"ID":68,"Name":"internal.metrics.executorCpuTime","Update":503010000,"Value":2748563000,"Internal":true,"Count Failed Values":true},{"ID":67,"Name":"internal.metrics.executorRunTime","Update":1596,"Value":6284,"Internal":true,"Count Failed Values":true},{"ID":66,"Name":"internal.metrics.executorDeserializeCpuTime","Update":112849000,"Value":281593000,"Internal":true,"Count Failed Values":true},{"ID":65,"Name":"internal.metrics.executorDeserializeTime","Update":260,"Value":984,"Internal":true,"Count Failed Values":true}]},"Task Executor Metrics":{"JVMHeapMemory":0,"JVMOffHeapMemory":0,"OnHeapExecutionMemory":0,"OffHeapExecutionMemory":0,"OnHeapStorageMemory":0,"OffHeapStorageMemory":0,"OnHeapUnifiedMemory":0,"OffHeapUnifiedMemory":0,"DirectPoolMemory":0,"MappedPoolMemory":0,"ProcessTreeJVMVMemory":0,"ProcessTreeJVMRSSMemory":0,"ProcessTreePythonVMem
ory":0,"ProcessTreePythonRSSMemory":0,"ProcessTreeOtherVMemory":0,"ProcessTreeOtherRSSMemory":0,"MinorGCCount":0,"MinorGCTime":0,"MajorGCCount":0,"MajorGCTime":0},"Task Metrics":{"Executor Deserialize Time":260,"Executor Deserialize CPU Time":112849000,"Executor Run Time":1596,"Executor CPU Time":503010000,"Peak Execution Memory":24040,"Result Size":2115,"JVM GC Time":62,"Result Serialization Time":0,"Memory Bytes Spilled":0,"Disk Bytes Spilled":0,"Shuffle Read Metrics":{"Remote Blocks Fetched":0,"Local Blocks Fetched":0,"Fetch Wait Time":0,"Remote Bytes Read":0,"Remote Bytes Read To Disk":0,"Local Bytes Read":0,"Total Records Read":0,"Shuffle Push Read Metrics":{"Corrupt Merged Block Chunks":10,"Merged Fallback Count":2,"Merged Remote Blocks Fetched":10,"Merged Local Blocks Fetched":15,"Merged Remote Chunks Fetched":12,"Merged Local Chunks Fetched":10,"Merged Remote Bytes Read":100,"Merged Local Bytes Read":100}},"Shuffle Write Metrics":{"Shuffle Bytes Written":0,"Shuffle Write Tim
e":0,"Shuffle Records Written":0},"Input Metrics":{"Bytes Read":0,"Records Read":0},"Output Metrics":{"Bytes Written":0,"Records Written":0},"Updated Blocks":[]}}
+{"Event":"SparkListenerTaskEnd","Stage ID":2,"Stage Attempt ID":0,"Task Type":"ResultTask","Task End Reason":{"Reason":"Success"},"Task Info":{"Task ID":35,"Index":3,"Attempt":0,"Launch Time":1594091481091,"Executor ID":"0","Host":"127.0.0.1","Locality":"PROCESS_LOCAL","Speculative":false,"Getting Result Time":0,"Finish Time":1594091483016,"Failed":false,"Killed":false,"Accumulables":[{"ID":0,"Name":"duration","Update":"1566","Value":"7723","Internal":true,"Count Failed Values":true,"Metadata":"sql"},{"ID":1,"Name":"number of output rows","Update":"375000","Value":"1875000","Internal":true,"Count Failed Values":true,"Metadata":"sql"},{"ID":14,"Name":"number of output rows","Update":"6250","Value":"31250","Internal":true,"Count Failed Values":true,"Metadata":"sql"},{"ID":74,"Name":"internal.metrics.peakExecutionMemory","Update":24040,"Value":120200,"Internal":true,"Count Failed Values":true},{"ID":70,"Name":"internal.metrics.jvmGCTime","Update":62,"Value":300,"Internal":true,"Count F
ailed Values":true},{"ID":69,"Name":"internal.metrics.resultSize","Update":2115,"Value":10618,"Internal":true,"Count Failed Values":true},{"ID":68,"Name":"internal.metrics.executorCpuTime","Update":502908000,"Value":3251471000,"Internal":true,"Count Failed Values":true},{"ID":67,"Name":"internal.metrics.executorRunTime","Update":1598,"Value":7882,"Internal":true,"Count Failed Values":true},{"ID":66,"Name":"internal.metrics.executorDeserializeCpuTime","Update":62944000,"Value":344537000,"Internal":true,"Count Failed Values":true},{"ID":65,"Name":"internal.metrics.executorDeserializeTime","Update":263,"Value":1247,"Internal":true,"Count Failed Values":true}]},"Task Executor Metrics":{"JVMHeapMemory":0,"JVMOffHeapMemory":0,"OnHeapExecutionMemory":0,"OffHeapExecutionMemory":0,"OnHeapStorageMemory":0,"OffHeapStorageMemory":0,"OnHeapUnifiedMemory":0,"OffHeapUnifiedMemory":0,"DirectPoolMemory":0,"MappedPoolMemory":0,"ProcessTreeJVMVMemory":0,"ProcessTreeJVMRSSMemory":0,"ProcessTreePythonVM
emory":0,"ProcessTreePythonRSSMemory":0,"ProcessTreeOtherVMemory":0,"ProcessTreeOtherRSSMemory":0,"MinorGCCount":0,"MinorGCTime":0,"MajorGCCount":0,"MajorGCTime":0},"Task Metrics":{"Executor Deserialize Time":263,"Executor Deserialize CPU Time":62944000,"Executor Run Time":1598,"Executor CPU Time":502908000,"Peak Execution Memory":24040,"Result Size":2115,"JVM GC Time":62,"Result Serialization Time":0,"Memory Bytes Spilled":0,"Disk Bytes Spilled":0,"Shuffle Read Metrics":{"Remote Blocks Fetched":0,"Local Blocks Fetched":0,"Fetch Wait Time":0,"Remote Bytes Read":0,"Remote Bytes Read To Disk":0,"Local Bytes Read":0,"Total Records Read":0,"Shuffle Push Read Metrics":{"Corrupt Merged Block Chunks":10,"Merged Fallback Count":2,"Merged Remote Blocks Fetched":10,"Merged Local Blocks Fetched":15,"Merged Remote Chunks Fetched":12,"Merged Local Chunks Fetched":10,"Merged Remote Bytes Read":100,"Merged Local Bytes Read":100}},"Shuffle Write Metrics":{"Shuffle Bytes Written":0,"Shuffle Write Ti
me":0,"Shuffle Records Written":0},"Input Metrics":{"Bytes Read":0,"Records Read":0},"Output Metrics":{"Bytes Written":0,"Records Written":0},"Updated Blocks":[]}}
+{"Event":"SparkListenerTaskEnd","Stage ID":2,"Stage Attempt ID":0,"Task Type":"ResultTask","Task End Reason":{"Reason":"Success"},"Task Info":{"Task ID":40,"Index":8,"Attempt":0,"Launch Time":1594091481112,"Executor ID":"0","Host":"127.0.0.1","Locality":"PROCESS_LOCAL","Speculative":false,"Getting Result Time":0,"Finish Time":1594091483016,"Failed":false,"Killed":false,"Accumulables":[{"ID":0,"Name":"duration","Update":"1562","Value":"9285","Internal":true,"Count Failed Values":true,"Metadata":"sql"},{"ID":1,"Name":"number of output rows","Update":"375000","Value":"2250000","Internal":true,"Count Failed Values":true,"Metadata":"sql"},{"ID":14,"Name":"number of output rows","Update":"6250","Value":"37500","Internal":true,"Count Failed Values":true,"Metadata":"sql"},{"ID":74,"Name":"internal.metrics.peakExecutionMemory","Update":24040,"Value":144240,"Internal":true,"Count Failed Values":true},{"ID":70,"Name":"internal.metrics.jvmGCTime","Update":62,"Value":362,"Internal":true,"Count F
ailed Values":true},{"ID":69,"Name":"internal.metrics.resultSize","Update":2115,"Value":12733,"Internal":true,"Count Failed Values":true},{"ID":68,"Name":"internal.metrics.executorCpuTime","Update":510597000,"Value":3762068000,"Internal":true,"Count Failed Values":true},{"ID":67,"Name":"internal.metrics.executorRunTime","Update":1595,"Value":9477,"Internal":true,"Count Failed Values":true},{"ID":66,"Name":"internal.metrics.executorDeserializeCpuTime","Update":69760000,"Value":414297000,"Internal":true,"Count Failed Values":true},{"ID":65,"Name":"internal.metrics.executorDeserializeTime","Update":246,"Value":1493,"Internal":true,"Count Failed Values":true}]},"Task Executor Metrics":{"JVMHeapMemory":0,"JVMOffHeapMemory":0,"OnHeapExecutionMemory":0,"OffHeapExecutionMemory":0,"OnHeapStorageMemory":0,"OffHeapStorageMemory":0,"OnHeapUnifiedMemory":0,"OffHeapUnifiedMemory":0,"DirectPoolMemory":0,"MappedPoolMemory":0,"ProcessTreeJVMVMemory":0,"ProcessTreeJVMRSSMemory":0,"ProcessTreePythonVM
emory":0,"ProcessTreePythonRSSMemory":0,"ProcessTreeOtherVMemory":0,"ProcessTreeOtherRSSMemory":0,"MinorGCCount":0,"MinorGCTime":0,"MajorGCCount":0,"MajorGCTime":0},"Task Metrics":{"Executor Deserialize Time":246,"Executor Deserialize CPU Time":69760000,"Executor Run Time":1595,"Executor CPU Time":510597000,"Peak Execution Memory":24040,"Result Size":2115,"JVM GC Time":62,"Result Serialization Time":0,"Memory Bytes Spilled":0,"Disk Bytes Spilled":0,"Shuffle Read Metrics":{"Remote Blocks Fetched":0,"Local Blocks Fetched":0,"Fetch Wait Time":0,"Remote Bytes Read":0,"Remote Bytes Read To Disk":0,"Local Bytes Read":0,"Total Records Read":0,"Shuffle Push Read Metrics":{"Corrupt Merged Block Chunks":10,"Merged Fallback Count":2,"Merged Remote Blocks Fetched":10,"Merged Local Blocks Fetched":15,"Merged Remote Chunks Fetched":12,"Merged Local Chunks Fetched":10,"Merged Remote Bytes Read":100,"Merged Local Bytes Read":100}},"Shuffle Write Metrics":{"Shuffle Bytes Written":0,"Shuffle Write Ti
me":0,"Shuffle Records Written":0},"Input Metrics":{"Bytes Read":0,"Records Read":0},"Output Metrics":{"Bytes Written":0,"Records Written":0},"Updated Blocks":[]}}
+{"Event":"SparkListenerTaskEnd","Stage ID":2,"Stage Attempt ID":0,"Task Type":"ResultTask","Task End Reason":{"Reason":"Success"},"Task Info":{"Task ID":45,"Index":13,"Attempt":0,"Launch Time":1594091481129,"Executor ID":"0","Host":"127.0.0.1","Locality":"PROCESS_LOCAL","Speculative":false,"Getting Result Time":0,"Finish Time":1594091483024,"Failed":false,"Killed":false,"Accumulables":[{"ID":0,"Name":"duration","Update":"1564","Value":"10849","Internal":true,"Count Failed Values":true,"Metadata":"sql"},{"ID":1,"Name":"number of output rows","Update":"375000","Value":"2625000","Internal":true,"Count Failed Values":true,"Metadata":"sql"},{"ID":14,"Name":"number of output rows","Update":"6250","Value":"43750","Internal":true,"Count Failed Values":true,"Metadata":"sql"},{"ID":74,"Name":"internal.metrics.peakExecutionMemory","Update":24040,"Value":168280,"Internal":true,"Count Failed Values":true},{"ID":70,"Name":"internal.metrics.jvmGCTime","Update":62,"Value":424,"Internal":true,"Count
Failed Values":true},{"ID":69,"Name":"internal.metrics.resultSize","Update":2115,"Value":14848,"Internal":true,"Count Failed Values":true},{"ID":68,"Name":"internal.metrics.executorCpuTime","Update":495138000,"Value":4257206000,"Internal":true,"Count Failed Values":true},{"ID":67,"Name":"internal.metrics.executorRunTime","Update":1595,"Value":11072,"Internal":true,"Count Failed Values":true},{"ID":66,"Name":"internal.metrics.executorDeserializeCpuTime","Update":54222000,"Value":468519000,"Internal":true,"Count Failed Values":true},{"ID":65,"Name":"internal.metrics.executorDeserializeTime","Update":221,"Value":1714,"Internal":true,"Count Failed Values":true}]},"Task Executor Metrics":{"JVMHeapMemory":0,"JVMOffHeapMemory":0,"OnHeapExecutionMemory":0,"OffHeapExecutionMemory":0,"OnHeapStorageMemory":0,"OffHeapStorageMemory":0,"OnHeapUnifiedMemory":0,"OffHeapUnifiedMemory":0,"DirectPoolMemory":0,"MappedPoolMemory":0,"ProcessTreeJVMVMemory":0,"ProcessTreeJVMRSSMemory":0,"ProcessTreePytho
nVMemory":0,"ProcessTreePythonRSSMemory":0,"ProcessTreeOtherVMemory":0,"ProcessTreeOtherRSSMemory":0,"MinorGCCount":0,"MinorGCTime":0,"MajorGCCount":0,"MajorGCTime":0},"Task Metrics":{"Executor Deserialize Time":221,"Executor Deserialize CPU Time":54222000,"Executor Run Time":1595,"Executor CPU Time":495138000,"Peak Execution Memory":24040,"Result Size":2115,"JVM GC Time":62,"Result Serialization Time":0,"Memory Bytes Spilled":0,"Disk Bytes Spilled":0,"Shuffle Read Metrics":{"Remote Blocks Fetched":0,"Local Blocks Fetched":0,"Fetch Wait Time":0,"Remote Bytes Read":0,"Remote Bytes Read To Disk":0,"Local Bytes Read":0,"Total Records Read":0,"Shuffle Push Read Metrics":{"Corrupt Merged Block Chunks":10,"Merged Fallback Count":2,"Merged Remote Blocks Fetched":10,"Merged Local Blocks Fetched":15,"Merged Remote Chunks Fetched":12,"Merged Local Chunks Fetched":10,"Merged Remote Bytes Read":100,"Merged Local Bytes Read":100}},"Shuffle Write Metrics":{"Shuffle Bytes Written":0,"Shuffle Write
Time":0,"Shuffle Records Written":0},"Input Metrics":{"Bytes Read":0,"Records Read":0},"Output Metrics":{"Bytes Written":0,"Records Written":0},"Updated Blocks":[]}}
+{"Event":"SparkListenerTaskEnd","Stage ID":2,"Stage Attempt ID":0,"Task Type":"ResultTask","Task End Reason":{"Reason":"Success"},"Task Info":{"Task ID":39,"Index":7,"Attempt":0,"Launch Time":1594091481109,"Executor ID":"0","Host":"127.0.0.1","Locality":"PROCESS_LOCAL","Speculative":false,"Getting Result Time":0,"Finish Time":1594091483024,"Failed":false,"Killed":false,"Accumulables":[{"ID":0,"Name":"duration","Update":"1566","Value":"12415","Internal":true,"Count Failed Values":true,"Metadata":"sql"},{"ID":1,"Name":"number of output rows","Update":"375000","Value":"3000000","Internal":true,"Count Failed Values":true,"Metadata":"sql"},{"ID":14,"Name":"number of output rows","Update":"6250","Value":"50000","Internal":true,"Count Failed Values":true,"Metadata":"sql"},{"ID":74,"Name":"internal.metrics.peakExecutionMemory","Update":24040,"Value":192320,"Internal":true,"Count Failed Values":true},{"ID":70,"Name":"internal.metrics.jvmGCTime","Update":62,"Value":486,"Internal":true,"Count
Failed Values":true},{"ID":69,"Name":"internal.metrics.resultSize","Update":2115,"Value":16963,"Internal":true,"Count Failed Values":true},{"ID":68,"Name":"internal.metrics.executorCpuTime","Update":539451000,"Value":4796657000,"Internal":true,"Count Failed Values":true},{"ID":67,"Name":"internal.metrics.executorRunTime","Update":1596,"Value":12668,"Internal":true,"Count Failed Values":true},{"ID":66,"Name":"internal.metrics.executorDeserializeCpuTime","Update":64380000,"Value":532899000,"Internal":true,"Count Failed Values":true},{"ID":65,"Name":"internal.metrics.executorDeserializeTime","Update":254,"Value":1968,"Internal":true,"Count Failed Values":true}]},"Task Executor Metrics":{"JVMHeapMemory":0,"JVMOffHeapMemory":0,"OnHeapExecutionMemory":0,"OffHeapExecutionMemory":0,"OnHeapStorageMemory":0,"OffHeapStorageMemory":0,"OnHeapUnifiedMemory":0,"OffHeapUnifiedMemory":0,"DirectPoolMemory":0,"MappedPoolMemory":0,"ProcessTreeJVMVMemory":0,"ProcessTreeJVMRSSMemory":0,"ProcessTreePython
VMemory":0,"ProcessTreePythonRSSMemory":0,"ProcessTreeOtherVMemory":0,"ProcessTreeOtherRSSMemory":0,"MinorGCCount":0,"MinorGCTime":0,"MajorGCCount":0,"MajorGCTime":0},"Task Metrics":{"Executor Deserialize Time":254,"Executor Deserialize CPU Time":64380000,"Executor Run Time":1596,"Executor CPU Time":539451000,"Peak Execution Memory":24040,"Result Size":2115,"JVM GC Time":62,"Result Serialization Time":0,"Memory Bytes Spilled":0,"Disk Bytes Spilled":0,"Shuffle Read Metrics":{"Remote Blocks Fetched":0,"Local Blocks Fetched":0,"Fetch Wait Time":0,"Remote Bytes Read":0,"Remote Bytes Read To Disk":0,"Local Bytes Read":0,"Total Records Read":0,"Shuffle Push Read Metrics":{"Corrupt Merged Block Chunks":10,"Merged Fallback Count":2,"Merged Remote Blocks Fetched":10,"Merged Local Blocks Fetched":15,"Merged Remote Chunks Fetched":12,"Merged Local Chunks Fetched":10,"Merged Remote Bytes Read":100,"Merged Local Bytes Read":100}},"Shuffle Write Metrics":{"Shuffle Bytes Written":0,"Shuffle Write
Time":0,"Shuffle Records Written":0},"Input Metrics":{"Bytes Read":0,"Records Read":0},"Output Metrics":{"Bytes Written":0,"Records Written":0},"Updated Blocks":[]}}
+{"Event":"SparkListenerTaskEnd","Stage ID":2,"Stage Attempt ID":0,"Task Type":"ResultTask","Task End Reason":{"Reason":"Success"},"Task Info":{"Task ID":33,"Index":1,"Attempt":0,"Launch Time":1594091481082,"Executor ID":"0","Host":"127.0.0.1","Locality":"PROCESS_LOCAL","Speculative":false,"Getting Result Time":0,"Finish Time":1594091483025,"Failed":false,"Killed":false,"Accumulables":[{"ID":0,"Name":"duration","Update":"1566","Value":"13981","Internal":true,"Count Failed Values":true,"Metadata":"sql"},{"ID":1,"Name":"number of output rows","Update":"375000","Value":"3375000","Internal":true,"Count Failed Values":true,"Metadata":"sql"},{"ID":14,"Name":"number of output rows","Update":"6250","Value":"56250","Internal":true,"Count Failed Values":true,"Metadata":"sql"},{"ID":74,"Name":"internal.metrics.peakExecutionMemory","Update":24040,"Value":216360,"Internal":true,"Count Failed Values":true},{"ID":70,"Name":"internal.metrics.jvmGCTime","Update":62,"Value":548,"Internal":true,"Count
Failed Values":true},{"ID":69,"Name":"internal.metrics.resultSize","Update":2115,"Value":19078,"Internal":true,"Count Failed Values":true},{"ID":68,"Name":"internal.metrics.executorCpuTime","Update":519178000,"Value":5315835000,"Internal":true,"Count Failed Values":true},{"ID":67,"Name":"internal.metrics.executorRunTime","Update":1597,"Value":14265,"Internal":true,"Count Failed Values":true},{"ID":66,"Name":"internal.metrics.executorDeserializeCpuTime","Update":54442000,"Value":587341000,"Internal":true,"Count Failed Values":true},{"ID":65,"Name":"internal.metrics.executorDeserializeTime","Update":267,"Value":2235,"Internal":true,"Count Failed Values":true}]},"Task Executor Metrics":{"JVMHeapMemory":0,"JVMOffHeapMemory":0,"OnHeapExecutionMemory":0,"OffHeapExecutionMemory":0,"OnHeapStorageMemory":0,"OffHeapStorageMemory":0,"OnHeapUnifiedMemory":0,"OffHeapUnifiedMemory":0,"DirectPoolMemory":0,"MappedPoolMemory":0,"ProcessTreeJVMVMemory":0,"ProcessTreeJVMRSSMemory":0,"ProcessTreePython
VMemory":0,"ProcessTreePythonRSSMemory":0,"ProcessTreeOtherVMemory":0,"ProcessTreeOtherRSSMemory":0,"MinorGCCount":0,"MinorGCTime":0,"MajorGCCount":0,"MajorGCTime":0},"Task Metrics":{"Executor Deserialize Time":267,"Executor Deserialize CPU Time":54442000,"Executor Run Time":1597,"Executor CPU Time":519178000,"Peak Execution Memory":24040,"Result Size":2115,"JVM GC Time":62,"Result Serialization Time":0,"Memory Bytes Spilled":0,"Disk Bytes Spilled":0,"Shuffle Read Metrics":{"Remote Blocks Fetched":0,"Local Blocks Fetched":0,"Fetch Wait Time":0,"Remote Bytes Read":0,"Remote Bytes Read To Disk":0,"Local Bytes Read":0,"Total Records Read":0,"Shuffle Push Read Metrics":{"Corrupt Merged Block Chunks":10,"Merged Fallback Count":2,"Merged Remote Blocks Fetched":10,"Merged Local Blocks Fetched":15,"Merged Remote Chunks Fetched":12,"Merged Local Chunks Fetched":10,"Merged Remote Bytes Read":100,"Merged Local Bytes Read":100}},"Shuffle Write Metrics":{"Shuffle Bytes Written":0,"Shuffle Write
Time":0,"Shuffle Records Written":0},"Input Metrics":{"Bytes Read":0,"Records Read":0},"Output Metrics":{"Bytes Written":0,"Records Written":0},"Updated Blocks":[]}}
+{"Event":"SparkListenerTaskEnd","Stage ID":2,"Stage Attempt ID":0,"Task Type":"ResultTask","Task End Reason":{"Reason":"Success"},"Task Info":{"Task ID":34,"Index":2,"Attempt":0,"Launch Time":1594091481087,"Executor ID":"0","Host":"127.0.0.1","Locality":"PROCESS_LOCAL","Speculative":false,"Getting Result Time":0,"Finish Time":1594091483026,"Failed":false,"Killed":false,"Accumulables":[{"ID":0,"Name":"duration","Update":"1574","Value":"15555","Internal":true,"Count Failed Values":true,"Metadata":"sql"},{"ID":1,"Name":"number of output rows","Update":"375000","Value":"3750000","Internal":true,"Count Failed Values":true,"Metadata":"sql"},{"ID":14,"Name":"number of output rows","Update":"6250","Value":"62500","Internal":true,"Count Failed Values":true,"Metadata":"sql"},{"ID":74,"Name":"internal.metrics.peakExecutionMemory","Update":24040,"Value":240400,"Internal":true,"Count Failed Values":true},{"ID":70,"Name":"internal.metrics.jvmGCTime","Update":66,"Value":614,"Internal":true,"Count
Failed Values":true},{"ID":69,"Name":"internal.metrics.resultSize","Update":2115,"Value":21193,"Internal":true,"Count Failed Values":true},{"ID":68,"Name":"internal.metrics.executorCpuTime","Update":508433000,"Value":5824268000,"Internal":true,"Count Failed Values":true},{"ID":67,"Name":"internal.metrics.executorRunTime","Update":1606,"Value":15871,"Internal":true,"Count Failed Values":true},{"ID":66,"Name":"internal.metrics.executorDeserializeCpuTime","Update":69492000,"Value":656833000,"Internal":true,"Count Failed Values":true},{"ID":65,"Name":"internal.metrics.executorDeserializeTime","Update":265,"Value":2500,"Internal":true,"Count Failed Values":true}]},"Task Executor Metrics":{"JVMHeapMemory":0,"JVMOffHeapMemory":0,"OnHeapExecutionMemory":0,"OffHeapExecutionMemory":0,"OnHeapStorageMemory":0,"OffHeapStorageMemory":0,"OnHeapUnifiedMemory":0,"OffHeapUnifiedMemory":0,"DirectPoolMemory":0,"MappedPoolMemory":0,"ProcessTreeJVMVMemory":0,"ProcessTreeJVMRSSMemory":0,"ProcessTreePython
VMemory":0,"ProcessTreePythonRSSMemory":0,"ProcessTreeOtherVMemory":0,"ProcessTreeOtherRSSMemory":0,"MinorGCCount":0,"MinorGCTime":0,"MajorGCCount":0,"MajorGCTime":0},"Task Metrics":{"Executor Deserialize Time":265,"Executor Deserialize CPU Time":69492000,"Executor Run Time":1606,"Executor CPU Time":508433000,"Peak Execution Memory":24040,"Result Size":2115,"JVM GC Time":66,"Result Serialization Time":0,"Memory Bytes Spilled":0,"Disk Bytes Spilled":0,"Shuffle Read Metrics":{"Remote Blocks Fetched":0,"Local Blocks Fetched":0,"Fetch Wait Time":0,"Remote Bytes Read":0,"Remote Bytes Read To Disk":0,"Local Bytes Read":0,"Total Records Read":0,"Shuffle Push Read Metrics":{"Corrupt Merged Block Chunks":10,"Merged Fallback Count":2,"Merged Remote Blocks Fetched":10,"Merged Local Blocks Fetched":15,"Merged Remote Chunks Fetched":12,"Merged Local Chunks Fetched":10,"Merged Remote Bytes Read":100,"Merged Local Bytes Read":100}},"Shuffle Write Metrics":{"Shuffle Bytes Written":0,"Shuffle Write
Time":0,"Shuffle Records Written":0},"Input Metrics":{"Bytes Read":0,"Records Read":0},"Output Metrics":{"Bytes Written":0,"Records Written":0},"Updated Blocks":[]}}
+{"Event":"SparkListenerTaskEnd","Stage ID":2,"Stage Attempt ID":0,"Task Type":"ResultTask","Task End Reason":{"Reason":"Success"},"Task Info":{"Task ID":43,"Index":11,"Attempt":0,"Launch Time":1594091481123,"Executor ID":"0","Host":"127.0.0.1","Locality":"PROCESS_LOCAL","Speculative":false,"Getting Result Time":0,"Finish Time":1594091483029,"Failed":false,"Killed":false,"Accumulables":[{"ID":0,"Name":"duration","Update":"1578","Value":"17133","Internal":true,"Count Failed Values":true,"Metadata":"sql"},{"ID":1,"Name":"number of output rows","Update":"375000","Value":"4125000","Internal":true,"Count Failed Values":true,"Metadata":"sql"},{"ID":14,"Name":"number of output rows","Update":"6250","Value":"68750","Internal":true,"Count Failed Values":true,"Metadata":"sql"},{"ID":74,"Name":"internal.metrics.peakExecutionMemory","Update":24040,"Value":264440,"Internal":true,"Count Failed Values":true},{"ID":70,"Name":"internal.metrics.jvmGCTime","Update":66,"Value":680,"Internal":true,"Count
Failed Values":true},{"ID":69,"Name":"internal.metrics.resultSize","Update":2115,"Value":23308,"Internal":true,"Count Failed Values":true},{"ID":68,"Name":"internal.metrics.executorCpuTime","Update":502120000,"Value":6326388000,"Internal":true,"Count Failed Values":true},{"ID":67,"Name":"internal.metrics.executorRunTime","Update":1609,"Value":17480,"Internal":true,"Count Failed Values":true},{"ID":66,"Name":"internal.metrics.executorDeserializeCpuTime","Update":48849000,"Value":705682000,"Internal":true,"Count Failed Values":true},{"ID":65,"Name":"internal.metrics.executorDeserializeTime","Update":225,"Value":2725,"Internal":true,"Count Failed Values":true}]},"Task Executor Metrics":{"JVMHeapMemory":0,"JVMOffHeapMemory":0,"OnHeapExecutionMemory":0,"OffHeapExecutionMemory":0,"OnHeapStorageMemory":0,"OffHeapStorageMemory":0,"OnHeapUnifiedMemory":0,"OffHeapUnifiedMemory":0,"DirectPoolMemory":0,"MappedPoolMemory":0,"ProcessTreeJVMVMemory":0,"ProcessTreeJVMRSSMemory":0,"ProcessTreePytho
nVMemory":0,"ProcessTreePythonRSSMemory":0,"ProcessTreeOtherVMemory":0,"ProcessTreeOtherRSSMemory":0,"MinorGCCount":0,"MinorGCTime":0,"MajorGCCount":0,"MajorGCTime":0},"Task Metrics":{"Executor Deserialize Time":225,"Executor Deserialize CPU Time":48849000,"Executor Run Time":1609,"Executor CPU Time":502120000,"Peak Execution Memory":24040,"Result Size":2115,"JVM GC Time":66,"Result Serialization Time":0,"Memory Bytes Spilled":0,"Disk Bytes Spilled":0,"Shuffle Read Metrics":{"Remote Blocks Fetched":0,"Local Blocks Fetched":0,"Fetch Wait Time":0,"Remote Bytes Read":0,"Remote Bytes Read To Disk":0,"Local Bytes Read":0,"Total Records Read":0,"Shuffle Push Read Metrics":{"Corrupt Merged Block Chunks":10,"Merged Fallback Count":2,"Merged Remote Blocks Fetched":10,"Merged Local Blocks Fetched":15,"Merged Remote Chunks Fetched":12,"Merged Local Chunks Fetched":10,"Merged Remote Bytes Read":100,"Merged Local Bytes Read":100}},"Shuffle Write Metrics":{"Shuffle Bytes Written":0,"Shuffle Write
Time":0,"Shuffle Records Written":0},"Input Metrics":{"Bytes Read":0,"Records Read":0},"Output Metrics":{"Bytes Written":0,"Records Written":0},"Updated Blocks":[]}}
+{"Event":"SparkListenerTaskEnd","Stage ID":2,"Stage Attempt ID":0,"Task Type":"ResultTask","Task End Reason":{"Reason":"Success"},"Task Info":{"Task ID":41,"Index":9,"Attempt":0,"Launch Time":1594091481116,"Executor ID":"0","Host":"127.0.0.1","Locality":"PROCESS_LOCAL","Speculative":false,"Getting Result Time":0,"Finish Time":1594091483032,"Failed":false,"Killed":false,"Accumulables":[{"ID":0,"Name":"duration","Update":"1583","Value":"18716","Internal":true,"Count Failed Values":true,"Metadata":"sql"},{"ID":1,"Name":"number of output rows","Update":"375000","Value":"4500000","Internal":true,"Count Failed Values":true,"Metadata":"sql"},{"ID":14,"Name":"number of output rows","Update":"6250","Value":"75000","Internal":true,"Count Failed Values":true,"Metadata":"sql"},{"ID":74,"Name":"internal.metrics.peakExecutionMemory","Update":24040,"Value":288480,"Internal":true,"Count Failed Values":true},{"ID":70,"Name":"internal.metrics.jvmGCTime","Update":66,"Value":746,"Internal":true,"Count
Failed Values":true},{"ID":69,"Name":"internal.metrics.resultSize","Update":2115,"Value":25423,"Internal":true,"Count Failed Values":true},{"ID":68,"Name":"internal.metrics.executorCpuTime","Update":489923000,"Value":6816311000,"Internal":true,"Count Failed Values":true},{"ID":67,"Name":"internal.metrics.executorRunTime","Update":1614,"Value":19094,"Internal":true,"Count Failed Values":true},{"ID":66,"Name":"internal.metrics.executorDeserializeCpuTime","Update":55787000,"Value":761469000,"Internal":true,"Count Failed Values":true},{"ID":65,"Name":"internal.metrics.executorDeserializeTime","Update":240,"Value":2965,"Internal":true,"Count Failed Values":true}]},"Task Executor Metrics":{"JVMHeapMemory":0,"JVMOffHeapMemory":0,"OnHeapExecutionMemory":0,"OffHeapExecutionMemory":0,"OnHeapStorageMemory":0,"OffHeapStorageMemory":0,"OnHeapUnifiedMemory":0,"OffHeapUnifiedMemory":0,"DirectPoolMemory":0,"MappedPoolMemory":0,"ProcessTreeJVMVMemory":0,"ProcessTreeJVMRSSMemory":0,"ProcessTreePython
VMemory":0,"ProcessTreePythonRSSMemory":0,"ProcessTreeOtherVMemory":0,"ProcessTreeOtherRSSMemory":0,"MinorGCCount":0,"MinorGCTime":0,"MajorGCCount":0,"MajorGCTime":0},"Task Metrics":{"Executor Deserialize Time":240,"Executor Deserialize CPU Time":55787000,"Executor Run Time":1614,"Executor CPU Time":489923000,"Peak Execution Memory":24040,"Result Size":2115,"JVM GC Time":66,"Result Serialization Time":0,"Memory Bytes Spilled":0,"Disk Bytes Spilled":0,"Shuffle Read Metrics":{"Remote Blocks Fetched":0,"Local Blocks Fetched":0,"Fetch Wait Time":0,"Remote Bytes Read":0,"Remote Bytes Read To Disk":0,"Local Bytes Read":0,"Total Records Read":0,"Shuffle Push Read Metrics":{"Corrupt Merged Block Chunks":10,"Merged Fallback Count":2,"Merged Remote Blocks Fetched":10,"Merged Local Blocks Fetched":15,"Merged Remote Chunks Fetched":12,"Merged Local Chunks Fetched":10,"Merged Remote Bytes Read":100,"Merged Local Bytes Read":100}},"Shuffle Write Metrics":{"Shuffle Bytes Written":0,"Shuffle Write
Time":0,"Shuffle Records Written":0},"Input Metrics":{"Bytes Read":0,"Records Read":0},"Output Metrics":{"Bytes Written":0,"Records Written":0},"Updated Blocks":[]}}
+{"Event":"SparkListenerTaskEnd","Stage ID":2,"Stage Attempt ID":0,"Task Type":"ResultTask","Task End Reason":{"Reason":"Success"},"Task Info":{"Task ID":46,"Index":14,"Attempt":0,"Launch Time":1594091481132,"Executor ID":"0","Host":"127.0.0.1","Locality":"PROCESS_LOCAL","Speculative":false,"Getting Result Time":0,"Finish Time":1594091483037,"Failed":false,"Killed":false,"Accumulables":[{"ID":0,"Name":"duration","Update":"1587","Value":"20303","Internal":true,"Count Failed Values":true,"Metadata":"sql"},{"ID":1,"Name":"number of output rows","Update":"375000","Value":"4875000","Internal":true,"Count Failed Values":true,"Metadata":"sql"},{"ID":14,"Name":"number of output rows","Update":"6250","Value":"81250","Internal":true,"Count Failed Values":true,"Metadata":"sql"},{"ID":74,"Name":"internal.metrics.peakExecutionMemory","Update":24040,"Value":312520,"Internal":true,"Count Failed Values":true},{"ID":70,"Name":"internal.metrics.jvmGCTime","Update":66,"Value":812,"Internal":true,"Count
Failed Values":true},{"ID":69,"Name":"internal.metrics.resultSize","Update":2115,"Value":27538,"Internal":true,"Count Failed Values":true},{"ID":68,"Name":"internal.metrics.executorCpuTime","Update":490927000,"Value":7307238000,"Internal":true,"Count Failed Values":true},{"ID":67,"Name":"internal.metrics.executorRunTime","Update":1618,"Value":20712,"Internal":true,"Count Failed Values":true},{"ID":66,"Name":"internal.metrics.executorDeserializeCpuTime","Update":51464000,"Value":812933000,"Internal":true,"Count Failed Values":true},{"ID":65,"Name":"internal.metrics.executorDeserializeTime","Update":218,"Value":3183,"Internal":true,"Count Failed Values":true}]},"Task Executor Metrics":{"JVMHeapMemory":0,"JVMOffHeapMemory":0,"OnHeapExecutionMemory":0,"OffHeapExecutionMemory":0,"OnHeapStorageMemory":0,"OffHeapStorageMemory":0,"OnHeapUnifiedMemory":0,"OffHeapUnifiedMemory":0,"DirectPoolMemory":0,"MappedPoolMemory":0,"ProcessTreeJVMVMemory":0,"ProcessTreeJVMRSSMemory":0,"ProcessTreePytho
nVMemory":0,"ProcessTreePythonRSSMemory":0,"ProcessTreeOtherVMemory":0,"ProcessTreeOtherRSSMemory":0,"MinorGCCount":0,"MinorGCTime":0,"MajorGCCount":0,"MajorGCTime":0},"Task Metrics":{"Executor Deserialize Time":218,"Executor Deserialize CPU Time":51464000,"Executor Run Time":1618,"Executor CPU Time":490927000,"Peak Execution Memory":24040,"Result Size":2115,"JVM GC Time":66,"Result Serialization Time":0,"Memory Bytes Spilled":0,"Disk Bytes Spilled":0,"Shuffle Read Metrics":{"Remote Blocks Fetched":0,"Local Blocks Fetched":0,"Fetch Wait Time":0,"Remote Bytes Read":0,"Remote Bytes Read To Disk":0,"Local Bytes Read":0,"Total Records Read":0,"Shuffle Push Read Metrics":{"Corrupt Merged Block Chunks":10,"Merged Fallback Count":2,"Merged Remote Blocks Fetched":10,"Merged Local Blocks Fetched":15,"Merged Remote Chunks Fetched":12,"Merged Local Chunks Fetched":10,"Merged Remote Bytes Read":100,"Merged Local Bytes Read":100}},"Shuffle Write Metrics":{"Shuffle Bytes Written":0,"Shuffle Write
Time":0,"Shuffle Records Written":0},"Input Metrics":{"Bytes Read":0,"Records Read":0},"Output Metrics":{"Bytes Written":0,"Records Written":0},"Updated Blocks":[]}}
+{"Event":"SparkListenerTaskEnd","Stage ID":2,"Stage Attempt ID":0,"Task Type":"ResultTask","Task End Reason":{"Reason":"Success"},"Task Info":{"Task ID":32,"Index":0,"Attempt":0,"Launch Time":1594091481077,"Executor ID":"0","Host":"127.0.0.1","Locality":"PROCESS_LOCAL","Speculative":false,"Getting Result Time":0,"Finish Time":1594091483037,"Failed":false,"Killed":false,"Accumulables":[{"ID":0,"Name":"duration","Update":"1587","Value":"21890","Internal":true,"Count Failed Values":true,"Metadata":"sql"},{"ID":1,"Name":"number of output rows","Update":"375000","Value":"5250000","Internal":true,"Count Failed Values":true,"Metadata":"sql"},{"ID":14,"Name":"number of output rows","Update":"6250","Value":"87500","Internal":true,"Count Failed Values":true,"Metadata":"sql"},{"ID":74,"Name":"internal.metrics.peakExecutionMemory","Update":24040,"Value":336560,"Internal":true,"Count Failed Values":true},{"ID":70,"Name":"internal.metrics.jvmGCTime","Update":66,"Value":878,"Internal":true,"Count
Failed Values":true},{"ID":69,"Name":"internal.metrics.resultSize","Update":2115,"Value":29653,"Internal":true,"Count Failed Values":true},{"ID":68,"Name":"internal.metrics.executorCpuTime","Update":496683000,"Value":7803921000,"Internal":true,"Count Failed Values":true},{"ID":67,"Name":"internal.metrics.executorRunTime","Update":1619,"Value":22331,"Internal":true,"Count Failed Values":true},{"ID":66,"Name":"internal.metrics.executorDeserializeCpuTime","Update":56827000,"Value":869760000,"Internal":true,"Count Failed Values":true},{"ID":65,"Name":"internal.metrics.executorDeserializeTime","Update":271,"Value":3454,"Internal":true,"Count Failed Values":true}]},"Task Executor Metrics":{"JVMHeapMemory":0,"JVMOffHeapMemory":0,"OnHeapExecutionMemory":0,"OffHeapExecutionMemory":0,"OnHeapStorageMemory":0,"OffHeapStorageMemory":0,"OnHeapUnifiedMemory":0,"OffHeapUnifiedMemory":0,"DirectPoolMemory":0,"MappedPoolMemory":0,"ProcessTreeJVMVMemory":0,"ProcessTreeJVMRSSMemory":0,"ProcessTreePython
VMemory":0,"ProcessTreePythonRSSMemory":0,"ProcessTreeOtherVMemory":0,"ProcessTreeOtherRSSMemory":0,"MinorGCCount":0,"MinorGCTime":0,"MajorGCCount":0,"MajorGCTime":0},"Task Metrics":{"Executor Deserialize Time":271,"Executor Deserialize CPU Time":56827000,"Executor Run Time":1619,"Executor CPU Time":496683000,"Peak Execution Memory":24040,"Result Size":2115,"JVM GC Time":66,"Result Serialization Time":0,"Memory Bytes Spilled":0,"Disk Bytes Spilled":0,"Shuffle Read Metrics":{"Remote Blocks Fetched":0,"Local Blocks Fetched":0,"Fetch Wait Time":0,"Remote Bytes Read":0,"Remote Bytes Read To Disk":0,"Local Bytes Read":0,"Total Records Read":0,"Shuffle Push Read Metrics":{"Corrupt Merged Block Chunks":10,"Merged Fallback Count":2,"Merged Remote Blocks Fetched":10,"Merged Local Blocks Fetched":15,"Merged Remote Chunks Fetched":12,"Merged Local Chunks Fetched":10,"Merged Remote Bytes Read":100,"Merged Local Bytes Read":100}},"Shuffle Write Metrics":{"Shuffle Bytes Written":0,"Shuffle Write
Time":0,"Shuffle Records Written":0},"Input Metrics":{"Bytes Read":0,"Records Read":0},"Output Metrics":{"Bytes Written":0,"Records Written":0},"Updated Blocks":[]}}
+{"Event":"SparkListenerTaskEnd","Stage ID":2,"Stage Attempt ID":0,"Task Type":"ResultTask","Task End Reason":{"Reason":"Success"},"Task Info":{"Task ID":42,"Index":10,"Attempt":0,"Launch Time":1594091481120,"Executor ID":"0","Host":"127.0.0.1","Locality":"PROCESS_LOCAL","Speculative":false,"Getting Result Time":0,"Finish Time":1594091483043,"Failed":false,"Killed":false,"Accumulables":[{"ID":0,"Name":"duration","Update":"1593","Value":"23483","Internal":true,"Count Failed Values":true,"Metadata":"sql"},{"ID":1,"Name":"number of output rows","Update":"375000","Value":"5625000","Internal":true,"Count Failed Values":true,"Metadata":"sql"},{"ID":14,"Name":"number of output rows","Update":"6250","Value":"93750","Internal":true,"Count Failed Values":true,"Metadata":"sql"},{"ID":74,"Name":"internal.metrics.peakExecutionMemory","Update":24040,"Value":360600,"Internal":true,"Count Failed Values":true},{"ID":70,"Name":"internal.metrics.jvmGCTime","Update":66,"Value":944,"Internal":true,"Count
Failed Values":true},{"ID":69,"Name":"internal.metrics.resultSize","Update":2115,"Value":31768,"Internal":true,"Count Failed Values":true},{"ID":68,"Name":"internal.metrics.executorCpuTime","Update":508230000,"Value":8312151000,"Internal":true,"Count Failed Values":true},{"ID":67,"Name":"internal.metrics.executorRunTime","Update":1624,"Value":23955,"Internal":true,"Count Failed Values":true},{"ID":66,"Name":"internal.metrics.executorDeserializeCpuTime","Update":58152000,"Value":927912000,"Internal":true,"Count Failed Values":true},{"ID":65,"Name":"internal.metrics.executorDeserializeTime","Update":229,"Value":3683,"Internal":true,"Count Failed Values":true}]},"Task Executor Metrics":{"JVMHeapMemory":0,"JVMOffHeapMemory":0,"OnHeapExecutionMemory":0,"OffHeapExecutionMemory":0,"OnHeapStorageMemory":0,"OffHeapStorageMemory":0,"OnHeapUnifiedMemory":0,"OffHeapUnifiedMemory":0,"DirectPoolMemory":0,"MappedPoolMemory":0,"ProcessTreeJVMVMemory":0,"ProcessTreeJVMRSSMemory":0,"ProcessTreePytho
nVMemory":0,"ProcessTreePythonRSSMemory":0,"ProcessTreeOtherVMemory":0,"ProcessTreeOtherRSSMemory":0,"MinorGCCount":0,"MinorGCTime":0,"MajorGCCount":0,"MajorGCTime":0},"Task Metrics":{"Executor Deserialize Time":229,"Executor Deserialize CPU Time":58152000,"Executor Run Time":1624,"Executor CPU Time":508230000,"Peak Execution Memory":24040,"Result Size":2115,"JVM GC Time":66,"Result Serialization Time":0,"Memory Bytes Spilled":0,"Disk Bytes Spilled":0,"Shuffle Read Metrics":{"Remote Blocks Fetched":0,"Local Blocks Fetched":0,"Fetch Wait Time":0,"Remote Bytes Read":0,"Remote Bytes Read To Disk":0,"Local Bytes Read":0,"Total Records Read":0,"Shuffle Push Read Metrics":{"Corrupt Merged Block Chunks":10,"Merged Fallback Count":2,"Merged Remote Blocks Fetched":10,"Merged Local Blocks Fetched":15,"Merged Remote Chunks Fetched":12,"Merged Local Chunks Fetched":10,"Merged Remote Bytes Read":100,"Merged Local Bytes Read":100}},"Shuffle Write Metrics":{"Shuffle Bytes Written":0,"Shuffle Write
Time":0,"Shuffle Records Written":0},"Input Metrics":{"Bytes Read":0,"Records Read":0},"Output Metrics":{"Bytes Written":0,"Records Written":0},"Updated Blocks":[]}}
+{"Event":"SparkListenerTaskEnd","Stage ID":2,"Stage Attempt ID":0,"Task Type":"ResultTask","Task End Reason":{"Reason":"Success"},"Task Info":{"Task ID":44,"Index":12,"Attempt":0,"Launch Time":1594091481126,"Executor ID":"0","Host":"127.0.0.1","Locality":"PROCESS_LOCAL","Speculative":false,"Getting Result Time":0,"Finish Time":1594091483043,"Failed":false,"Killed":false,"Accumulables":[{"ID":0,"Name":"duration","Update":"1593","Value":"25076","Internal":true,"Count Failed Values":true,"Metadata":"sql"},{"ID":1,"Name":"number of output rows","Update":"375000","Value":"6000000","Internal":true,"Count Failed Values":true,"Metadata":"sql"},{"ID":14,"Name":"number of output rows","Update":"6250","Value":"100000","Internal":true,"Count Failed Values":true,"Metadata":"sql"},{"ID":74,"Name":"internal.metrics.peakExecutionMemory","Update":24040,"Value":384640,"Internal":true,"Count Failed Values":true},{"ID":70,"Name":"internal.metrics.jvmGCTime","Update":66,"Value":1010,"Internal":true,"Cou
nt Failed Values":true},{"ID":69,"Name":"internal.metrics.resultSize","Update":2115,"Value":33883,"Internal":true,"Count Failed Values":true},{"ID":68,"Name":"internal.metrics.executorCpuTime","Update":498187000,"Value":8810338000,"Internal":true,"Count Failed Values":true},{"ID":67,"Name":"internal.metrics.executorRunTime","Update":1624,"Value":25579,"Internal":true,"Count Failed Values":true},{"ID":66,"Name":"internal.metrics.executorDeserializeCpuTime","Update":51988000,"Value":979900000,"Internal":true,"Count Failed Values":true},{"ID":65,"Name":"internal.metrics.executorDeserializeTime","Update":222,"Value":3905,"Internal":true,"Count Failed Values":true}]},"Task Executor Metrics":{"JVMHeapMemory":0,"JVMOffHeapMemory":0,"OnHeapExecutionMemory":0,"OffHeapExecutionMemory":0,"OnHeapStorageMemory":0,"OffHeapStorageMemory":0,"OnHeapUnifiedMemory":0,"OffHeapUnifiedMemory":0,"DirectPoolMemory":0,"MappedPoolMemory":0,"ProcessTreeJVMVMemory":0,"ProcessTreeJVMRSSMemory":0,"ProcessTreePyt
honVMemory":0,"ProcessTreePythonRSSMemory":0,"ProcessTreeOtherVMemory":0,"ProcessTreeOtherRSSMemory":0,"MinorGCCount":0,"MinorGCTime":0,"MajorGCCount":0,"MajorGCTime":0},"Task Metrics":{"Executor Deserialize Time":222,"Executor Deserialize CPU Time":51988000,"Executor Run Time":1624,"Executor CPU Time":498187000,"Peak Execution Memory":24040,"Result Size":2115,"JVM GC Time":66,"Result Serialization Time":0,"Memory Bytes Spilled":0,"Disk Bytes Spilled":0,"Shuffle Read Metrics":{"Remote Blocks Fetched":0,"Local Blocks Fetched":0,"Fetch Wait Time":0,"Remote Bytes Read":0,"Remote Bytes Read To Disk":0,"Local Bytes Read":0,"Total Records Read":0,"Shuffle Push Read Metrics":{"Corrupt Merged Block Chunks":10,"Merged Fallback Count":2,"Merged Remote Blocks Fetched":10,"Merged Local Blocks Fetched":15,"Merged Remote Chunks Fetched":12,"Merged Local Chunks Fetched":10,"Merged Remote Bytes Read":100,"Merged Local Bytes Read":100}},"Shuffle Write Metrics":{"Shuffle Bytes Written":0,"Shuffle Wri
te Time":0,"Shuffle Records Written":0},"Input Metrics":{"Bytes Read":0,"Records Read":0},"Output Metrics":{"Bytes Written":0,"Records Written":0},"Updated Blocks":[]}}
{"Event":"SparkListenerStageExecutorMetrics","Executor ID":"driver","Stage ID":2,"Stage Attempt ID":0,"Executor Metrics":{"JVMHeapMemory":213367864,"JVMOffHeapMemory":189011656,"OnHeapExecutionMemory":0,"OffHeapExecutionMemory":0,"OnHeapStorageMemory":2133349,"OffHeapStorageMemory":0,"OnHeapUnifiedMemory":2133349,"OffHeapUnifiedMemory":0,"DirectPoolMemory":282024,"MappedPoolMemory":0,"ProcessTreeJVMVMemory":0,"ProcessTreeJVMRSSMemory":0,"ProcessTreePythonVMemory":0,"ProcessTreePythonRSSMemory":0,"ProcessTreeOtherVMemory":0,"ProcessTreeOtherRSSMemory":0,"MinorGCCount":13,"MinorGCTime":115,"MajorGCCount":4,"MajorGCTime":339}}
{"Event":"SparkListenerStageExecutorMetrics","Executor ID":"0","Stage ID":2,"Stage Attempt ID":0,"Executor Metrics":{"JVMHeapMemory":0,"JVMOffHeapMemory":0,"OnHeapExecutionMemory":0,"OffHeapExecutionMemory":0,"OnHeapStorageMemory":0,"OffHeapStorageMemory":0,"OnHeapUnifiedMemory":0,"OffHeapUnifiedMemory":0,"DirectPoolMemory":0,"MappedPoolMemory":0,"ProcessTreeJVMVMemory":0,"ProcessTreeJVMRSSMemory":0,"ProcessTreePythonVMemory":0,"ProcessTreePythonRSSMemory":0,"ProcessTreeOtherVMemory":0,"ProcessTreeOtherRSSMemory":0,"MinorGCCount":0,"MinorGCTime":0,"MajorGCCount":0,"MajorGCTime":0}}
{"Event":"SparkListenerStageCompleted","Stage Info":{"Stage ID":2,"Stage Attempt ID":0,"Stage Name":"foreach at <console>:26","Number of Tasks":16,"RDD Info":[{"RDD ID":10,"Name":"MapPartitionsRDD","Scope":"{\"id\":\"18\",\"name\":\"mapPartitions\"}","Callsite":"foreach at <console>:26","Parent IDs":[9],"Storage Level":{"Use Disk":false,"Use Memory":false,"Deserialized":false,"Replication":1},"Barrier":false,"Number of Partitions":16,"Number of Cached Partitions":0,"Memory Size":0,"Disk Size":0},{"RDD ID":8,"Name":"MapPartitionsRDD","Scope":"{\"id\":\"0\",\"name\":\"DeserializeToObject\"}","Callsite":"foreach at <console>:26","Parent IDs":[7],"Storage Level":{"Use Disk":false,"Use Memory":false,"Deserialized":false,"Replication":1},"Barrier":false,"Number of Partitions":16,"Number of Cached Partitions":0,"Memory Size":0,"Disk Size":0},{"RDD ID":6,"Name":"ParallelCollectionRDD","Scope":"{\"id\":\"1\",\"name\":\"WholeStageCodegen (2)\"}","Callsite":"foreach at <console>:26","Parent ID
s":[],"Storage Level":{"Use Disk":false,"Use Memory":false,"Deserialized":false,"Replication":1},"Barrier":false,"Number of Partitions":16,"Number of Cached Partitions":0,"Memory Size":0,"Disk Size":0},{"RDD ID":7,"Name":"MapPartitionsRDD","Scope":"{\"id\":\"1\",\"name\":\"WholeStageCodegen (2)\"}","Callsite":"foreach at <console>:26","Parent IDs":[6],"Storage Level":{"Use Disk":false,"Use Memory":false,"Deserialized":false,"Replication":1},"Barrier":false,"Number of Partitions":16,"Number of Cached Partitions":0,"Memory Size":0,"Disk Size":0},{"RDD ID":9,"Name":"SQLExecutionRDD","Callsite":"foreach at <console>:26","Parent IDs":[8],"Storage Level":{"Use Disk":false,"Use Memory":false,"Deserialized":false,"Replication":1},"Barrier":false,"Number of Partitions":16,"Number of Cached Partitions":0,"Memory Size":0,"Disk Size":0}],"Parent IDs":[],"Details":"org.apache.spark.sql.Dataset.foreach(Dataset.scala:2862)\n$line19.$read$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw.<init>(<console>:26)\n$line1
9.$read$$iw$$iw$$iw$$iw$$iw$$iw$$iw.<init>(<console>:30)\n$line19.$read$$iw$$iw$$iw$$iw$$iw$$iw.<init>(<console>:32)\n$line19.$read$$iw$$iw$$iw$$iw$$iw.<init>(<console>:34)\n$line19.$read$$iw$$iw$$iw$$iw.<init>(<console>:36)\n$line19.$read$$iw$$iw$$iw.<init>(<console>:38)\n$line19.$read$$iw$$iw.<init>(<console>:40)\n$line19.$read$$iw.<init>(<console>:42)\n$line19.$read.<init>(<console>:44)\n$line19.$read$.<init>(<console>:48)\n$line19.$read$.<clinit>(<console>)\n$line19.$eval$.$print$lzycompute(<console>:7)\n$line19.$eval$.$print(<console>:6)\n$line19.$eval.$print(<console>)\nsun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)\nsun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)\nsun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)\njava.lang.reflect.Method.invoke(Method.java:498)\nscala.tools.nsc.interpreter.IMain$ReadEvalPrint.call(IMain.scala:745)","Submission Time":1594091481040,"Completion Time":1594091483044,"A
ccumulables":[{"ID":68,"Name":"internal.metrics.executorCpuTime","Value":8810338000,"Internal":true,"Count Failed Values":true},{"ID":71,"Name":"internal.metrics.resultSerializationTime","Value":11,"Internal":true,"Count Failed Values":true},{"ID":74,"Name":"internal.metrics.peakExecutionMemory","Value":384640,"Internal":true,"Count Failed Values":true},{"ID":65,"Name":"internal.metrics.executorDeserializeTime","Value":3905,"Internal":true,"Count Failed Values":true},{"ID":14,"Name":"number of output rows","Value":"100000","Internal":true,"Count Failed Values":true,"Metadata":"sql"},{"ID":67,"Name":"internal.metrics.executorRunTime","Value":25579,"Internal":true,"Count Failed Values":true},{"ID":70,"Name":"internal.metrics.jvmGCTime","Value":1010,"Internal":true,"Count Failed Values":true},{"ID":1,"Name":"number of output rows","Value":"6000000","Internal":true,"Count Failed Values":true,"Metadata":"sql"},{"ID":69,"Name":"internal.metrics.resultSize","Value":33883,"Internal":true,"C
ount Failed Values":true},{"ID":66,"Name":"internal.metrics.executorDeserializeCpuTime","Value":979900000,"Internal":true,"Count Failed Values":true},{"ID":0,"Name":"duration","Value":"25076","Internal":true,"Count Failed Values":true,"Metadata":"sql"}],"Resource Profile Id":0}}
{"Event":"SparkListenerJobEnd","Job ID":2,"Completion Time":1594091483044,"Job Result":{"Result":"JobSucceeded"}}
{"Event":"org.apache.spark.sql.execution.ui.SparkListenerSQLExecutionEnd","executionId":0,"time":1594091483045}
-{"Event":"SparkListenerApplicationEnd","Timestamp":1594091824231}
+{"Event":"SparkListenerApplicationEnd","Timestamp":1594091824231}
Review Comment:
Missing new line at the end of file
--
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
[GitHub] [spark] otterc commented on a diff in pull request #36165: [SPARK-36620][SHUFFLE] Add Push Based Shuffle client side read metrics
Posted by GitBox <gi...@apache.org>.
otterc commented on code in PR #36165:
URL: https://github.com/apache/spark/pull/36165#discussion_r1046313315
##########
core/src/main/scala/org/apache/spark/storage/ShuffleBlockFetcherIterator.scala:
##########
@@ -726,6 +736,56 @@ final class ShuffleBlockFetcherIterator(
}
}
+ // Number of map blocks in a ShuffleBlockChunk
+ private def getShuffleChunkCardinality(blockId: ShuffleBlockChunkId): Int = {
+ pushBasedFetchHelper.getRoaringBitMap(blockId).map(_.getCardinality).getOrElse(0)
+ }
Review Comment:
Should we move this to `PushBasedFetchHelper` since most of the utils belonging to pushShuffle are there?
--
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
[GitHub] [spark] otterc commented on a diff in pull request #36165: [SPARK-36620][SHUFFLE] Add Push Based Shuffle client side read metrics
Posted by GitBox <gi...@apache.org>.
otterc commented on code in PR #36165:
URL: https://github.com/apache/spark/pull/36165#discussion_r1043766268
##########
core/src/main/scala/org/apache/spark/storage/ShuffleBlockFetcherIterator.scala:
##########
@@ -726,6 +736,61 @@ final class ShuffleBlockFetcherIterator(
}
}
+ // Number of map blocks in a ShuffleBlockChunk
+ private def getShuffleChunkCardinality(blockId: ShuffleBlockChunkId): Int = {
+ pushBasedFetchHelper.getRoaringBitMap(blockId).map(_.getCardinality).getOrElse(0)
+ }
+
+ // Check if the merged block is local to the host or not
+ private def isLocalMergedBlockAddress(address: BlockManagerId): Boolean = {
+ address.executorId.isEmpty && address.host == blockManager.blockManagerId.host
Review Comment:
Is the executorId empty? or it is a special string? Need to check.
##########
core/src/main/scala/org/apache/spark/storage/ShuffleBlockFetcherIterator.scala:
##########
@@ -726,6 +736,61 @@ final class ShuffleBlockFetcherIterator(
}
}
+ // Number of map blocks in a ShuffleBlockChunk
+ private def getShuffleChunkCardinality(blockId: ShuffleBlockChunkId): Int = {
+ pushBasedFetchHelper.getRoaringBitMap(blockId).map(_.getCardinality).getOrElse(0)
+ }
+
+ // Check if the merged block is local to the host or not
+ private def isLocalMergedBlockAddress(address: BlockManagerId): Boolean = {
+ address.executorId.isEmpty && address.host == blockManager.blockManagerId.host
+ }
+
+ private def shuffleMetricsUpdate(
+ blockId: BlockId,
+ buf: ManagedBuffer,
+ local: Boolean): Unit = {
+ if (local) {
+ shuffleLocalMetricsUpdate(blockId, buf)
+ } else {
+ shuffleRemoteMetricsUpdate(blockId, buf)
+ }
+ }
+
+ private def shuffleLocalMetricsUpdate(blockId: BlockId, buf: ManagedBuffer): Unit = {
+ // Check if the block is within the host-local blocks set, or if it is a merged local
Review Comment:
Where is this check? Do we need to rephrase this comment?
##########
core/src/main/scala/org/apache/spark/storage/ShuffleBlockFetcherIterator.scala:
##########
@@ -343,6 +352,7 @@ final class ShuffleBlockFetcherIterator(
val block = BlockId(blockId)
if (block.isShuffleChunk) {
remainingBlocks -= blockId
+ updateMergedReqsDuration(wasReqForMergedChunks = true)
Review Comment:
Note to self: We are updating these metrics from the netty thread and not the task thread unlike other metrics. We do synchronize on the instance of the iterator. Shouldn't be a problem.
--
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
[GitHub] [spark] thejdeep commented on a diff in pull request #36165: [SPARK-36620][SHUFFLE] Add Push Based Shuffle client side read metrics
Posted by GitBox <gi...@apache.org>.
thejdeep commented on code in PR #36165:
URL: https://github.com/apache/spark/pull/36165#discussion_r1042043726
##########
core/src/test/resources/HistoryServerExpectations/excludeOnFailure_node_for_stage_expectation.json:
##########
@@ -81,7 +93,19 @@
"remoteBytesRead" : 0,
"remoteBytesReadToDisk" : 0,
"localBytesRead" : 0,
- "recordsRead" : 0
+ "recordsRead" : 0,
+ "pushBased": {
Review Comment:
Changed it to "pushBasedShuffle", thanks for the suggestion.
--
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
[GitHub] [spark] otterc commented on a diff in pull request #36165: [SPARK-36620][SHUFFLE] Add Push Based Shuffle client side read metrics
Posted by GitBox <gi...@apache.org>.
otterc commented on code in PR #36165:
URL: https://github.com/apache/spark/pull/36165#discussion_r1044862699
##########
core/src/test/resources/HistoryServerExpectations/one_stage_json_with_partitionId_expectation.json:
##########
@@ -26,13 +26,23 @@
"outputBytes" : 0,
"outputRecords" : 0,
"shuffleRemoteBlocksFetched" : 0,
- "shuffleLocalBlocksFetched" : 80,
+ "shuffleLocalBlocksFetched" : 0,
Review Comment:
Same here. Why did we change the value?
##########
core/src/test/resources/HistoryServerExpectations/one_stage_json_with_partitionId_expectation.json:
##########
@@ -26,13 +26,23 @@
"outputBytes" : 0,
"outputRecords" : 0,
"shuffleRemoteBlocksFetched" : 0,
- "shuffleLocalBlocksFetched" : 80,
+ "shuffleLocalBlocksFetched" : 0,
"shuffleFetchWaitTime" : 0,
"shuffleRemoteBytesRead" : 0,
"shuffleRemoteBytesReadToDisk" : 0,
- "shuffleLocalBytesRead" : 3760,
- "shuffleReadBytes" : 3760,
- "shuffleReadRecords" : 100,
+ "shuffleLocalBytesRead" : 0,
+ "shuffleReadBytes" : 0,
+ "shuffleReadRecords" : 0,
Review Comment:
Same for these.
##########
core/src/test/resources/HistoryServerExpectations/one_stage_json_with_partitionId_expectation.json:
##########
@@ -76,12 +86,24 @@
},
"shuffleReadMetrics" : {
"remoteBlocksFetched" : 0,
- "localBlocksFetched" : 8,
+ "localBlocksFetched" : 0,
"fetchWaitTime" : 0,
"remoteBytesRead" : 0,
"remoteBytesReadToDisk" : 0,
- "localBytesRead" : 376,
- "recordsRead" : 10
+ "localBytesRead" : 0,
+ "recordsRead" : 0,
Review Comment:
this as well
##########
core/src/test/resources/HistoryServerExpectations/one_stage_json_with_partitionId_expectation.json:
##########
@@ -127,12 +149,24 @@
},
"shuffleReadMetrics" : {
"remoteBlocksFetched" : 0,
- "localBlocksFetched" : 8,
+ "localBlocksFetched" : 0,
"fetchWaitTime" : 0,
"remoteBytesRead" : 0,
"remoteBytesReadToDisk" : 0,
- "localBytesRead" : 376,
- "recordsRead" : 10
+ "localBytesRead" : 0,
+ "recordsRead" : 0,
Review Comment:
existing value changed
##########
core/src/test/resources/HistoryServerExpectations/failed_stage_list_json_expectation.json:
##########
@@ -66,5 +76,7 @@
"MajorGCCount" : 0,
"MajorGCTime" : 0,
"TotalGCTime" : 0
- }
-} ]
+ },
+ "isPushBasedShuffleEnabled" : false,
+ "shuffleMergersCount" : 0
+} ]
Review Comment:
No newline at the end of file
##########
core/src/test/resources/HistoryServerExpectations/one_stage_json_with_partitionId_expectation.json:
##########
@@ -76,12 +86,24 @@
},
"shuffleReadMetrics" : {
"remoteBlocksFetched" : 0,
- "localBlocksFetched" : 8,
+ "localBlocksFetched" : 0,
Review Comment:
here as well
##########
core/src/test/resources/HistoryServerExpectations/one_stage_json_with_partitionId_expectation.json:
##########
@@ -178,12 +212,24 @@
},
"shuffleReadMetrics" : {
"remoteBlocksFetched" : 0,
- "localBlocksFetched" : 8,
+ "localBlocksFetched" : 0,
Review Comment:
same here
##########
core/src/test/resources/HistoryServerExpectations/one_stage_json_with_partitionId_expectation.json:
##########
@@ -178,12 +212,24 @@
},
"shuffleReadMetrics" : {
"remoteBlocksFetched" : 0,
- "localBlocksFetched" : 8,
+ "localBlocksFetched" : 0,
"fetchWaitTime" : 0,
"remoteBytesRead" : 0,
"remoteBytesReadToDisk" : 0,
- "localBytesRead" : 372,
- "recordsRead" : 9
+ "localBytesRead" : 0,
+ "recordsRead" : 0,
Review Comment:
same here
##########
core/src/test/resources/HistoryServerExpectations/one_stage_json_with_partitionId_expectation.json:
##########
@@ -127,12 +149,24 @@
},
"shuffleReadMetrics" : {
"remoteBlocksFetched" : 0,
- "localBlocksFetched" : 8,
+ "localBlocksFetched" : 0,
Review Comment:
value is changed
##########
core/src/test/resources/HistoryServerExpectations/failed_stage_list_json_expectation.json:
##########
@@ -27,13 +27,23 @@
"outputBytes" : 0,
"outputRecords" : 0,
"shuffleRemoteBlocksFetched" : 0,
- "shuffleLocalBlocksFetched" : 64,
- "shuffleFetchWaitTime" : 1,
+ "shuffleLocalBlocksFetched" : 0,
+ "shuffleFetchWaitTime" : 0,
Review Comment:
Why are we changing these values?
##########
core/src/test/resources/HistoryServerExpectations/one_stage_json_with_partitionId_expectation.json:
##########
@@ -620,5 +750,7 @@
"MajorGCCount" : 0,
"MajorGCTime" : 0,
"TotalGCTime" : 0
- }
+ },
+ "isPushBasedShuffleEnabled" : false,
+ "shuffleMergersCount" : 0
} ]
Review Comment:
We can add a new line at the end of file.
--
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
[GitHub] [spark] AmplabJenkins commented on pull request #36165: [SPARK-36620][SHUFFLE] Add Push Based Shuffle client side metrics
Posted by GitBox <gi...@apache.org>.
AmplabJenkins commented on PR #36165:
URL: https://github.com/apache/spark/pull/36165#issuecomment-1099601795
Can one of the admins verify this patch?
--
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
[GitHub] [spark] thejdeep commented on a diff in pull request #36165: [SPARK-36620][SHUFFLE] Add Push Based Shuffle client side metrics
Posted by GitBox <gi...@apache.org>.
thejdeep commented on code in PR #36165:
URL: https://github.com/apache/spark/pull/36165#discussion_r1029584122
##########
core/src/main/scala/org/apache/spark/executor/ShuffleReadMetrics.scala:
##########
@@ -146,6 +268,16 @@ private[spark] class TempShuffleReadMetrics extends ShuffleReadMetricsReporter {
override def incLocalBytesRead(v: Long): Unit = _localBytesRead += v
override def incFetchWaitTime(v: Long): Unit = _fetchWaitTime += v
override def incRecordsRead(v: Long): Unit = _recordsRead += v
+ override def incCorruptMergedBlockChunks(v: Long): Unit = _corruptMergedBlockChunks += v
+ override def incFallbackCount(v: Long): Unit = _fallbackCount += v
+ override def incRemoteMergedBlocksFetched(v: Long): Unit = _remoteMergedBlocksFetched += v
+ override def incLocalMergedBlocksFetched(v: Long): Unit = _localMergedBlocksFetched += v
+ override def incRemoteMergedChunksFetched(v: Long): Unit = _remoteMergedChunksFetched += v
+ override def incLocalMergedChunksFetched(v: Long): Unit = _localMergedChunksFetched += v
+ override def incRemoteMergedBlocksBytesRead(v: Long): Unit = _remoteMergedBlocksBytesRead += v
+ override def incLocalMergedBlocksBytesRead(v: Long): Unit = _localMergedBlocksBytesRead += v
+ override def incRemoteReqsDuration(v: Long): Unit = _remoteReqsDuration += v
+ override def incRemoteMergedReqsDuration(v: Long): Unit = _remoteMergedReqsDuration += v
Review Comment:
Updated `ShuffleBlockFetchIterator` to include these metrics too, thanks for catching it.
--
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
[GitHub] [spark] thejdeep commented on a diff in pull request #36165: [SPARK-36620][SHUFFLE] Add Push Based Shuffle client side metrics
Posted by GitBox <gi...@apache.org>.
thejdeep commented on code in PR #36165:
URL: https://github.com/apache/spark/pull/36165#discussion_r1010566832
##########
core/src/main/scala/org/apache/spark/status/storeTypes.scala:
##########
@@ -138,6 +138,16 @@ private[spark] object TaskIndexNames {
final val SHUFFLE_WRITE_RECORDS = "swr"
final val SHUFFLE_WRITE_SIZE = "sws"
final val SHUFFLE_WRITE_TIME = "swt"
+ final val PUSH_BASED_SHUFFLE_CORRUPT_MERGED_BLOCK_CHUNKS = "scmbc"
+ final val PUSH_BASED_SHUFFLE_FALLBACK_COUNT = "sfc"
+ final val PUSH_BASED_SHUFFLE_MERGED_REMOTE_BLOCKS = "smrb"
+ final val PUSH_BASED_SHUFFLE_MERGED_LOCAL_BLOCKS = "smlb"
+ final val PUSH_BASED_SHUFFLE_MERGED_REMOTE_CHUNKS = "smrc"
+ final val PUSH_BASED_SHUFFLE_MERGED_LOCAL_CHUNKS = "smlc"
+ final val PUSH_BASED_SHUFFLE_MERGED_REMOTE_READS = "smrr"
+ final val PUSH_BASED_SHUFFLE_MERGED_LOCAL_READS = "smlr"
+ final val PUSH_BASED_SHUFFLE_REMOTE_REQS_DURATION = "srrd"
+ final val PUSH_BASED_SHUFFLE_MERGED_REMOTE_REQS_DURATION = "smrrd"
Review Comment:
Added prefix, 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: 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
[GitHub] [spark] gengliangwang commented on a diff in pull request #36165: [SPARK-36620][SHUFFLE] Add Push Based Shuffle client side read metrics
Posted by GitBox <gi...@apache.org>.
gengliangwang commented on code in PR #36165:
URL: https://github.com/apache/spark/pull/36165#discussion_r1061202859
##########
core/src/main/scala/org/apache/spark/status/storeTypes.scala:
##########
@@ -233,6 +243,38 @@ private[spark] class TaskDataWrapper(
val shuffleLocalBytesRead: Long,
@KVIndexParam(value = TaskIndexNames.SHUFFLE_READ_RECORDS, parent = TaskIndexNames.STAGE)
val shuffleRecordsRead: Long,
+ @KVIndexParam(
Review Comment:
QQ: Do we have the requirement to sort task data by the new 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: 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
[GitHub] [spark] JoshRosen commented on pull request #36165: [SPARK-36620][SHUFFLE] Add Push Based Shuffle client side read metrics
Posted by "JoshRosen (via GitHub)" <gi...@apache.org>.
JoshRosen commented on PR #36165:
URL: https://github.com/apache/spark/pull/36165#issuecomment-1405674124
I think that we should skip logging of these metrics in JsonProtocol when push-based shuffle is disabled, so I've filed https://issues.apache.org/jira/browse/SPARK-42203
--
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
[GitHub] [spark] mridulm commented on pull request #36165: [SPARK-36620][SHUFFLE] Add Push Based Shuffle client side read metrics
Posted by GitBox <gi...@apache.org>.
mridulm commented on PR #36165:
URL: https://github.com/apache/spark/pull/36165#issuecomment-1373325023
Merged to master.
Thanks for working on this @thejdeep !
Thanks for the reviews @otterc, @zhouyejoe, @gengliangwang :-)
--
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
[GitHub] [spark] thejdeep commented on a diff in pull request #36165: [SPARK-36620][SHUFFLE] Add Push Based Shuffle client side read metrics
Posted by GitBox <gi...@apache.org>.
thejdeep commented on code in PR #36165:
URL: https://github.com/apache/spark/pull/36165#discussion_r1061723365
##########
core/src/main/scala/org/apache/spark/status/storeTypes.scala:
##########
@@ -233,6 +243,38 @@ private[spark] class TaskDataWrapper(
val shuffleLocalBytesRead: Long,
@KVIndexParam(value = TaskIndexNames.SHUFFLE_READ_RECORDS, parent = TaskIndexNames.STAGE)
val shuffleRecordsRead: Long,
+ @KVIndexParam(
Review Comment:
Although we do not sort these in the UI, adding indices for these are a requirement for computing quantiles and while fetching `TaskDataWrapper`s.
--
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
[GitHub] [spark] otterc commented on a diff in pull request #36165: [SPARK-36620][SHUFFLE] Add Push Based Shuffle client side metrics
Posted by GitBox <gi...@apache.org>.
otterc commented on code in PR #36165:
URL: https://github.com/apache/spark/pull/36165#discussion_r1008467914
##########
core/src/main/scala/org/apache/spark/storage/PushBasedFetchHelper.scala:
##########
@@ -290,6 +301,7 @@ private class PushBasedFetchHelper(
address: BlockManagerId): Unit = {
assert(blockId.isInstanceOf[ShuffleMergedBlockId] || blockId.isInstanceOf[ShuffleBlockChunkId])
logWarning(s"Falling back to fetch the original blocks for push-merged block $blockId")
+ shuffleMetrics.incFallbackCount(1)
Review Comment:
Fallback count measures the number of times fallback is initiated for a block which is either a `shuffleMergedBlock` or a `shuffleBlockChunk`. I don't understand why we need to move into the case that you suggested.
--
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
[GitHub] [spark] mridulm commented on a diff in pull request #36165: [SPARK-36620][SHUFFLE] Add Push Based Shuffle client side metrics
Posted by GitBox <gi...@apache.org>.
mridulm commented on code in PR #36165:
URL: https://github.com/apache/spark/pull/36165#discussion_r1008794924
##########
core/src/main/scala/org/apache/spark/storage/PushBasedFetchHelper.scala:
##########
@@ -290,6 +301,7 @@ private class PushBasedFetchHelper(
address: BlockManagerId): Unit = {
assert(blockId.isInstanceOf[ShuffleMergedBlockId] || blockId.isInstanceOf[ShuffleBlockChunkId])
logWarning(s"Falling back to fetch the original blocks for push-merged block $blockId")
+ shuffleMetrics.incFallbackCount(1)
Review Comment:
I misread the change, makes sense- 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: 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
[GitHub] [spark] otterc commented on a diff in pull request #36165: [SPARK-36620][SHUFFLE] Add Push Based Shuffle client side read metrics
Posted by GitBox <gi...@apache.org>.
otterc commented on code in PR #36165:
URL: https://github.com/apache/spark/pull/36165#discussion_r1044679570
##########
core/src/main/scala/org/apache/spark/storage/ShuffleBlockFetcherIterator.scala:
##########
@@ -726,6 +736,61 @@ final class ShuffleBlockFetcherIterator(
}
}
+ // Number of map blocks in a ShuffleBlockChunk
+ private def getShuffleChunkCardinality(blockId: ShuffleBlockChunkId): Int = {
+ pushBasedFetchHelper.getRoaringBitMap(blockId).map(_.getCardinality).getOrElse(0)
+ }
+
+ // Check if the merged block is local to the host or not
+ private def isLocalMergedBlockAddress(address: BlockManagerId): Boolean = {
+ address.executorId.isEmpty && address.host == blockManager.blockManagerId.host
Review Comment:
This method is there in PushBasedFetchHelper. `PushBasedFetchHelper.isLocalPushMergedBlockAddress`. We should try to just use that. We can maybe make that static. The executor id is not empty. It's `SHUFFLE_MERGER_IDENTIFIER`
--
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
[GitHub] [spark] otterc commented on a diff in pull request #36165: [SPARK-36620][SHUFFLE] Add Push Based Shuffle client side read metrics
Posted by GitBox <gi...@apache.org>.
otterc commented on code in PR #36165:
URL: https://github.com/apache/spark/pull/36165#discussion_r1046299412
##########
core/src/main/scala/org/apache/spark/status/LiveEntity.scala:
##########
@@ -843,6 +917,25 @@ private[spark] object LiveEntityHelpers {
m2.shuffleReadMetrics.remoteBytesReadToDisk * mult,
m1.shuffleReadMetrics.localBytesRead + m2.shuffleReadMetrics.localBytesRead * mult,
m1.shuffleReadMetrics.recordsRead + m2.shuffleReadMetrics.recordsRead * mult,
+ m1.shuffleReadMetrics.shufflePushReadMetrics.corruptMergedBlockChunks +
+ m2.shuffleReadMetrics.shufflePushReadMetrics.corruptMergedBlockChunks * mult,
+ m1.shuffleReadMetrics.shufflePushReadMetrics.mergedFetchFallbackCount +
+ m2.shuffleReadMetrics.shufflePushReadMetrics.mergedFetchFallbackCount * mult,
+ m1.shuffleReadMetrics.shufflePushReadMetrics.remoteMergedBlocksFetched +
+ m2.shuffleReadMetrics.shufflePushReadMetrics.remoteMergedBlocksFetched * mult,
+ m1.shuffleReadMetrics.shufflePushReadMetrics.localMergedBlocksFetched +
+ m2.shuffleReadMetrics.shufflePushReadMetrics.localMergedBlocksFetched * mult,
+ m1.shuffleReadMetrics.shufflePushReadMetrics.remoteMergedChunksFetched +
+ m2.shuffleReadMetrics.shufflePushReadMetrics.remoteMergedChunksFetched * mult,
+ m1.shuffleReadMetrics.shufflePushReadMetrics.localMergedChunksFetched +
+ m2.shuffleReadMetrics.shufflePushReadMetrics.localMergedChunksFetched * mult,
+ m1.shuffleReadMetrics.shufflePushReadMetrics.remoteMergedBytesRead +
+ m2.shuffleReadMetrics.shufflePushReadMetrics.remoteMergedBytesRead * mult,
+ m1.shuffleReadMetrics.shufflePushReadMetrics.localMergedBytesRead +
+ m2.shuffleReadMetrics.shufflePushReadMetrics.localMergedBytesRead * mult,
+ m1.shuffleReadMetrics.remoteReqsDuration + m2.shuffleReadMetrics.remoteReqsDuration * mult,
+ m1.shuffleReadMetrics.shufflePushReadMetrics.remoteMergedReqsDuration +
+ m2.shuffleReadMetrics.shufflePushReadMetrics.remoteMergedChunksFetched * mult,
Review Comment:
line 938: This should be `remoteMergedReqsDuration` instead of `remoteMergedChunksFetched`
--
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
[GitHub] [spark] mridulm commented on a diff in pull request #36165: [SPARK-36620][SHUFFLE] Add Push Based Shuffle client side read metrics
Posted by GitBox <gi...@apache.org>.
mridulm commented on code in PR #36165:
URL: https://github.com/apache/spark/pull/36165#discussion_r1041749200
##########
core/src/test/resources/HistoryServerExpectations/excludeOnFailure_node_for_stage_expectation.json:
##########
@@ -81,7 +93,19 @@
"remoteBytesRead" : 0,
"remoteBytesReadToDisk" : 0,
"localBytesRead" : 0,
- "recordsRead" : 0
+ "recordsRead" : 0,
+ "pushBased": {
Review Comment:
Yes, that would be better.
--
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
[GitHub] [spark] mridulm commented on pull request #36165: [SPARK-36620][SHUFFLE] Add Push Based Shuffle client side read metrics
Posted by GitBox <gi...@apache.org>.
mridulm commented on PR #36165:
URL: https://github.com/apache/spark/pull/36165#issuecomment-1373021422
Looks good to me, will wait for successful build before merging.
--
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
[GitHub] [spark] zhouyejoe commented on pull request #36165: [SPARK-36620][SHUFFLE] Add Push Based Shuffle client side read metrics
Posted by GitBox <gi...@apache.org>.
zhouyejoe commented on PR #36165:
URL: https://github.com/apache/spark/pull/36165#issuecomment-1373028152
LGTM. Thanks for working on this PR. @thejdeep
--
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
[GitHub] [spark] otterc commented on a diff in pull request #36165: [SPARK-36620][SHUFFLE] Add Push Based Shuffle client side read metrics
Posted by GitBox <gi...@apache.org>.
otterc commented on code in PR #36165:
URL: https://github.com/apache/spark/pull/36165#discussion_r1043701720
##########
core/src/main/scala/org/apache/spark/status/api/v1/api.scala:
##########
@@ -302,7 +312,9 @@ class StageData private[spark](
@JsonDeserialize(using = classOf[ExecutorMetricsJsonDeserializer])
val peakExecutorMetrics: Option[ExecutorMetrics],
val taskMetricsDistributions: Option[TaskMetricDistributions],
- val executorMetricsDistributions: Option[ExecutorMetricsDistributions])
+ val executorMetricsDistributions: Option[ExecutorMetricsDistributions],
+ val isPushBasedShuffleEnabled: Boolean,
Review Comment:
Nit: We have been referring to `pushBasedShuffle` in the OSS as just either `push` or `pushMerge`. I think we should remove `Based` from all the metrics related code. Rename this to `isShufflePushEnabled`?
##########
core/src/main/scala/org/apache/spark/status/storeTypes.scala:
##########
@@ -138,6 +138,16 @@ private[spark] object TaskIndexNames {
final val SHUFFLE_WRITE_RECORDS = "swr"
final val SHUFFLE_WRITE_SIZE = "sws"
final val SHUFFLE_WRITE_TIME = "swt"
+ final val SHUFFLE_REMOTE_REQS_DURATION = "srrd"
+ final val PUSH_BASED_SHUFFLE_CORRUPT_MERGED_BLOCK_CHUNKS = "pbscmbc"
+ final val PUSH_BASED_SHUFFLE_MERGED_FETCH_FALLBACK_COUNT = "pbsmffc"
+ final val PUSH_BASED_SHUFFLE_MERGED_REMOTE_BLOCKS = "pbsmrb"
+ final val PUSH_BASED_SHUFFLE_MERGED_LOCAL_BLOCKS = "pbsmlb"
+ final val PUSH_BASED_SHUFFLE_MERGED_REMOTE_CHUNKS = "pbsmrc"
+ final val PUSH_BASED_SHUFFLE_MERGED_LOCAL_CHUNKS = "pbsmlc"
+ final val PUSH_BASED_SHUFFLE_MERGED_REMOTE_READS = "pbsmrr"
+ final val PUSH_BASED_SHUFFLE_MERGED_LOCAL_READS = "pbsmlr"
+ final val PUSH_BASED_SHUFFLE_MERGED_REMOTE_REQS_DURATION = "pbsmrrd"
Review Comment:
Nit: Remove `BASED`
##########
core/src/main/scala/org/apache/spark/status/api/v1/api.scala:
##########
@@ -349,14 +361,27 @@ class OutputMetrics private[spark](
val bytesWritten: Long,
val recordsWritten: Long)
+class ShufflePushReadMetrics private[spark](
+ val corruptMergedBlockChunks: Long,
+ val mergedFetchFallbackCount: Long,
+ val remoteMergedBlocksFetched: Long,
+ val localMergedBlocksFetched: Long,
+ val remoteMergedChunksFetched: Long,
+ val localMergedChunksFetched: Long,
+ val remoteMergedBytesRead: Long,
+ val localMergedBytesRead: Long,
+ val remoteMergedReqsDuration: Long)
+
class ShuffleReadMetrics private[spark](
val remoteBlocksFetched: Long,
val localBlocksFetched: Long,
val fetchWaitTime: Long,
val remoteBytesRead: Long,
val remoteBytesReadToDisk: Long,
val localBytesRead: Long,
- val recordsRead: Long)
+ val recordsRead: Long,
+ val remoteReqsDuration: Long,
+ val pushBasedShuffle: ShufflePushReadMetrics)
Review Comment:
Nit: rename to `shufflePushReadMetrics`
##########
core/src/main/scala/org/apache/spark/executor/ExecutorSource.scala:
##########
@@ -115,6 +115,26 @@ class ExecutorSource(
metricRegistry.counter(MetricRegistry.name("shuffleBytesWritten"))
val METRIC_SHUFFLE_RECORDS_WRITTEN =
metricRegistry.counter(MetricRegistry.name("shuffleRecordsWritten"))
+ val METRIC_SHUFFLE_REMOTE_REQS_DURATION =
+ metricRegistry.counter(MetricRegistry.name("shuffleRemoteReqsDuration"))
+ val METRIC_PUSH_BASED_SHUFFLE_CORRUPT_MERGED_BLOCK_CHUNKS =
+ metricRegistry.counter(MetricRegistry.name("shuffleCorruptMergedBlockChunks"))
+ val METRIC_PUSH_BASED_SHUFFLE_MERGED_FETCH_FALLBACK_COUNT =
+ metricRegistry.counter(MetricRegistry.name("shuffleMergedFetchFallbackCount"))
+ val METRIC_PUSH_BASED_SHUFFLE_MERGED_REMOTE_BLOCKS_FETCHED =
+ metricRegistry.counter(MetricRegistry.name("shuffleMergedRemoteBlocksFetched"))
+ val METRIC_PUSH_BASED_SHUFFLE_MERGED_LOCAL_BLOCKS_FETCHED =
+ metricRegistry.counter(MetricRegistry.name("shuffleMergedLocalBlocksFetched"))
+ val METRIC_PUSH_BASED_SHUFFLE_MERGED_REMOTE_CHUNKS_FETCHED =
+ metricRegistry.counter(MetricRegistry.name("shuffleMergedRemoteChunksFetched"))
+ val METRIC_PUSH_BASED_SHUFFLE_MERGED_LOCAL_CHUNKS_FETCHED =
+ metricRegistry.counter(MetricRegistry.name("shuffleMergedLocalChunksFetched"))
+ val METRIC_PUSH_BASED_SHUFFLE_MERGED_REMOTE_BYTES_READ =
+ metricRegistry.counter(MetricRegistry.name("shuffleMergedRemoteBytesRead"))
+ val METRIC_PUSH_BASED_SHUFFLE_MERGED_LOCAL_BYTES_READ =
+ metricRegistry.counter(MetricRegistry.name("shuffleMergedLocalBytesRead"))
+ val METRIC_PUSH_BASED_SHUFFLE_MERGED_REMOTE_REQS_DURATION =
Review Comment:
Nit: Remove `BASED`
##########
core/src/main/scala/org/apache/spark/status/api/v1/api.scala:
##########
@@ -436,7 +472,9 @@ class ShuffleReadMetricDistributions private[spark](
val fetchWaitTime: IndexedSeq[Double],
val remoteBytesRead: IndexedSeq[Double],
val remoteBytesReadToDisk: IndexedSeq[Double],
- val totalBlocksFetched: IndexedSeq[Double])
+ val totalBlocksFetched: IndexedSeq[Double],
+ val remoteReqsDuration: IndexedSeq[Double],
+ val pushBasedShuffle: ShufflePushReadMetricDistributions)
Review Comment:
Nit: same here. rename to `shufflePushMetricsDist`
##########
core/src/main/scala/org/apache/spark/status/LiveEntity.scala:
##########
@@ -843,6 +917,25 @@ private[spark] object LiveEntityHelpers {
m2.shuffleReadMetrics.remoteBytesReadToDisk * mult,
m1.shuffleReadMetrics.localBytesRead + m2.shuffleReadMetrics.localBytesRead * mult,
m1.shuffleReadMetrics.recordsRead + m2.shuffleReadMetrics.recordsRead * mult,
+ m1.shuffleReadMetrics.pushBasedShuffle.corruptMergedBlockChunks +
+ m2.shuffleReadMetrics.pushBasedShuffle.corruptMergedBlockChunks * mult,
+ m1.shuffleReadMetrics.pushBasedShuffle.mergedFetchFallbackCount +
+ m2.shuffleReadMetrics.pushBasedShuffle.mergedFetchFallbackCount * mult,
+ m1.shuffleReadMetrics.pushBasedShuffle.remoteMergedBlocksFetched +
+ m2.shuffleReadMetrics.pushBasedShuffle.remoteMergedBlocksFetched * mult,
+ m1.shuffleReadMetrics.pushBasedShuffle.localMergedBlocksFetched +
+ m2.shuffleReadMetrics.pushBasedShuffle.localMergedBlocksFetched * mult,
+ m1.shuffleReadMetrics.pushBasedShuffle.remoteMergedChunksFetched +
+ m2.shuffleReadMetrics.pushBasedShuffle.remoteMergedChunksFetched * mult,
+ m1.shuffleReadMetrics.pushBasedShuffle.localMergedChunksFetched +
+ m2.shuffleReadMetrics.pushBasedShuffle.localMergedChunksFetched * mult,
+ m1.shuffleReadMetrics.pushBasedShuffle.remoteMergedBytesRead +
+ m2.shuffleReadMetrics.pushBasedShuffle.remoteMergedBytesRead * mult,
+ m1.shuffleReadMetrics.pushBasedShuffle.localMergedBytesRead +
+ m2.shuffleReadMetrics.pushBasedShuffle.localMergedBytesRead * mult,
+ m1.shuffleReadMetrics.remoteReqsDuration + m2.shuffleReadMetrics.remoteReqsDuration * mult,
+ m1.shuffleReadMetrics.pushBasedShuffle.remoteMergedReqsDuration +
+ m2.shuffleReadMetrics.pushBasedShuffle.remoteMergedChunksFetched * mult,
Review Comment:
This should be `remoteMergedReqsDuration` instead of `remoteMergedChunksFetched`
##########
core/src/main/scala/org/apache/spark/status/AppStatusStore.scala:
##########
@@ -405,11 +416,42 @@ private[spark] class AppStatusStore(
},
scanTasks(TaskIndexNames.SHUFFLE_TOTAL_BLOCKS) { m =>
m.shuffleLocalBlocksFetched + m.shuffleRemoteBlocksFetched
- }),
- shuffleWriteMetrics = new v1.ShuffleWriteMetricDistributions(
- scanTasks(TaskIndexNames.SHUFFLE_WRITE_SIZE) { t => t.shuffleBytesWritten },
- scanTasks(TaskIndexNames.SHUFFLE_WRITE_RECORDS) { t => t.shuffleRecordsWritten },
- scanTasks(TaskIndexNames.SHUFFLE_WRITE_TIME) { t => t.shuffleWriteTime }))
+ },
+ scanTasks(TaskIndexNames.SHUFFLE_REMOTE_REQS_DURATION) {
+ t => t.shuffleRemoteReqsDuration
+ },
+ new v1.ShufflePushReadMetricDistributions(
+ scanTasks(TaskIndexNames.PUSH_BASED_SHUFFLE_CORRUPT_MERGED_BLOCK_CHUNKS) { t =>
+ t.shuffleCorruptMergedBlockChunks
+ },
+ scanTasks(TaskIndexNames.PUSH_BASED_SHUFFLE_MERGED_FETCH_FALLBACK_COUNT) {
+ t => t.shuffleMergedFetchFallbackCount
+ },
+ scanTasks(TaskIndexNames.PUSH_BASED_SHUFFLE_MERGED_REMOTE_BLOCKS) { t =>
+ t.shuffleMergedRemoteBlocksFetched
+ },
+ scanTasks(TaskIndexNames.PUSH_BASED_SHUFFLE_MERGED_LOCAL_BLOCKS) { t =>
+ t.shuffleMergedLocalBlocksFetched
+ },
+ scanTasks(TaskIndexNames.PUSH_BASED_SHUFFLE_MERGED_REMOTE_CHUNKS) { t =>
+ t.shuffleMergedRemoteChunksFetched
+ },
+ scanTasks(TaskIndexNames.PUSH_BASED_SHUFFLE_MERGED_LOCAL_CHUNKS) { t =>
+ t.shuffleMergedLocalChunksFetched
+ },
+ scanTasks(TaskIndexNames.PUSH_BASED_SHUFFLE_MERGED_REMOTE_READS) { t =>
+ t.shuffleMergedRemoteBytesRead
+ },
+ scanTasks(TaskIndexNames.PUSH_BASED_SHUFFLE_MERGED_LOCAL_READS) { t =>
+ t.shuffleMergedLocalBytesRead
+ },
+ scanTasks(TaskIndexNames.PUSH_BASED_SHUFFLE_MERGED_REMOTE_REQS_DURATION) { t =>
+ t.shuffleMergedRemoteReqDuration
+ })),
Review Comment:
Nit: Remove `BASED` from TaskIndexesNames
--
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
[GitHub] [spark] mridulm commented on a diff in pull request #36165: [SPARK-36620][SHUFFLE] Add Push Based Shuffle client side metrics
Posted by GitBox <gi...@apache.org>.
mridulm commented on code in PR #36165:
URL: https://github.com/apache/spark/pull/36165#discussion_r1014392620
##########
core/src/test/scala/org/apache/spark/util/JsonProtocolSuite.scala:
##########
@@ -1780,7 +1802,19 @@ private[spark] object JsonProtocolSuite extends Assertions {
| "Remote Bytes Read" : 0,
| "Remote Bytes Read To Disk" : 0,
| "Local Bytes Read" : 0,
- | "Total Records Read" : 0
+ | "Total Records Read" : 0,
+ | "Remote Requests Duration": 0,
+ | "Push Based": {
Review Comment:
+CC @zhouyejoe
--
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
[GitHub] [spark] mridulm commented on a diff in pull request #36165: [SPARK-36620][SHUFFLE] Add Push Based Shuffle client side read metrics
Posted by GitBox <gi...@apache.org>.
mridulm commented on code in PR #36165:
URL: https://github.com/apache/spark/pull/36165#discussion_r1044823829
##########
core/src/main/scala/org/apache/spark/status/api/v1/api.scala:
##########
@@ -302,7 +312,9 @@ class StageData private[spark](
@JsonDeserialize(using = classOf[ExecutorMetricsJsonDeserializer])
val peakExecutorMetrics: Option[ExecutorMetrics],
val taskMetricsDistributions: Option[TaskMetricDistributions],
- val executorMetricsDistributions: Option[ExecutorMetricsDistributions])
+ val executorMetricsDistributions: Option[ExecutorMetricsDistributions],
+ val isPushBasedShuffleEnabled: Boolean,
Review Comment:
We actually have a lot of `pushBased*` references in the codebase.
Given the block id's are named shufflePush, that might be a good prefix to settle on.
--
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
[GitHub] [spark] mridulm commented on a diff in pull request #36165: [SPARK-36620][SHUFFLE] Add Push Based Shuffle client side read metrics
Posted by GitBox <gi...@apache.org>.
mridulm commented on code in PR #36165:
URL: https://github.com/apache/spark/pull/36165#discussion_r1058701274
##########
core/src/main/protobuf/org/apache/spark/status/protobuf/store_types.proto:
##########
@@ -100,11 +100,21 @@ message TaskDataWrapper {
int64 shuffle_remote_bytes_read_to_disk = 34;
int64 shuffle_local_bytes_read = 35;
int64 shuffle_records_read = 36;
- int64 shuffle_bytes_written = 37;
- int64 shuffle_write_time = 38;
- int64 shuffle_records_written = 39;
- int64 stage_id = 40;
- int32 stage_attempt_id = 41;
+ int64 shuffle_corrupt_merged_block_chunks = 37;
+ int64 shuffle_merged_fetch_fallback_count = 38;
+ int64 shuffle_merged_remote_blocks_fetched = 39;
+ int64 shuffle_merged_local_blocks_fetched = 40;
+ int64 shuffle_merged_remote_chunks_fetched = 41;
+ int64 shuffle_merged_local_chunks_fetched = 42;
+ int64 shuffle_merged_remote_bytes_read = 43;
+ int64 shuffle_merged_local_bytes_read = 44;
+ int64 shuffle_remote_reqs_duration = 45;
+ int64 shuffle_merged_remote_req_duration = 46;
+ int64 shuffle_bytes_written = 47;
+ int64 shuffle_write_time = 48;
+ int64 shuffle_records_written = 49;
+ int64 stage_id = 50;
+ int32 stage_attempt_id = 51;
Review Comment:
Protobuf id's should never be changed.
--
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
[GitHub] [spark] mridulm commented on a diff in pull request #36165: [SPARK-36620][SHUFFLE] Add Push Based Shuffle client side metrics
Posted by GitBox <gi...@apache.org>.
mridulm commented on code in PR #36165:
URL: https://github.com/apache/spark/pull/36165#discussion_r978229580
##########
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 FALLBACK_COUNT = SHUFFLE_PUSH_READ_METRICS_PREFIX + "fallbackCount"
+ val REMOTE_MERGED_BLOCKS_FETCHED = SHUFFLE_PUSH_READ_METRICS_PREFIX + "remoteMergedBlocksFetched"
+ val LOCAL_MERGED_BLOCKS_FETCHED = SHUFFLE_PUSH_READ_METRICS_PREFIX + "localMergedBlocksFetched"
+ val REMOTE_MERGED_CHUNKS_FETCHED = SHUFFLE_PUSH_READ_METRICS_PREFIX + "remoteMergedChunksFetched"
+ val LOCAL_MERGED_CHUNKS_FETCHED = SHUFFLE_PUSH_READ_METRICS_PREFIX + "localMergedChunksFetched"
+ val REMOTE_MERGED_BLOCKS_READ = SHUFFLE_PUSH_READ_METRICS_PREFIX + "remoteMergedBytesRead"
+ val LOCAL_MERGED_BLOCKS_READ = SHUFFLE_PUSH_READ_METRICS_PREFIX + "localMergedBytesRead"
+ val REMOTE_MERGED_REQS_DURATION = SHUFFLE_PUSH_READ_METRICS_PREFIX + "remoteMergedReqsDuration"
Review Comment:
Can we check if all the metrics are getting populated ?
A cursory indicated the following are not populated:
* `remoteReqsDuration`
* `remoteMergedReqsDuration`
Also, make `remoteMergedBytesRead` and `localMergedBytesRead` consistent with the name of its corresponding accumulator ?
##########
core/src/main/scala/org/apache/spark/status/storeTypes.scala:
##########
@@ -233,6 +243,38 @@ private[spark] class TaskDataWrapper(
val shuffleLocalBytesRead: Long,
@KVIndexParam(value = TaskIndexNames.SHUFFLE_READ_RECORDS, parent = TaskIndexNames.STAGE)
val shuffleRecordsRead: Long,
+ @KVIndexParam(
+ value = TaskIndexNames.PUSH_BASED_SHUFFLE_CORRUPT_MERGED_BLOCK_CHUNKS,
+ parent = TaskIndexNames.STAGE)
+ val shuffleCorruptMergedBlockChunks: Long,
+ @KVIndexParam(value = TaskIndexNames.PUSH_BASED_SHUFFLE_FALLBACK_COUNT,
+ parent = TaskIndexNames.STAGE)
+ val shuffleFallbackCount: Long,
+ @KVIndexParam(
+ value = TaskIndexNames.PUSH_BASED_SHUFFLE_MERGED_REMOTE_BLOCKS, parent = TaskIndexNames.STAGE)
+ val shuffleMergedRemoteBlocksFetched: Long,
+ @KVIndexParam(
+ value = TaskIndexNames.PUSH_BASED_SHUFFLE_MERGED_LOCAL_BLOCKS, parent = TaskIndexNames.STAGE)
+ val shuffleMergedLocalBlocksFetched: Long,
+ @KVIndexParam(
+ value = TaskIndexNames.PUSH_BASED_SHUFFLE_MERGED_REMOTE_CHUNKS, parent = TaskIndexNames.STAGE)
+ val shuffleMergedRemoteChunksFetched: Long,
+ @KVIndexParam(
+ value = TaskIndexNames.PUSH_BASED_SHUFFLE_MERGED_LOCAL_CHUNKS, parent = TaskIndexNames.STAGE)
+ val shuffleMergedLocalChunksFetched: Long,
+ @KVIndexParam(
+ value = TaskIndexNames.PUSH_BASED_SHUFFLE_MERGED_REMOTE_READS, parent = TaskIndexNames.STAGE)
+ val shuffleMergedRemoteBlocksBytesRead: Long,
+ @KVIndexParam(
+ value = TaskIndexNames.PUSH_BASED_SHUFFLE_MERGED_LOCAL_READS, parent = TaskIndexNames.STAGE)
+ val shuffleMergedLocalBlocksBytesRead: Long,
+ @KVIndexParam(
+ value = TaskIndexNames.PUSH_BASED_SHUFFLE_REMOTE_REQS_DURATION, parent = TaskIndexNames.STAGE)
+ val shuffleRemoteReqsDuration: Long,
+ @KVIndexParam(
+ value = TaskIndexNames.PUSH_BASED_SHUFFLE_MERGED_REMOTE_REQS_DURATION,
+ parent = TaskIndexNames.STAGE)
Review Comment:
Do we need to index any of these ? None of these are sortable columns in the UI right ?
##########
core/src/main/scala/org/apache/spark/executor/Executor.scala:
##########
@@ -654,6 +654,27 @@ private[spark] class Executor(
executorSource.METRIC_RESULT_SIZE.inc(task.metrics.resultSize)
executorSource.METRIC_DISK_BYTES_SPILLED.inc(task.metrics.diskBytesSpilled)
executorSource.METRIC_MEMORY_BYTES_SPILLED.inc(task.metrics.memoryBytesSpilled)
+ executorSource.METRIC_PUSH_BASED_SHUFFLE_CORRUPT_MERGED_BLOCK_CHUNKS
+ .inc(task.metrics.shuffleReadMetrics.corruptMergedBlockChunks)
+ executorSource.METRIC_PUSH_BASED_SHUFFLE_FALLBACK_COUNT
+ .inc(task.metrics.shuffleReadMetrics.fallbackCount)
+ executorSource.METRIC_PUSH_BASED_SHUFFLE_MERGED_REMOTE_BLOCKS_FETCHED
+ .inc(task.metrics.shuffleReadMetrics.remoteMergedBlocksFetched)
+ executorSource.METRIC_PUSH_BASED_SHUFFLE_MERGED_LOCAL_BLOCKS_FETCHED
+ .inc(task.metrics.shuffleReadMetrics.localMergedBlocksFetched)
+ executorSource.METRIC_PUSH_BASED_SHUFFLE_MERGED_REMOTE_CHUNKS_FETCHED
+ .inc(task.metrics.shuffleReadMetrics.remoteMergedChunksFetched)
+ executorSource.METRIC_PUSH_BASED_SHUFFLE_MERGED_LOCAL_CHUNKS_FETCHED
+ .inc(task.metrics.shuffleReadMetrics.localMergedChunksFetched)
+ executorSource.METRIC_PUSH_BASED_SHUFFLE_MERGED_REMOTE_BYTES_READ
+ .inc(task.metrics.shuffleReadMetrics.remoteMergedBlocksBytesRead)
+
+ executorSource.METRIC_PUSH_BASED_SHUFFLE_MERGED_LOCAL_BLOCKS_FETCHED
+ .inc(task.metrics.shuffleReadMetrics.localMergedBlocksBytesRead)
+ executorSource.METRIC_PUSH_BASED_SHUFFLE_REMOTE_REQS_DURATION
+ .inc(task.metrics.shuffleReadMetrics.remoteReqsDuration)
+ executorSource.METRIC_PUSH_BASED_SHUFFLE_MERGED_REMOTE_REQS_DURATION
+ .inc(task.metrics.shuffleReadMetrics.remoteMergedReqsDuration)
Review Comment:
Given the number of shuffle related metrics getting updated now - move all `METRIC_*SHUFFLE_*` metric updates to a different method ? Both the newly introduced metrics and the existing `METRIC_SHUFFLE_` metrics
##########
core/src/main/scala/org/apache/spark/storage/PushBasedFetchHelper.scala:
##########
@@ -43,7 +44,8 @@ private class PushBasedFetchHelper(
private val iterator: ShuffleBlockFetcherIterator,
private val shuffleClient: BlockStoreClient,
private val blockManager: BlockManager,
- private val mapOutputTracker: MapOutputTracker) extends Logging {
+ private val mapOutputTracker: MapOutputTracker,
+ private val shuffleMetrics: ShuffleReadMetricsReporter) extends Logging {
Review Comment:
fix indentation
##########
core/src/main/scala/org/apache/spark/storage/ShuffleBlockFetcherIterator.scala:
##########
@@ -722,6 +718,63 @@ final class ShuffleBlockFetcherIterator(
}
}
+ // Number of map blocks in a ShuffleBlockChunk
+ private def getShuffleChunkCardinality(blockId: ShuffleBlockChunkId): Int = {
+ val chunkTracker = pushBasedFetchHelper.getRoaringBitMap(blockId)
+ chunkTracker match {
+ case Some(bitmap) => bitmap.getCardinality
+ case None => 0
+ }
+ }
+
+ // Check if the merged block is local to the host or not
+ private def isLocalMergedBlockAddress(address: BlockManagerId): Boolean = {
+ address.executorId.isEmpty && address.host == blockManager.blockManagerId.host
+ }
+
+ private def shuffleMetricsUpdate(
+ blockId: BlockId,
+ buf: ManagedBuffer,
+ isLocal: Boolean): Unit = {
Review Comment:
nit: `isLocal` -> `local`
##########
core/src/main/scala/org/apache/spark/storage/ShuffleBlockFetcherIterator.scala:
##########
@@ -722,6 +718,63 @@ final class ShuffleBlockFetcherIterator(
}
}
+ // Number of map blocks in a ShuffleBlockChunk
+ private def getShuffleChunkCardinality(blockId: ShuffleBlockChunkId): Int = {
+ val chunkTracker = pushBasedFetchHelper.getRoaringBitMap(blockId)
+ chunkTracker match {
+ case Some(bitmap) => bitmap.getCardinality
+ case None => 0
+ }
+ }
+
+ // Check if the merged block is local to the host or not
+ private def isLocalMergedBlockAddress(address: BlockManagerId): Boolean = {
+ address.executorId.isEmpty && address.host == blockManager.blockManagerId.host
+ }
+
+ private def shuffleMetricsUpdate(
+ blockId: BlockId,
+ buf: ManagedBuffer,
+ isLocal: Boolean): Unit = {
+ if (isLocal) {
+ shuffleLocalMetricsUpdate(blockId, buf)
+ } else {
+ shuffleRemoteMetricsUpdate(blockId, buf)
+ }
+ }
+
+ private def shuffleLocalMetricsUpdate(blockId: BlockId, buf: ManagedBuffer): Unit = {
+ // Check if the block is within the host-local blocks set, or if it is a merged local
+ // block. In either case, it is local read and we need to increase the local
+ // shuffle read metrics.
+ blockId match {
+ case chunkId: ShuffleBlockChunkId =>
+ shuffleMetrics.incLocalMergedChunksFetched(1)
+ shuffleMetrics.incLocalMergedBlocksFetched(getShuffleChunkCardinality(chunkId))
+ shuffleMetrics.incLocalMergedBlocksBytesRead(buf.size)
+ shuffleMetrics.incLocalBlocksFetched(getShuffleChunkCardinality(chunkId))
Review Comment:
Compute `getShuffleChunkCardinality` once and reuse for both invocation. Here and in `shuffleRemoteMetricsUpdate` below.
##########
core/src/test/scala/org/apache/spark/storage/ShuffleBlockFetcherIteratorSuite.scala:
##########
@@ -1219,7 +1219,12 @@ class ShuffleBlockFetcherIteratorSuite extends SparkFunSuite with PrivateMethodT
new RuntimeException("forced error"))
}
Review Comment:
Also update the other tests, particularly related to push based shuffle to assert on the expected metrics ?
Ideally, the original tests should have done it (if the metrics had been added initially) - and so now we have to retroactively fix them ...
##########
core/src/test/scala/org/apache/spark/util/JsonProtocolSuite.scala:
##########
@@ -1780,7 +1802,19 @@ private[spark] object JsonProtocolSuite extends Assertions {
| "Remote Bytes Read" : 0,
| "Remote Bytes Read To Disk" : 0,
| "Local Bytes Read" : 0,
- | "Total Records Read" : 0
+ | "Total Records Read" : 0,
+ | "Remote Requests Duration": 0,
+ | "Push Based": {
Review Comment:
Rename this to "Push Based Shuffle" instead ?
It is within the context of `Shuffle Read Metrics`, but makes it more clear.
Thoughts ?
+CC @otterc, @Ngone51
##########
core/src/test/scala/org/apache/spark/ui/StagePageSuite.scala:
##########
@@ -78,6 +78,16 @@ class StagePageSuite extends SparkFunSuite with LocalSparkContext {
shuffleLocalBytesRead = 1L,
shuffleReadBytes = 1L,
shuffleReadRecords = 1L,
+ shuffleCorruptMergedBlockChunks = 2L,
+ shuffleFallbackCount = 2L,
+ shuffleMergedRemoteBlocksFetched = 1L,
+ shuffleMergedLocalBlocksFetched = 1L,
+ shuffleMergedRemoteChunksFetched = 1L,
+ shuffleMergedLocalChunksFetched = 1L,
+ shuffleMergedRemoteBytesRead = 1L,
+ shuffleMergedLocalBytesRead = 1L,
+ shuffleRemoteReqsDuration = 1L,
+ shuffleMergedRemoteReqsDuration = 1L,
Review Comment:
Any particular reason these are set to `1` ?
Particularly when `shuffleMergersCount` == 0 ?
##########
core/src/test/scala/org/apache/spark/util/JsonProtocolSuite.scala:
##########
@@ -1653,7 +1663,19 @@ private[spark] object JsonProtocolSuite extends Assertions {
| "Remote Bytes Read": 1000,
| "Remote Bytes Read To Disk": 400,
| "Local Bytes Read": 1100,
- | "Total Records Read": 10
+ | "Total Records Read": 10,
+ | "Remote Requests Duration": 900,
+ | "Push Based": {
+ | "Corrupt Merged Block Chunks" : 100,
+ | "Fallback Count" : 100,
Review Comment:
Curious why these are nonzero if push based shuffle was disabled ?
Are these simply values to test serde ?
##########
core/src/main/scala/org/apache/spark/storage/ShuffleBlockFetcherIterator.scala:
##########
@@ -233,15 +233,11 @@ final class ShuffleBlockFetcherIterator(
result match {
case SuccessFetchResult(blockId, mapIndex, address, _, buf, _) =>
if (address != blockManager.blockManagerId) {
- if (hostLocalBlocks.contains(blockId -> mapIndex)) {
- shuffleMetrics.incLocalBlocksFetched(1)
- shuffleMetrics.incLocalBytesRead(buf.size)
+ if (hostLocalBlocks.contains(blockId -> mapIndex)
+ || isLocalMergedBlockAddress(address)) {
Review Comment:
super nit: check for `isLocalMergedBlockAddress` first ?
##########
core/src/main/scala/org/apache/spark/storage/PushBasedFetchHelper.scala:
##########
@@ -290,6 +301,7 @@ private class PushBasedFetchHelper(
address: BlockManagerId): Unit = {
assert(blockId.isInstanceOf[ShuffleMergedBlockId] || blockId.isInstanceOf[ShuffleBlockChunkId])
logWarning(s"Falling back to fetch the original blocks for push-merged block $blockId")
+ shuffleMetrics.incFallbackCount(1)
Review Comment:
This should be move into the `case shuffleBlockId: ShuffleMergedBlockId` block below.
Else, for the `case _` block, we will be off by `1`
##########
core/src/main/scala/org/apache/spark/status/storeTypes.scala:
##########
@@ -138,6 +138,16 @@ private[spark] object TaskIndexNames {
final val SHUFFLE_WRITE_RECORDS = "swr"
final val SHUFFLE_WRITE_SIZE = "sws"
final val SHUFFLE_WRITE_TIME = "swt"
+ final val PUSH_BASED_SHUFFLE_CORRUPT_MERGED_BLOCK_CHUNKS = "scmbc"
+ final val PUSH_BASED_SHUFFLE_FALLBACK_COUNT = "sfc"
+ final val PUSH_BASED_SHUFFLE_MERGED_REMOTE_BLOCKS = "smrb"
+ final val PUSH_BASED_SHUFFLE_MERGED_LOCAL_BLOCKS = "smlb"
+ final val PUSH_BASED_SHUFFLE_MERGED_REMOTE_CHUNKS = "smrc"
+ final val PUSH_BASED_SHUFFLE_MERGED_LOCAL_CHUNKS = "smlc"
+ final val PUSH_BASED_SHUFFLE_MERGED_REMOTE_READS = "smrr"
+ final val PUSH_BASED_SHUFFLE_MERGED_LOCAL_READS = "smlr"
+ final val PUSH_BASED_SHUFFLE_REMOTE_REQS_DURATION = "srrd"
+ final val PUSH_BASED_SHUFFLE_MERGED_REMOTE_REQS_DURATION = "smrrd"
Review Comment:
add `pb` as prefix ?
##########
core/src/test/scala/org/apache/spark/util/JsonProtocolSuite.scala:
##########
@@ -2623,48 +2677,118 @@ private[spark] object JsonProtocolSuite extends Assertions {
| },
| {
| "ID": 19,
+ | "Name": "${shuffleRead.FALLBACK_COUNT}",
+ | "Update": 0,
+ | "Internal": true,
+ | "Count Failed Values": true
+ | },
+ | {
+ | "ID" : 20,
+ | "Name" : "${shuffleRead.REMOTE_MERGED_BLOCKS_FETCHED}",
+ | "Update" : 0,
+ | "Internal" : true,
+ | "Count Failed Values" : true
+ | },
+ | {
+ | "ID" : 21,
+ | "Name" : "${shuffleRead.LOCAL_MERGED_BLOCKS_FETCHED}",
+ | "Update" : 0,
+ | "Internal" : true,
+ | "Count Failed Values" : true
+ | },
+ | {
+ | "ID" : 22,
+ | "Name" : "${shuffleRead.REMOTE_MERGED_CHUNKS_FETCHED}",
+ | "Update" : 0,
+ | "Internal" : true,
+ | "Count Failed Values" : true
+ | },
+ | {
+ | "ID" : 23,
+ | "Name" : "${shuffleRead.LOCAL_MERGED_CHUNKS_FETCHED}",
+ | "Update" : 0,
+ | "Internal" : true,
+ | "Count Failed Values" : true
+ | },
+ | {
+ | "ID" : 24,
+ | "Name" : "${shuffleRead.REMOTE_MERGED_BLOCKS_READ}",
+ | "Update" : 0,
+ | "Internal" : true,
+ | "Count Failed Values" : true
+ | },
+ | {
+ | "ID" : 25,
+ | "Name" : "${shuffleRead.LOCAL_MERGED_BLOCKS_READ}",
+ | "Update" : 0,
+ | "Internal" : true,
+ | "Count Failed Values" : true
+ | },
+ | {
+ | "ID" : 26,
+ | "Name" : "${shuffleRead.REMOTE_REQS_DURATION}",
+ | "Update" : 0,
+ | "Internal" : true,
+ | "Count Failed Values" : true
+ | },
+ | {
+ | "ID" : 27,
+ | "Name" : "${shuffleRead.REMOTE_MERGED_REQS_DURATION}",
+ | "Update" : 0,
+ | "Internal" : true,
+ | "Count Failed Values" : true
+ | },
+ | {
+ | "ID": 28,
+ | "Name": "${shuffleWrite.BYTES_WRITTEN}",
+ | "Update": 0,
+ | "Internal": true,
+ | "Count Failed Values": true
+ | },
+ | {
+ | "ID": 29,
| "Name": "${shuffleWrite.RECORDS_WRITTEN}",
| "Update": 0,
| "Internal": true,
| "Count Failed Values": true
| },
| {
- | "ID": 20,
+ | "ID": 30,
| "Name": "${shuffleWrite.WRITE_TIME}",
| "Update": 0,
| "Internal": true,
| "Count Failed Values": true
| },
| {
- | "ID": 21,
+ | "ID": 31,
| "Name": "${input.BYTES_READ}",
| "Update": 2100,
| "Internal": true,
| "Count Failed Values": true
| },
| {
- | "ID": 22,
+ | "ID": 32,
| "Name": "${input.RECORDS_READ}",
| "Update": 21,
| "Internal": true,
| "Count Failed Values": true
| },
| {
- | "ID": 23,
+ | "ID": 33,
| "Name": "${output.BYTES_WRITTEN}",
| "Update": 1200,
| "Internal": true,
| "Count Failed Values": true
| },
| {
- | "ID": 24,
+ | "ID": 34,
| "Name": "${output.RECORDS_WRITTEN}",
Review Comment:
Curious why some of these values changed.
I am yet to look at tests more carefully though ...
--
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
[GitHub] [spark] mridulm commented on a diff in pull request #36165: [SPARK-36620][SHUFFLE] Add Push Based Shuffle client side metrics
Posted by GitBox <gi...@apache.org>.
mridulm commented on code in PR #36165:
URL: https://github.com/apache/spark/pull/36165#discussion_r1008794591
##########
core/src/main/scala/org/apache/spark/status/storeTypes.scala:
##########
@@ -233,6 +243,38 @@ private[spark] class TaskDataWrapper(
val shuffleLocalBytesRead: Long,
@KVIndexParam(value = TaskIndexNames.SHUFFLE_READ_RECORDS, parent = TaskIndexNames.STAGE)
val shuffleRecordsRead: Long,
+ @KVIndexParam(
+ value = TaskIndexNames.PUSH_BASED_SHUFFLE_CORRUPT_MERGED_BLOCK_CHUNKS,
+ parent = TaskIndexNames.STAGE)
+ val shuffleCorruptMergedBlockChunks: Long,
+ @KVIndexParam(value = TaskIndexNames.PUSH_BASED_SHUFFLE_FALLBACK_COUNT,
+ parent = TaskIndexNames.STAGE)
+ val shuffleFallbackCount: Long,
+ @KVIndexParam(
+ value = TaskIndexNames.PUSH_BASED_SHUFFLE_MERGED_REMOTE_BLOCKS, parent = TaskIndexNames.STAGE)
+ val shuffleMergedRemoteBlocksFetched: Long,
+ @KVIndexParam(
+ value = TaskIndexNames.PUSH_BASED_SHUFFLE_MERGED_LOCAL_BLOCKS, parent = TaskIndexNames.STAGE)
+ val shuffleMergedLocalBlocksFetched: Long,
+ @KVIndexParam(
+ value = TaskIndexNames.PUSH_BASED_SHUFFLE_MERGED_REMOTE_CHUNKS, parent = TaskIndexNames.STAGE)
+ val shuffleMergedRemoteChunksFetched: Long,
+ @KVIndexParam(
+ value = TaskIndexNames.PUSH_BASED_SHUFFLE_MERGED_LOCAL_CHUNKS, parent = TaskIndexNames.STAGE)
+ val shuffleMergedLocalChunksFetched: Long,
+ @KVIndexParam(
+ value = TaskIndexNames.PUSH_BASED_SHUFFLE_MERGED_REMOTE_READS, parent = TaskIndexNames.STAGE)
+ val shuffleMergedRemoteBlocksBytesRead: Long,
+ @KVIndexParam(
+ value = TaskIndexNames.PUSH_BASED_SHUFFLE_MERGED_LOCAL_READS, parent = TaskIndexNames.STAGE)
+ val shuffleMergedLocalBlocksBytesRead: Long,
+ @KVIndexParam(
+ value = TaskIndexNames.PUSH_BASED_SHUFFLE_REMOTE_REQS_DURATION, parent = TaskIndexNames.STAGE)
+ val shuffleRemoteReqsDuration: Long,
+ @KVIndexParam(
+ value = TaskIndexNames.PUSH_BASED_SHUFFLE_MERGED_REMOTE_REQS_DURATION,
+ parent = TaskIndexNames.STAGE)
Review Comment:
Makes sense, thanks for clarifying !
--
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
[GitHub] [spark] mridulm commented on pull request #36165: [SPARK-36620][SHUFFLE] Add Push Based Shuffle client side read metrics
Posted by GitBox <gi...@apache.org>.
mridulm commented on PR #36165:
URL: https://github.com/apache/spark/pull/36165#issuecomment-1329952299
Please don't force push the code, unless it is to resolve conflicts/etc which can be handled otherwise - it makes reviewing what changed much harder
--
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
[GitHub] [spark] thejdeep commented on a diff in pull request #36165: [SPARK-36620][SHUFFLE] Add Push Based Shuffle client side metrics
Posted by GitBox <gi...@apache.org>.
thejdeep commented on code in PR #36165:
URL: https://github.com/apache/spark/pull/36165#discussion_r1034008608
##########
project/MimaExcludes.scala:
##########
@@ -87,6 +87,11 @@ object MimaExcludes {
// [SPARK-36511][MINOR][SQL] Remove ColumnIOUtil
ProblemFilters.exclude[MissingClassProblem]("org.apache.parquet.io.ColumnIOUtil"),
+ // [SPARK-36620] [SHUFFLE] Expose push based shuffle metrics
+ ProblemFilters.exclude[DirectMissingMethodProblem]("org.apache.spark.status.api.v1.ShuffleReadMetricDistributions.this"),
+ ProblemFilters.exclude[DirectMissingMethodProblem]("org.apache.spark.status.api.v1.ShuffleReadMetrics.this"),
+ ProblemFilters.exclude[MissingMethodProblem]("org.apache.spark.status.api.v1.StageData.this"),
+
// [SPARK-40324][SQL] Provide query context in AnalysisException
Review Comment:
There are some binary incompatibilities that had to be added to the excludes list.
--
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
[GitHub] [spark] mridulm commented on a diff in pull request #36165: [SPARK-36620][SHUFFLE] Add Push Based Shuffle client side metrics
Posted by GitBox <gi...@apache.org>.
mridulm commented on code in PR #36165:
URL: https://github.com/apache/spark/pull/36165#discussion_r1029927503
##########
core/src/main/scala/org/apache/spark/storage/ShuffleBlockFetcherIterator.scala:
##########
@@ -282,6 +280,17 @@ final class ShuffleBlockFetcherIterator(
}
}
+ @inline def updateMergedReqsDuration(wasReqForMergedChunks: Boolean = false): Unit = {
+ if (remainingBlocks.isEmpty) {
+ val durationMs = TimeUnit.NANOSECONDS.toMillis(clock.nanoTime() - requestStartTime)
+ if (wasReqForMergedChunks) {
+ shuffleMetrics.incRemoteMergedReqsDuration(durationMs)
+ } else {
+ shuffleMetrics.incRemoteReqsDuration(durationMs)
+ }
Review Comment:
`shuffleLocalMetricsUpdate` and `shuffleRemoteMetricsUpdate` is treating merged fetches as a subset of fetches for remote bytes - while here we are treating them separately.
Any particular reason for this difference ? (A throughput computation becomes tricky due to this difference)
--
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
[GitHub] [spark] thejdeep commented on a diff in pull request #36165: [SPARK-36620][SHUFFLE] Add Push Based Shuffle client side metrics
Posted by GitBox <gi...@apache.org>.
thejdeep commented on code in PR #36165:
URL: https://github.com/apache/spark/pull/36165#discussion_r1010567174
##########
core/src/main/scala/org/apache/spark/executor/Executor.scala:
##########
@@ -654,6 +654,27 @@ private[spark] class Executor(
executorSource.METRIC_RESULT_SIZE.inc(task.metrics.resultSize)
executorSource.METRIC_DISK_BYTES_SPILLED.inc(task.metrics.diskBytesSpilled)
executorSource.METRIC_MEMORY_BYTES_SPILLED.inc(task.metrics.memoryBytesSpilled)
+ executorSource.METRIC_PUSH_BASED_SHUFFLE_CORRUPT_MERGED_BLOCK_CHUNKS
+ .inc(task.metrics.shuffleReadMetrics.corruptMergedBlockChunks)
+ executorSource.METRIC_PUSH_BASED_SHUFFLE_FALLBACK_COUNT
+ .inc(task.metrics.shuffleReadMetrics.fallbackCount)
+ executorSource.METRIC_PUSH_BASED_SHUFFLE_MERGED_REMOTE_BLOCKS_FETCHED
+ .inc(task.metrics.shuffleReadMetrics.remoteMergedBlocksFetched)
+ executorSource.METRIC_PUSH_BASED_SHUFFLE_MERGED_LOCAL_BLOCKS_FETCHED
+ .inc(task.metrics.shuffleReadMetrics.localMergedBlocksFetched)
+ executorSource.METRIC_PUSH_BASED_SHUFFLE_MERGED_REMOTE_CHUNKS_FETCHED
+ .inc(task.metrics.shuffleReadMetrics.remoteMergedChunksFetched)
+ executorSource.METRIC_PUSH_BASED_SHUFFLE_MERGED_LOCAL_CHUNKS_FETCHED
+ .inc(task.metrics.shuffleReadMetrics.localMergedChunksFetched)
+ executorSource.METRIC_PUSH_BASED_SHUFFLE_MERGED_REMOTE_BYTES_READ
+ .inc(task.metrics.shuffleReadMetrics.remoteMergedBlocksBytesRead)
+
+ executorSource.METRIC_PUSH_BASED_SHUFFLE_MERGED_LOCAL_BLOCKS_FETCHED
+ .inc(task.metrics.shuffleReadMetrics.localMergedBlocksBytesRead)
+ executorSource.METRIC_PUSH_BASED_SHUFFLE_REMOTE_REQS_DURATION
+ .inc(task.metrics.shuffleReadMetrics.remoteReqsDuration)
+ executorSource.METRIC_PUSH_BASED_SHUFFLE_MERGED_REMOTE_REQS_DURATION
+ .inc(task.metrics.shuffleReadMetrics.remoteMergedReqsDuration)
Review Comment:
Added method to populate these.
--
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
[GitHub] [spark] zhouyejoe commented on pull request #36165: [SPARK-36620][SHUFFLE] Add Push Based Shuffle client side metrics
Posted by GitBox <gi...@apache.org>.
zhouyejoe commented on PR #36165:
URL: https://github.com/apache/spark/pull/36165#issuecomment-1139937725
@thejdeep This PR should have missed the changes required in ShuffleBlockFetchIterator which update the ShuffleMetrics while shuffle fetching is on-going.
--
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
[GitHub] [spark] otterc commented on a diff in pull request #36165: [SPARK-36620][SHUFFLE] Add Push Based Shuffle client side metrics
Posted by GitBox <gi...@apache.org>.
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