Skip to content

[SPARK-38309][CORE] Fix SHS shuffleTotalReads and shuffleTotalBlocks percentile metrics #35637

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
robreeves wants to merge 2 commits into from
Closed

[SPARK-38309][CORE] Fix SHS shuffleTotalReads and shuffleTotalBlocks percentile metrics #35637

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
da39a32
fixed issue and added unit test
robreeves Feb 23, 2022
7715538
Trigger Build
robreeves Feb 24, 2022
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
4 changes: 2 additions & 2 deletions core/src/main/scala/org/apache/spark/status/storeTypes.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -344,7 +344,7 @@ private[spark] class TaskDataWrapper(
@JsonIgnore @KVIndex(value = TaskIndexNames.SHUFFLE_TOTAL_READS, parent = TaskIndexNames.STAGE)
private def shuffleTotalReads: Long = {
if (hasMetrics) {
getMetricValue(shuffleLocalBytesRead) + getMetricValue(shuffleRemoteBytesRead)
shuffleLocalBytesRead + shuffleRemoteBytesRead
} else {
-1L
}
Expand All @@ -353,7 +353,7 @@ private[spark] class TaskDataWrapper(
@JsonIgnore @KVIndex(value = TaskIndexNames.SHUFFLE_TOTAL_BLOCKS, parent = TaskIndexNames.STAGE)
private def shuffleTotalBlocks: Long = {
if (hasMetrics) {
getMetricValue(shuffleLocalBlocksFetched) + getMetricValue(shuffleRemoteBlocksFetched)
shuffleLocalBlocksFetched + shuffleRemoteBlocksFetched
} else {
-1L
}
Expand Down
108 changes: 79 additions & 29 deletions core/src/test/scala/org/apache/spark/status/AppStatusStoreSuite.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -17,6 +17,8 @@

package org.apache.spark.status

import scala.util.Random

import org.apache.spark.{SparkConf, SparkFunSuite}
import org.apache.spark.executor.TaskMetrics
import org.apache.spark.internal.config.History.{HYBRID_STORE_DISK_BACKEND, HybridStoreDiskBackend}
Expand Down Expand Up @@ -137,13 +139,52 @@ class AppStatusStoreSuite extends SparkFunSuite {
* Task summary will consider (1, 3, 5) only
*/
val summary = appStore.taskSummary(stageId, attemptId, uiQuantiles).get
val successfulTasks = Array(getTaskMetrics(1), getTaskMetrics(3), getTaskMetrics(5))

val values = Array(1.0, 3.0, 5.0)
def assertQuantiles(metricGetter: TaskMetrics => Double,
actualQuantiles: Seq[Double]): Unit = {
val values = successfulTasks.map(metricGetter)
val expectedQuantiles = new Distribution(values, 0, values.length)
.getQuantiles(uiQuantiles.sorted)

val dist = new Distribution(values, 0, values.length).getQuantiles(uiQuantiles.sorted)
dist.zip(summary.executorRunTime).foreach { case (expected, actual) =>
assert(expected === actual)
assert(actualQuantiles === expectedQuantiles)
}

assertQuantiles(_.executorDeserializeTime, summary.executorDeserializeTime)
assertQuantiles(_.executorDeserializeCpuTime, summary.executorDeserializeCpuTime)
assertQuantiles(_.executorRunTime, summary.executorRunTime)
assertQuantiles(_.executorRunTime, summary.executorRunTime)
assertQuantiles(_.executorCpuTime, summary.executorCpuTime)
assertQuantiles(_.resultSize, summary.resultSize)
assertQuantiles(_.jvmGCTime, summary.jvmGcTime)
assertQuantiles(_.resultSerializationTime, summary.resultSerializationTime)
assertQuantiles(_.memoryBytesSpilled, summary.memoryBytesSpilled)
assertQuantiles(_.diskBytesSpilled, summary.diskBytesSpilled)
assertQuantiles(_.peakExecutionMemory, summary.peakExecutionMemory)
assertQuantiles(_.inputMetrics.bytesRead, summary.inputMetrics.bytesRead)
assertQuantiles(_.inputMetrics.recordsRead, summary.inputMetrics.recordsRead)
assertQuantiles(_.outputMetrics.bytesWritten, summary.outputMetrics.bytesWritten)
assertQuantiles(_.outputMetrics.recordsWritten, summary.outputMetrics.recordsWritten)
assertQuantiles(_.shuffleReadMetrics.remoteBlocksFetched,
summary.shuffleReadMetrics.remoteBlocksFetched)
assertQuantiles(_.shuffleReadMetrics.localBlocksFetched,
summary.shuffleReadMetrics.localBlocksFetched)
assertQuantiles(_.shuffleReadMetrics.fetchWaitTime, summary.shuffleReadMetrics.fetchWaitTime)
assertQuantiles(_.shuffleReadMetrics.remoteBytesRead,
summary.shuffleReadMetrics.remoteBytesRead)
assertQuantiles(_.shuffleReadMetrics.remoteBytesReadToDisk,
summary.shuffleReadMetrics.remoteBytesReadToDisk)
assertQuantiles(
t => t.shuffleReadMetrics.localBytesRead + t.shuffleReadMetrics.remoteBytesRead,
summary.shuffleReadMetrics.readBytes)
assertQuantiles(
t => t.shuffleReadMetrics.localBlocksFetched + t.shuffleReadMetrics.remoteBlocksFetched,
summary.shuffleReadMetrics.totalBlocksFetched)
assertQuantiles(_.shuffleWriteMetrics.bytesWritten, summary.shuffleWriteMetrics.writeBytes)
assertQuantiles(_.shuffleWriteMetrics.writeTime, summary.shuffleWriteMetrics.writeTime)
assertQuantiles(_.shuffleWriteMetrics.recordsWritten,
summary.shuffleWriteMetrics.writeRecords)

appStore.close()
}
}
Expand Down Expand Up @@ -227,32 +268,41 @@ class AppStatusStoreSuite extends SparkFunSuite {
liveTask.write(store.asInstanceOf[ElementTrackingStore], 1L)
}

private def getTaskMetrics(i: Int): TaskMetrics = {
/**
* Creates fake task metrics
* @param seed The random seed. The output will be reproducible for a given seed.
* @return The test metrics object with fake data
*/
private def getTaskMetrics(seed: Int): TaskMetrics = {
val random = new Random(seed)
val randomMax = 1000
def nextInt(): Int = random.nextInt(randomMax)
Copy link
Contributor

@mridulm mridulm Mar 1, 2022

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Note: This can end up becoming zero.

Copy link
Contributor Author

@robreeves robreeves Mar 1, 2022

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This should have no impact on the validity of the tests here.

Copy link
Contributor Author

@robreeves robreeves Mar 1, 2022

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I think you might have mentioned this due to the metric value being made negative for tasks that aren't successful. Zero is accounted for here.


val taskMetrics = new TaskMetrics()
taskMetrics.setExecutorDeserializeTime(i)
taskMetrics.setExecutorDeserializeCpuTime(i)
taskMetrics.setExecutorRunTime(i)
taskMetrics.setExecutorCpuTime(i)
taskMetrics.setResultSize(i)
taskMetrics.setJvmGCTime(i)
taskMetrics.setResultSerializationTime(i)
taskMetrics.incMemoryBytesSpilled(i)
taskMetrics.incDiskBytesSpilled(i)
taskMetrics.incPeakExecutionMemory(i)
taskMetrics.inputMetrics.incBytesRead(i)
taskMetrics.inputMetrics.incRecordsRead(i)
taskMetrics.outputMetrics.setBytesWritten(i)
taskMetrics.outputMetrics.setRecordsWritten(i)
taskMetrics.shuffleReadMetrics.incRemoteBlocksFetched(i)
taskMetrics.shuffleReadMetrics.incLocalBlocksFetched(i)
taskMetrics.shuffleReadMetrics.incFetchWaitTime(i)
taskMetrics.shuffleReadMetrics.incRemoteBytesRead(i)
taskMetrics.shuffleReadMetrics.incRemoteBytesReadToDisk(i)
taskMetrics.shuffleReadMetrics.incLocalBytesRead(i)
taskMetrics.shuffleReadMetrics.incRecordsRead(i)
taskMetrics.shuffleWriteMetrics.incBytesWritten(i)
taskMetrics.shuffleWriteMetrics.incWriteTime(i)
taskMetrics.shuffleWriteMetrics.incRecordsWritten(i)
taskMetrics.setExecutorDeserializeTime(nextInt())
taskMetrics.setExecutorDeserializeCpuTime(nextInt())
taskMetrics.setExecutorRunTime(nextInt())
taskMetrics.setExecutorCpuTime(nextInt())
taskMetrics.setResultSize(nextInt())
taskMetrics.setJvmGCTime(nextInt())
taskMetrics.setResultSerializationTime(nextInt())
taskMetrics.incMemoryBytesSpilled(nextInt())
taskMetrics.incDiskBytesSpilled(nextInt())
taskMetrics.incPeakExecutionMemory(nextInt())
taskMetrics.inputMetrics.incBytesRead(nextInt())
taskMetrics.inputMetrics.incRecordsRead(nextInt())
taskMetrics.outputMetrics.setBytesWritten(nextInt())
taskMetrics.outputMetrics.setRecordsWritten(nextInt())
taskMetrics.shuffleReadMetrics.incRemoteBlocksFetched(nextInt())
taskMetrics.shuffleReadMetrics.incLocalBlocksFetched(nextInt())
taskMetrics.shuffleReadMetrics.incFetchWaitTime(nextInt())
taskMetrics.shuffleReadMetrics.incRemoteBytesRead(nextInt())
taskMetrics.shuffleReadMetrics.incRemoteBytesReadToDisk(nextInt())
taskMetrics.shuffleReadMetrics.incLocalBytesRead(nextInt())
taskMetrics.shuffleReadMetrics.incRecordsRead(nextInt())
taskMetrics.shuffleWriteMetrics.incBytesWritten(nextInt())
taskMetrics.shuffleWriteMetrics.incWriteTime(nextInt())
taskMetrics.shuffleWriteMetrics.incRecordsWritten(nextInt())
taskMetrics
}

Expand Down