Add support for kudo write metrics #11784
Open
+121
−63
Conversation
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Signed-off-by: liurenjie1024 <[email protected]>
jlowe
reviewed
Nov 27, 2024
| val METRIC_DATA_READ_SIZE = "dataReadSize" | ||
| val METRIC_DESC_DATA_READ_SIZE = "data read size" | ||
| val METRIC_SHUFFLE_SERIALIZATION_TIME = "rapidsShuffleSerializationTime" | ||
| val METRIC_DESC_SHUFFLE_SERIALIZATION_TIME = "rs. serialization time" |
There was a problem hiding this comment.
Not new with this PR, but "rs." is not a very helpful description and is more cryptic than the metric name itself.
| @@ -315,8 +356,11 @@ object GpuShuffleExchangeExecBase { | |||
| rdd | |||
| } | |||
| val partitioner: GpuExpression = getPartitioner(newRdd, outputAttributes, newPartitioning) | |||
| val partitonTime: GpuMetric = metrics(METRIC_SHUFFLE_PARTITION_TIME) | |||
There was a problem hiding this comment.
Suggested change
| val partitonTime: GpuMetric = metrics(METRIC_SHUFFLE_PARTITION_TIME) | |
| val partitionTime: GpuMetric = metrics(METRIC_SHUFFLE_PARTITION_TIME) |
| @@ -37,6 +38,7 @@ import org.apache.spark.sql.execution.exchange.{Exchange, ShuffleExchangeExec} | |||
| import org.apache.spark.sql.execution.metric._ | |||
| import org.apache.spark.sql.internal.SQLConf | |||
| import org.apache.spark.sql.rapids.GpuShuffleDependency | |||
| import org.apache.spark.sql.rapids.execution.GpuShuffleExchangeExecBase.{createAdditionalExchangeMetris, METRIC_DATA_READ_SIZE, METRIC_DATA_SIZE, METRIC_DESC_DATA_READ_SIZE, METRIC_DESC_DATA_SIZE, METRIC_DESC_SHUFFLE_COMBINE_TIME, METRIC_DESC_SHUFFLE_DESERIALIZATION_TIME, METRIC_DESC_SHUFFLE_PARTITION_TIME, METRIC_DESC_SHUFFLE_READ_TIME, METRIC_DESC_SHUFFLE_SER_CALC_HEADER_TIME, METRIC_DESC_SHUFFLE_SER_COPY_BUFFER_TIME, METRIC_DESC_SHUFFLE_SER_COPY_HEADER_TIME, METRIC_DESC_SHUFFLE_SERIALIZATION_TIME, METRIC_DESC_SHUFFLE_WRITE_IO_TIME, METRIC_DESC_SHUFFLE_WRITE_TIME, METRIC_SHUFFLE_COMBINE_TIME, METRIC_SHUFFLE_DESERIALIZATION_TIME, METRIC_SHUFFLE_PARTITION_TIME, METRIC_SHUFFLE_READ_TIME, METRIC_SHUFFLE_SER_CALC_HEADER_TIME, METRIC_SHUFFLE_SER_COPY_BUFFER_TIME, METRIC_SHUFFLE_SER_COPY_HEADER_TIME, METRIC_SHUFFLE_SERIALIZATION_TIME, METRIC_SHUFFLE_WRITE_IO_TIME, METRIC_SHUFFLE_WRITE_TIME} | |||
There was a problem hiding this comment.
Suggested change
| import org.apache.spark.sql.rapids.execution.GpuShuffleExchangeExecBase.{createAdditionalExchangeMetris, METRIC_DATA_READ_SIZE, METRIC_DATA_SIZE, METRIC_DESC_DATA_READ_SIZE, METRIC_DESC_DATA_SIZE, METRIC_DESC_SHUFFLE_COMBINE_TIME, METRIC_DESC_SHUFFLE_DESERIALIZATION_TIME, METRIC_DESC_SHUFFLE_PARTITION_TIME, METRIC_DESC_SHUFFLE_READ_TIME, METRIC_DESC_SHUFFLE_SER_CALC_HEADER_TIME, METRIC_DESC_SHUFFLE_SER_COPY_BUFFER_TIME, METRIC_DESC_SHUFFLE_SER_COPY_HEADER_TIME, METRIC_DESC_SHUFFLE_SERIALIZATION_TIME, METRIC_DESC_SHUFFLE_WRITE_IO_TIME, METRIC_DESC_SHUFFLE_WRITE_TIME, METRIC_SHUFFLE_COMBINE_TIME, METRIC_SHUFFLE_DESERIALIZATION_TIME, METRIC_SHUFFLE_PARTITION_TIME, METRIC_SHUFFLE_READ_TIME, METRIC_SHUFFLE_SER_CALC_HEADER_TIME, METRIC_SHUFFLE_SER_COPY_BUFFER_TIME, METRIC_SHUFFLE_SER_COPY_HEADER_TIME, METRIC_SHUFFLE_SERIALIZATION_TIME, METRIC_SHUFFLE_WRITE_IO_TIME, METRIC_SHUFFLE_WRITE_TIME} | |
| import org.apache.spark.sql.rapids.execution.GpuShuffleExchangeExecBase.createAdditionalExchangeMetrics |
| val METRIC_SHUFFLE_SER_COPY_BUFFER_TIME = "rapidsShuffleSerializationCopyBufferTime" | ||
| val METRIC_DESC_SHUFFLE_SER_COPY_BUFFER_TIME = "rs. serialization copy buffer time" | ||
|
|
||
| def createAdditionalExchangeMetris(gpu: GpuExec): Map[String, GpuMetric] = Map( |
There was a problem hiding this comment.
Suggested change
| def createAdditionalExchangeMetris(gpu: GpuExec): Map[String, GpuMetric] = Map( | |
| def createAdditionalExchangeMetrics(gpu: GpuExec): Map[String, GpuMetric] = Map( |
| @@ -679,7 +675,6 @@ abstract class RapidsShuffleThreadedReaderBase[K, C]( | |||
| } | |||
| val res = currentIter.next() | |||
| val fetchTime = System.nanoTime() - fetchTimeStart | |||
| deserializationTimeNs.foreach(_ += (fetchTime - readBlockedTime)) | |||
There was a problem hiding this comment.
I'm not sure the new deserialization time metric does the same thing the old metric did. The new metric is timing how long it takes the serializer to produce a new batch, but that's measuring I/O time as well as deserialization time once the I/O is completed. The original metric appears to go out of its way to remove the time spent on I/O to get a more accurate deserialization metric.
There was a problem hiding this comment.
The case is different when using rss such as celeborn, e.g. it's pure serialization time which simply write to in memory buffer. I will restore the change here and add a new metric for this.
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
This pr address two things: