This is an automated email from the ASF dual-hosted git repository. srowen pushed a commit to branch master in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/master by this push: new 8ed5808 [SPARK-34488][CORE] Support task Metrics Distributions and executor Metrics Distributions in the REST API call for a specified stage 8ed5808 is described below commit 8ed5808f64e83a9f085d456c6ab9188c49992eae Author: Angerszhuuuu <angers....@gmail.com> AuthorDate: Wed Mar 24 08:50:45 2021 -0500 [SPARK-34488][CORE] Support task Metrics Distributions and executor Metrics Distributions in the REST API call for a specified stage ### What changes were proposed in this pull request? For a specific stage, it is useful to show the task metrics in percentile distribution. This information can help users know whether or not there is a skew/bottleneck among tasks in a given stage. We list an example in taskMetricsDistributions.json Similarly, it is useful to show the executor metrics in percentile distribution for a specific stage. This information can show whether or not there is a skewed load on some executors. We list an example in executorMetricsDistributions.json We define `withSummaries` and `quantiles` query parameter in the REST API for a specific stage as: applications/<application_id>/<application_attempt/stages/<stage_id>/<stage_attempt>?withSummaries=[true|false]& quantiles=0.05,0.25,0.5,0.75,0.95 1. withSummaries: default is false, define whether to show current stage's taskMetricsDistribution and executorMetricsDistribution 2. quantiles: default is `0.0,0.25,0.5,0.75,1.0` only effect when `withSummaries=true`, it define the quantiles we use when calculating metrics distributions. When withSummaries=true, both task metrics in percentile distribution and executor metrics in percentile distribution are included in the REST API output. The default value of withSummaries is false, i.e. no metrics percentile distribution will be included in the REST API output. ### Why are the changes needed? For a specific stage, it is useful to show the task metrics in percentile distribution. This information can help users know whether or not there is a skew/bottleneck among tasks in a given stage. We list an example in taskMetricsDistributions.json ### Does this PR introduce _any_ user-facing change? User can use below restful API to get task metrics distribution and executor metrics distribution for indivial stage ``` applications/<application_id>/<application_attempt/stages/<stage_id>/<stage_attempt>?withSummaries=[true|false] ``` ### How was this patch tested? Added UT Closes #31611 from AngersZhuuuu/SPARK-34488. Authored-by: Angerszhuuuu <angers....@gmail.com> Signed-off-by: Sean Owen <sro...@gmail.com> --- .../org/apache/spark/status/AppStatusStore.scala | 206 ++-- .../scala/org/apache/spark/status/LiveEntity.scala | 4 +- .../spark/status/api/v1/StagesResource.scala | 38 +- .../scala/org/apache/spark/status/api/v1/api.scala | 52 +- .../scala/org/apache/spark/ui/jobs/JobPage.scala | 4 +- .../stage_with_summaries_expectation.json | 1077 ++++++++++++++++++++ .../spark/deploy/history/HistoryServerSuite.scala | 1 + .../scala/org/apache/spark/ui/StagePageSuite.scala | 4 +- docs/monitoring.md | 18 +- 9 files changed, 1326 insertions(+), 78 deletions(-) diff --git a/core/src/main/scala/org/apache/spark/status/AppStatusStore.scala b/core/src/main/scala/org/apache/spark/status/AppStatusStore.scala index b9cc914..8d43bef 100644 --- a/core/src/main/scala/org/apache/spark/status/AppStatusStore.scala +++ b/core/src/main/scala/org/apache/spark/status/AppStatusStore.scala @@ -113,10 +113,15 @@ private[spark] class AppStatusStore( } } - def stageData(stageId: Int, details: Boolean = false): Seq[v1.StageData] = { + def stageData( + stageId: Int, + details: Boolean = false, + withSummaries: Boolean = false, + unsortedQuantiles: Array[Double] = Array.empty[Double]): Seq[v1.StageData] = { store.view(classOf[StageDataWrapper]).index("stageId").first(stageId).last(stageId) .asScala.map { s => - if (details) stageWithDetails(s.info) else s.info + newStageData(s.info, withDetail = details, withSummaries = withSummaries, + unsortedQuantiles = unsortedQuantiles) }.toSeq } @@ -138,11 +143,15 @@ private[spark] class AppStatusStore( } } - def stageAttempt(stageId: Int, stageAttemptId: Int, - details: Boolean = false): (v1.StageData, Seq[Int]) = { + def stageAttempt( + stageId: Int, stageAttemptId: Int, + details: Boolean = false, + withSummaries: Boolean = false, + unsortedQuantiles: Array[Double] = Array.empty[Double]): (v1.StageData, Seq[Int]) = { val stageKey = Array(stageId, stageAttemptId) val stageDataWrapper = store.read(classOf[StageDataWrapper], stageKey) - val stage = if (details) stageWithDetails(stageDataWrapper.info) else stageDataWrapper.info + val stage = newStageData(stageDataWrapper.info, withDetail = details, + withSummaries = withSummaries, unsortedQuantiles = unsortedQuantiles) (stage, stageDataWrapper.jobIds.toSeq) } @@ -453,61 +462,138 @@ private[spark] class AppStatusStore( } } - private def stageWithDetails(stage: v1.StageData): v1.StageData = { - val tasks = taskList(stage.stageId, stage.attemptId, Int.MaxValue) - .map { t => (t.taskId, t) } - .toMap - - new v1.StageData( - status = stage.status, - stageId = stage.stageId, - attemptId = stage.attemptId, - numTasks = stage.numTasks, - numActiveTasks = stage.numActiveTasks, - numCompleteTasks = stage.numCompleteTasks, - numFailedTasks = stage.numFailedTasks, - numKilledTasks = stage.numKilledTasks, - numCompletedIndices = stage.numCompletedIndices, - submissionTime = stage.submissionTime, - firstTaskLaunchedTime = stage.firstTaskLaunchedTime, - completionTime = stage.completionTime, - failureReason = stage.failureReason, - executorDeserializeTime = stage.executorDeserializeTime, - executorDeserializeCpuTime = stage.executorDeserializeCpuTime, - executorRunTime = stage.executorRunTime, - executorCpuTime = stage.executorCpuTime, - resultSize = stage.resultSize, - jvmGcTime = stage.jvmGcTime, - resultSerializationTime = stage.resultSerializationTime, - memoryBytesSpilled = stage.memoryBytesSpilled, - diskBytesSpilled = stage.diskBytesSpilled, - peakExecutionMemory = stage.peakExecutionMemory, - inputBytes = stage.inputBytes, - inputRecords = stage.inputRecords, - outputBytes = stage.outputBytes, - outputRecords = stage.outputRecords, - shuffleRemoteBlocksFetched = stage.shuffleRemoteBlocksFetched, - shuffleLocalBlocksFetched = stage.shuffleLocalBlocksFetched, - shuffleFetchWaitTime = stage.shuffleFetchWaitTime, - shuffleRemoteBytesRead = stage.shuffleRemoteBytesRead, - shuffleRemoteBytesReadToDisk = stage.shuffleRemoteBytesReadToDisk, - shuffleLocalBytesRead = stage.shuffleLocalBytesRead, - shuffleReadBytes = stage.shuffleReadBytes, - shuffleReadRecords = stage.shuffleReadRecords, - shuffleWriteBytes = stage.shuffleWriteBytes, - shuffleWriteTime = stage.shuffleWriteTime, - shuffleWriteRecords = stage.shuffleWriteRecords, - name = stage.name, - description = stage.description, - details = stage.details, - schedulingPool = stage.schedulingPool, - rddIds = stage.rddIds, - accumulatorUpdates = stage.accumulatorUpdates, - tasks = Some(tasks), - executorSummary = Some(executorSummary(stage.stageId, stage.attemptId)), - killedTasksSummary = stage.killedTasksSummary, - resourceProfileId = stage.resourceProfileId, - peakExecutorMetrics = stage.peakExecutorMetrics) + def newStageData( + stage: v1.StageData, + withDetail: Boolean = false, + withSummaries: Boolean = false, + unsortedQuantiles: Array[Double] = Array.empty[Double]): v1.StageData = { + if (!withDetail && !withSummaries) { + stage + } else { + val quantiles = unsortedQuantiles.sorted + val tasks: Option[Map[Long, v1.TaskData]] = if (withDetail) { + val tasks = taskList(stage.stageId, stage.attemptId, Int.MaxValue) + .map { t => (t.taskId, t) } + .toMap + Some(tasks) + } else { + None + } + val executorSummaries: Option[Map[String, v1.ExecutorStageSummary]] = if (withDetail) { + Some(executorSummary(stage.stageId, stage.attemptId)) + } else { + None + } + val taskMetricsDistribution: Option[v1.TaskMetricDistributions] = if (withSummaries) { + taskSummary(stage.stageId, stage.attemptId, quantiles) + } else { + None + } + val executorMetricsDistributions: Option[v1.ExecutorMetricsDistributions] = + if (withSummaries) { + stageExecutorSummary(stage.stageId, stage.attemptId, quantiles) + } else { + None + } + + new v1.StageData( + status = stage.status, + stageId = stage.stageId, + attemptId = stage.attemptId, + numTasks = stage.numTasks, + numActiveTasks = stage.numActiveTasks, + numCompleteTasks = stage.numCompleteTasks, + numFailedTasks = stage.numFailedTasks, + numKilledTasks = stage.numKilledTasks, + numCompletedIndices = stage.numCompletedIndices, + submissionTime = stage.submissionTime, + firstTaskLaunchedTime = stage.firstTaskLaunchedTime, + completionTime = stage.completionTime, + failureReason = stage.failureReason, + executorDeserializeTime = stage.executorDeserializeTime, + executorDeserializeCpuTime = stage.executorDeserializeCpuTime, + executorRunTime = stage.executorRunTime, + executorCpuTime = stage.executorCpuTime, + resultSize = stage.resultSize, + jvmGcTime = stage.jvmGcTime, + resultSerializationTime = stage.resultSerializationTime, + memoryBytesSpilled = stage.memoryBytesSpilled, + diskBytesSpilled = stage.diskBytesSpilled, + peakExecutionMemory = stage.peakExecutionMemory, + inputBytes = stage.inputBytes, + inputRecords = stage.inputRecords, + outputBytes = stage.outputBytes, + outputRecords = stage.outputRecords, + shuffleRemoteBlocksFetched = stage.shuffleRemoteBlocksFetched, + shuffleLocalBlocksFetched = stage.shuffleLocalBlocksFetched, + shuffleFetchWaitTime = stage.shuffleFetchWaitTime, + shuffleRemoteBytesRead = stage.shuffleRemoteBytesRead, + shuffleRemoteBytesReadToDisk = stage.shuffleRemoteBytesReadToDisk, + shuffleLocalBytesRead = stage.shuffleLocalBytesRead, + shuffleReadBytes = stage.shuffleReadBytes, + shuffleReadRecords = stage.shuffleReadRecords, + shuffleWriteBytes = stage.shuffleWriteBytes, + shuffleWriteTime = stage.shuffleWriteTime, + shuffleWriteRecords = stage.shuffleWriteRecords, + name = stage.name, + description = stage.description, + details = stage.details, + schedulingPool = stage.schedulingPool, + rddIds = stage.rddIds, + accumulatorUpdates = stage.accumulatorUpdates, + tasks = tasks, + executorSummary = executorSummaries, + killedTasksSummary = stage.killedTasksSummary, + resourceProfileId = stage.resourceProfileId, + peakExecutorMetrics = stage.peakExecutorMetrics, + taskMetricsDistributions = taskMetricsDistribution, + executorMetricsDistributions = executorMetricsDistributions) + } + } + + def stageExecutorSummary( + stageId: Int, + stageAttemptId: Int, + unsortedQuantiles: Array[Double]): Option[v1.ExecutorMetricsDistributions] = { + val quantiles = unsortedQuantiles.sorted + val summary = executorSummary(stageId, stageAttemptId) + if (summary.isEmpty) { + None + } else { + val values = summary.values.toIndexedSeq + Some(new v1.ExecutorMetricsDistributions( + quantiles = quantiles, + taskTime = getQuantilesValue(values.map(_.taskTime.toDouble).sorted, quantiles), + failedTasks = getQuantilesValue(values.map(_.failedTasks.toDouble).sorted, quantiles), + succeededTasks = getQuantilesValue(values.map(_.succeededTasks.toDouble).sorted, quantiles), + killedTasks = getQuantilesValue(values.map(_.killedTasks.toDouble).sorted, quantiles), + inputBytes = getQuantilesValue(values.map(_.inputBytes.toDouble).sorted, quantiles), + inputRecords = getQuantilesValue(values.map(_.inputRecords.toDouble).sorted, quantiles), + outputBytes = getQuantilesValue(values.map(_.outputBytes.toDouble).sorted, quantiles), + outputRecords = getQuantilesValue(values.map(_.outputRecords.toDouble).sorted, quantiles), + shuffleRead = getQuantilesValue(values.map(_.shuffleRead.toDouble).sorted, quantiles), + shuffleReadRecords = + getQuantilesValue(values.map(_.shuffleReadRecords.toDouble).sorted, quantiles), + shuffleWrite = getQuantilesValue(values.map(_.shuffleWrite.toDouble).sorted, quantiles), + shuffleWriteRecords = + getQuantilesValue(values.map(_.shuffleWriteRecords.toDouble).sorted, quantiles), + memoryBytesSpilled = + getQuantilesValue(values.map(_.memoryBytesSpilled.toDouble).sorted, quantiles), + diskBytesSpilled = + getQuantilesValue(values.map(_.diskBytesSpilled.toDouble).sorted, quantiles), + peakMemoryMetrics = + new v1.ExecutorPeakMetricsDistributions(quantiles, + values.flatMap(_.peakMemoryMetrics)) + )) + } + } + + def getQuantilesValue( + values: IndexedSeq[Double], + quantiles: Array[Double]): IndexedSeq[Double] = { + val count = values.size + val indices = quantiles.map { q => math.min((q * count).toLong, count - 1) } + indices.map(i => values(i.toInt)).toIndexedSeq } def rdd(rddId: Int): v1.RDDStorageInfo = { diff --git a/core/src/main/scala/org/apache/spark/status/LiveEntity.scala b/core/src/main/scala/org/apache/spark/status/LiveEntity.scala index 3105e7b..b6dfe30 100644 --- a/core/src/main/scala/org/apache/spark/status/LiveEntity.scala +++ b/core/src/main/scala/org/apache/spark/status/LiveEntity.scala @@ -493,7 +493,9 @@ private class LiveStage extends LiveEntity { executorSummary = None, killedTasksSummary = killedSummary, resourceProfileId = info.resourceProfileId, - Some(peakExecutorMetrics).filter(_.isSet)) + peakExecutorMetrics = Some(peakExecutorMetrics).filter(_.isSet), + taskMetricsDistributions = None, + executorMetricsDistributions = None) } override protected def doUpdate(): Any = { diff --git a/core/src/main/scala/org/apache/spark/status/api/v1/StagesResource.scala b/core/src/main/scala/org/apache/spark/status/api/v1/StagesResource.scala index 84bd430..be7a4a6 100644 --- a/core/src/main/scala/org/apache/spark/status/api/v1/StagesResource.scala +++ b/core/src/main/scala/org/apache/spark/status/api/v1/StagesResource.scala @@ -36,9 +36,14 @@ private[v1] class StagesResource extends BaseAppResource { @Path("{stageId: \\d+}") def stageData( @PathParam("stageId") stageId: Int, - @QueryParam("details") @DefaultValue("true") details: Boolean): Seq[StageData] = { + @QueryParam("details") @DefaultValue("true") details: Boolean, + @QueryParam("withSummaries") @DefaultValue("false") withSummaries: Boolean, + @QueryParam("quantiles") @DefaultValue("0.0,0.25,0.5,0.75,1.0") quantileString: String): + Seq[StageData] = { withUI { ui => - val ret = ui.store.stageData(stageId, details = details) + val quantiles = parseQuantileString(quantileString) + val ret = ui.store.stageData(stageId, details = details, + withSummaries = withSummaries, unsortedQuantiles = quantiles) if (ret.nonEmpty) { ret } else { @@ -52,9 +57,14 @@ private[v1] class StagesResource extends BaseAppResource { def oneAttemptData( @PathParam("stageId") stageId: Int, @PathParam("stageAttemptId") stageAttemptId: Int, - @QueryParam("details") @DefaultValue("true") details: Boolean): StageData = withUI { ui => + @QueryParam("details") @DefaultValue("true") details: Boolean, + @QueryParam("withSummaries") @DefaultValue("false") withSummaries: Boolean, + @QueryParam("quantiles") @DefaultValue("0.0,0.25,0.5,0.75,1.0") quantileString: String): + StageData = withUI { ui => try { - ui.store.stageAttempt(stageId, stageAttemptId, details = details)._1 + val quantiles = parseQuantileString(quantileString) + ui.store.stageAttempt(stageId, stageAttemptId, details = details, + withSummaries = withSummaries, unsortedQuantiles = quantiles)._1 } catch { case _: NoSuchElementException => // Change the message depending on whether there are any attempts for the requested stage. @@ -76,15 +86,7 @@ private[v1] class StagesResource extends BaseAppResource { @PathParam("stageAttemptId") stageAttemptId: Int, @DefaultValue("0.05,0.25,0.5,0.75,0.95") @QueryParam("quantiles") quantileString: String) : TaskMetricDistributions = withUI { ui => - val quantiles = quantileString.split(",").map { s => - try { - s.toDouble - } catch { - case nfe: NumberFormatException => - throw new BadParameterException("quantiles", "double", s) - } - } - + val quantiles = parseQuantileString(quantileString) ui.store.taskSummary(stageId, stageAttemptId, quantiles).getOrElse( throw new NotFoundException(s"No tasks reported metrics for $stageId / $stageAttemptId yet.")) } @@ -226,4 +228,14 @@ private[v1] class StagesResource extends BaseAppResource { filteredTaskDataSequence } + def parseQuantileString(quantileString: String): Array[Double] = { + quantileString.split(",").map { s => + try { + s.toDouble + } catch { + case nfe: NumberFormatException => + throw new BadParameterException("quantiles", "double", s) + } + } + } } diff --git a/core/src/main/scala/org/apache/spark/status/api/v1/api.scala b/core/src/main/scala/org/apache/spark/status/api/v1/api.scala index 96f5b7b..52fcb29 100644 --- a/core/src/main/scala/org/apache/spark/status/api/v1/api.scala +++ b/core/src/main/scala/org/apache/spark/status/api/v1/api.scala @@ -170,6 +170,19 @@ private[spark] class ExecutorMetricsJsonSerializer value.isEmpty } +private[spark] class ExecutorPeakMetricsDistributionsJsonSerializer + extends JsonSerializer[ExecutorPeakMetricsDistributions] { + override def serialize( + metrics: ExecutorPeakMetricsDistributions, + jsonGenerator: JsonGenerator, + serializerProvider: SerializerProvider): Unit = { + val metricsMap = ExecutorMetricType.metricToOffset.map { case (metric, _) => + metric -> metrics.getMetricDistribution(metric) + } + jsonGenerator.writeObject(metricsMap) + } +} + class JobData private[spark]( val jobId: Int, val name: String, @@ -279,7 +292,9 @@ class StageData private[spark]( val resourceProfileId: Int, @JsonSerialize(using = classOf[ExecutorMetricsJsonSerializer]) @JsonDeserialize(using = classOf[ExecutorMetricsJsonDeserializer]) - val peakExecutorMetrics: Option[ExecutorMetrics]) + val peakExecutorMetrics: Option[ExecutorMetrics], + val taskMetricsDistributions: Option[TaskMetricDistributions], + val executorMetricsDistributions: Option[ExecutorMetricsDistributions]) class TaskData private[spark]( val taskId: Long, @@ -368,6 +383,41 @@ class OutputMetricDistributions private[spark]( val bytesWritten: IndexedSeq[Double], val recordsWritten: IndexedSeq[Double]) +class ExecutorMetricsDistributions private[spark]( + val quantiles: IndexedSeq[Double], + + val taskTime: IndexedSeq[Double], + val failedTasks: IndexedSeq[Double], + val succeededTasks: IndexedSeq[Double], + val killedTasks: IndexedSeq[Double], + val inputBytes: IndexedSeq[Double], + val inputRecords: IndexedSeq[Double], + val outputBytes: IndexedSeq[Double], + val outputRecords: IndexedSeq[Double], + val shuffleRead: IndexedSeq[Double], + val shuffleReadRecords: IndexedSeq[Double], + val shuffleWrite: IndexedSeq[Double], + val shuffleWriteRecords: IndexedSeq[Double], + val memoryBytesSpilled: IndexedSeq[Double], + val diskBytesSpilled: IndexedSeq[Double], + @JsonSerialize(using = classOf[ExecutorPeakMetricsDistributionsJsonSerializer]) + val peakMemoryMetrics: ExecutorPeakMetricsDistributions +) + +@JsonSerialize(using = classOf[ExecutorPeakMetricsDistributionsJsonSerializer]) +class ExecutorPeakMetricsDistributions private[spark]( + val quantiles: IndexedSeq[Double], + val executorMetrics: IndexedSeq[ExecutorMetrics]) { + private lazy val count = executorMetrics.length + private lazy val indices = quantiles.map { q => math.min((q * count).toLong, count - 1) } + + /** Returns the distributions for the specified metric. */ + def getMetricDistribution(metricName: String): IndexedSeq[Double] = { + val sorted = executorMetrics.map(_.getMetricValue(metricName)).sorted + indices.map(i => sorted(i.toInt).toDouble).toIndexedSeq + } +} + class ShuffleReadMetricDistributions private[spark]( val readBytes: IndexedSeq[Double], val readRecords: IndexedSeq[Double], diff --git a/core/src/main/scala/org/apache/spark/ui/jobs/JobPage.scala b/core/src/main/scala/org/apache/spark/ui/jobs/JobPage.scala index 1dfbce8..a19daec 100644 --- a/core/src/main/scala/org/apache/spark/ui/jobs/JobPage.scala +++ b/core/src/main/scala/org/apache/spark/ui/jobs/JobPage.scala @@ -257,7 +257,9 @@ private[ui] class JobPage(parent: JobsTab, store: AppStatusStore) extends WebUIP executorSummary = None, killedTasksSummary = Map(), ResourceProfile.UNKNOWN_RESOURCE_PROFILE_ID, - peakExecutorMetrics = None) + peakExecutorMetrics = None, + taskMetricsDistributions = None, + executorMetricsDistributions = None) } } diff --git a/core/src/test/resources/HistoryServerExpectations/stage_with_summaries_expectation.json b/core/src/test/resources/HistoryServerExpectations/stage_with_summaries_expectation.json new file mode 100644 index 0000000..179e46f --- /dev/null +++ b/core/src/test/resources/HistoryServerExpectations/stage_with_summaries_expectation.json @@ -0,0 +1,1077 @@ +{ + "status" : "COMPLETE", + "stageId" : 2, + "attemptId" : 0, + "numTasks" : 16, + "numActiveTasks" : 0, + "numCompleteTasks" : 16, + "numFailedTasks" : 0, + "numKilledTasks" : 0, + "numCompletedIndices" : 16, + "submissionTime" : "2020-07-07T03:11:21.040GMT", + "firstTaskLaunchedTime" : "2020-07-07T03:11:21.077GMT", + "completionTime" : "2020-07-07T03:11:23.044GMT", + "executorDeserializeTime" : 3905, + "executorDeserializeCpuTime" : 979900000, + "executorRunTime" : 25579, + "executorCpuTime" : 8810338000, + "resultSize" : 33883, + "jvmGcTime" : 1010, + "resultSerializationTime" : 11, + "memoryBytesSpilled" : 0, + "diskBytesSpilled" : 0, + "peakExecutionMemory" : 384640, + "inputBytes" : 0, + "inputRecords" : 0, + "outputBytes" : 0, + "outputRecords" : 0, + "shuffleRemoteBlocksFetched" : 0, + "shuffleLocalBlocksFetched" : 0, + "shuffleFetchWaitTime" : 0, + "shuffleRemoteBytesRead" : 0, + "shuffleRemoteBytesReadToDisk" : 0, + "shuffleLocalBytesRead" : 0, + "shuffleReadBytes" : 0, + "shuffleReadRecords" : 0, + "shuffleWriteBytes" : 0, + "shuffleWriteTime" : 0, + "shuffleWriteRecords" : 0, + "name" : "foreach at <console>:26", + "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$line19.$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>(<c [...] + "schedulingPool" : "default", + "rddIds" : [ 10, 8, 6, 7, 9 ], + "accumulatorUpdates" : [ ], + "tasks" : { + "42" : { + "taskId" : 42, + "index" : 10, + "attempt" : 0, + "launchTime" : "2020-07-07T03:11:21.120GMT", + "duration" : 1923, + "executorId" : "0", + "host" : "127.0.0.1", + "status" : "SUCCESS", + "taskLocality" : "PROCESS_LOCAL", + "speculative" : false, + "accumulatorUpdates" : [ ], + "taskMetrics" : { + "executorDeserializeTime" : 229, + "executorDeserializeCpuTime" : 58152000, + "executorRunTime" : 1624, + "executorCpuTime" : 508230000, + "resultSize" : 2115, + "jvmGcTime" : 66, + "resultSerializationTime" : 0, + "memoryBytesSpilled" : 0, + "diskBytesSpilled" : 0, + "peakExecutionMemory" : 24040, + "inputMetrics" : { + "bytesRead" : 0, + "recordsRead" : 0 + }, + "outputMetrics" : { + "bytesWritten" : 0, + "recordsWritten" : 0 + }, + "shuffleReadMetrics" : { + "remoteBlocksFetched" : 0, + "localBlocksFetched" : 0, + "fetchWaitTime" : 0, + "remoteBytesRead" : 0, + "remoteBytesReadToDisk" : 0, + "localBytesRead" : 0, + "recordsRead" : 0 + }, + "shuffleWriteMetrics" : { + "bytesWritten" : 0, + "writeTime" : 0, + "recordsWritten" : 0 + } + }, + "executorLogs" : { + "stdout" : "http://127.0.0.1:8081/logPage/?appId=app-20200706201101-0003&executorId=0&logType=stdout", + "stderr" : "http://127.0.0.1:8081/logPage/?appId=app-20200706201101-0003&executorId=0&logType=stderr" + }, + "schedulerDelay" : 70, + "gettingResultTime" : 0 + }, + "37" : { + "taskId" : 37, + "index" : 5, + "attempt" : 0, + "launchTime" : "2020-07-07T03:11:21.100GMT", + "duration" : 1915, + "executorId" : "0", + "host" : "127.0.0.1", + "status" : "SUCCESS", + "taskLocality" : "PROCESS_LOCAL", + "speculative" : false, + "accumulatorUpdates" : [ ], + "taskMetrics" : { + "executorDeserializeTime" : 256, + "executorDeserializeCpuTime" : 60890000, + "executorRunTime" : 1596, + "executorCpuTime" : 507192000, + "resultSize" : 2115, + "jvmGcTime" : 62, + "resultSerializationTime" : 0, + "memoryBytesSpilled" : 0, + "diskBytesSpilled" : 0, + "peakExecutionMemory" : 24040, + "inputMetrics" : { + "bytesRead" : 0, + "recordsRead" : 0 + }, + "outputMetrics" : { + "bytesWritten" : 0, + "recordsWritten" : 0 + }, + "shuffleReadMetrics" : { + "remoteBlocksFetched" : 0, + "localBlocksFetched" : 0, + "fetchWaitTime" : 0, + "remoteBytesRead" : 0, + "remoteBytesReadToDisk" : 0, + "localBytesRead" : 0, + "recordsRead" : 0 + }, + "shuffleWriteMetrics" : { + "bytesWritten" : 0, + "writeTime" : 0, + "recordsWritten" : 0 + } + }, + "executorLogs" : { + "stdout" : "http://127.0.0.1:8081/logPage/?appId=app-20200706201101-0003&executorId=0&logType=stdout", + "stderr" : "http://127.0.0.1:8081/logPage/?appId=app-20200706201101-0003&executorId=0&logType=stderr" + }, + "schedulerDelay" : 63, + "gettingResultTime" : 0 + }, + "46" : { + "taskId" : 46, + "index" : 14, + "attempt" : 0, + "launchTime" : "2020-07-07T03:11:21.132GMT", + "duration" : 1905, + "executorId" : "0", + "host" : "127.0.0.1", + "status" : "SUCCESS", + "taskLocality" : "PROCESS_LOCAL", + "speculative" : false, + "accumulatorUpdates" : [ ], + "taskMetrics" : { + "executorDeserializeTime" : 218, + "executorDeserializeCpuTime" : 51464000, + "executorRunTime" : 1618, + "executorCpuTime" : 490927000, + "resultSize" : 2115, + "jvmGcTime" : 66, + "resultSerializationTime" : 0, + "memoryBytesSpilled" : 0, + "diskBytesSpilled" : 0, + "peakExecutionMemory" : 24040, + "inputMetrics" : { + "bytesRead" : 0, + "recordsRead" : 0 + }, + "outputMetrics" : { + "bytesWritten" : 0, + "recordsWritten" : 0 + }, + "shuffleReadMetrics" : { + "remoteBlocksFetched" : 0, + "localBlocksFetched" : 0, + "fetchWaitTime" : 0, + "remoteBytesRead" : 0, + "remoteBytesReadToDisk" : 0, + "localBytesRead" : 0, + "recordsRead" : 0 + }, + "shuffleWriteMetrics" : { + "bytesWritten" : 0, + "writeTime" : 0, + "recordsWritten" : 0 + } + }, + "executorLogs" : { + "stdout" : "http://127.0.0.1:8081/logPage/?appId=app-20200706201101-0003&executorId=0&logType=stdout", + "stderr" : "http://127.0.0.1:8081/logPage/?appId=app-20200706201101-0003&executorId=0&logType=stderr" + }, + "schedulerDelay" : 69, + "gettingResultTime" : 0 + }, + "38" : { + "taskId" : 38, + "index" : 6, + "attempt" : 0, + "launchTime" : "2020-07-07T03:11:21.104GMT", + "duration" : 1835, + "executorId" : "0", + "host" : "127.0.0.1", + "status" : "SUCCESS", + "taskLocality" : "PROCESS_LOCAL", + "speculative" : false, + "accumulatorUpdates" : [ ], + "taskMetrics" : { + "executorDeserializeTime" : 255, + "executorDeserializeCpuTime" : 60358000, + "executorRunTime" : 1498, + "executorCpuTime" : 414110000, + "resultSize" : 2158, + "jvmGcTime" : 62, + "resultSerializationTime" : 11, + "memoryBytesSpilled" : 0, + "diskBytesSpilled" : 0, + "peakExecutionMemory" : 24040, + "inputMetrics" : { + "bytesRead" : 0, + "recordsRead" : 0 + }, + "outputMetrics" : { + "bytesWritten" : 0, + "recordsWritten" : 0 + }, + "shuffleReadMetrics" : { + "remoteBlocksFetched" : 0, + "localBlocksFetched" : 0, + "fetchWaitTime" : 0, + "remoteBytesRead" : 0, + "remoteBytesReadToDisk" : 0, + "localBytesRead" : 0, + "recordsRead" : 0 + }, + "shuffleWriteMetrics" : { + "bytesWritten" : 0, + "writeTime" : 0, + "recordsWritten" : 0 + } + }, + "executorLogs" : { + "stdout" : "http://127.0.0.1:8081/logPage/?appId=app-20200706201101-0003&executorId=0&logType=stdout", + "stderr" : "http://127.0.0.1:8081/logPage/?appId=app-20200706201101-0003&executorId=0&logType=stderr" + }, + "schedulerDelay" : 71, + "gettingResultTime" : 0 + }, + "33" : { + "taskId" : 33, + "index" : 1, + "attempt" : 0, + "launchTime" : "2020-07-07T03:11:21.082GMT", + "duration" : 1943, + "executorId" : "0", + "host" : "127.0.0.1", + "status" : "SUCCESS", + "taskLocality" : "PROCESS_LOCAL", + "speculative" : false, + "accumulatorUpdates" : [ ], + "taskMetrics" : { + "executorDeserializeTime" : 267, + "executorDeserializeCpuTime" : 54442000, + "executorRunTime" : 1597, + "executorCpuTime" : 519178000, + "resultSize" : 2115, + "jvmGcTime" : 62, + "resultSerializationTime" : 0, + "memoryBytesSpilled" : 0, + "diskBytesSpilled" : 0, + "peakExecutionMemory" : 24040, + "inputMetrics" : { + "bytesRead" : 0, + "recordsRead" : 0 + }, + "outputMetrics" : { + "bytesWritten" : 0, + "recordsWritten" : 0 + }, + "shuffleReadMetrics" : { + "remoteBlocksFetched" : 0, + "localBlocksFetched" : 0, + "fetchWaitTime" : 0, + "remoteBytesRead" : 0, + "remoteBytesReadToDisk" : 0, + "localBytesRead" : 0, + "recordsRead" : 0 + }, + "shuffleWriteMetrics" : { + "bytesWritten" : 0, + "writeTime" : 0, + "recordsWritten" : 0 + } + }, + "executorLogs" : { + "stdout" : "http://127.0.0.1:8081/logPage/?appId=app-20200706201101-0003&executorId=0&logType=stdout", + "stderr" : "http://127.0.0.1:8081/logPage/?appId=app-20200706201101-0003&executorId=0&logType=stderr" + }, + "schedulerDelay" : 79, + "gettingResultTime" : 0 + }, + "41" : { + "taskId" : 41, + "index" : 9, + "attempt" : 0, + "launchTime" : "2020-07-07T03:11:21.116GMT", + "duration" : 1916, + "executorId" : "0", + "host" : "127.0.0.1", + "status" : "SUCCESS", + "taskLocality" : "PROCESS_LOCAL", + "speculative" : false, + "accumulatorUpdates" : [ ], + "taskMetrics" : { + "executorDeserializeTime" : 240, + "executorDeserializeCpuTime" : 55787000, + "executorRunTime" : 1614, + "executorCpuTime" : 489923000, + "resultSize" : 2115, + "jvmGcTime" : 66, + "resultSerializationTime" : 0, + "memoryBytesSpilled" : 0, + "diskBytesSpilled" : 0, + "peakExecutionMemory" : 24040, + "inputMetrics" : { + "bytesRead" : 0, + "recordsRead" : 0 + }, + "outputMetrics" : { + "bytesWritten" : 0, + "recordsWritten" : 0 + }, + "shuffleReadMetrics" : { + "remoteBlocksFetched" : 0, + "localBlocksFetched" : 0, + "fetchWaitTime" : 0, + "remoteBytesRead" : 0, + "remoteBytesReadToDisk" : 0, + "localBytesRead" : 0, + "recordsRead" : 0 + }, + "shuffleWriteMetrics" : { + "bytesWritten" : 0, + "writeTime" : 0, + "recordsWritten" : 0 + } + }, + "executorLogs" : { + "stdout" : "http://127.0.0.1:8081/logPage/?appId=app-20200706201101-0003&executorId=0&logType=stdout", + "stderr" : "http://127.0.0.1:8081/logPage/?appId=app-20200706201101-0003&executorId=0&logType=stderr" + }, + "schedulerDelay" : 62, + "gettingResultTime" : 0 + }, + "32" : { + "taskId" : 32, + "index" : 0, + "attempt" : 0, + "launchTime" : "2020-07-07T03:11:21.077GMT", + "duration" : 1960, + "executorId" : "0", + "host" : "127.0.0.1", + "status" : "SUCCESS", + "taskLocality" : "PROCESS_LOCAL", + "speculative" : false, + "accumulatorUpdates" : [ ], + "taskMetrics" : { + "executorDeserializeTime" : 271, + "executorDeserializeCpuTime" : 56827000, + "executorRunTime" : 1619, + "executorCpuTime" : 496683000, + "resultSize" : 2115, + "jvmGcTime" : 66, + "resultSerializationTime" : 0, + "memoryBytesSpilled" : 0, + "diskBytesSpilled" : 0, + "peakExecutionMemory" : 24040, + "inputMetrics" : { + "bytesRead" : 0, + "recordsRead" : 0 + }, + "outputMetrics" : { + "bytesWritten" : 0, + "recordsWritten" : 0 + }, + "shuffleReadMetrics" : { + "remoteBlocksFetched" : 0, + "localBlocksFetched" : 0, + "fetchWaitTime" : 0, + "remoteBytesRead" : 0, + "remoteBytesReadToDisk" : 0, + "localBytesRead" : 0, + "recordsRead" : 0 + }, + "shuffleWriteMetrics" : { + "bytesWritten" : 0, + "writeTime" : 0, + "recordsWritten" : 0 + } + }, + "executorLogs" : { + "stdout" : "http://127.0.0.1:8081/logPage/?appId=app-20200706201101-0003&executorId=0&logType=stdout", + "stderr" : "http://127.0.0.1:8081/logPage/?appId=app-20200706201101-0003&executorId=0&logType=stderr" + }, + "schedulerDelay" : 70, + "gettingResultTime" : 0 + }, + "34" : { + "taskId" : 34, + "index" : 2, + "attempt" : 0, + "launchTime" : "2020-07-07T03:11:21.087GMT", + "duration" : 1939, + "executorId" : "0", + "host" : "127.0.0.1", + "status" : "SUCCESS", + "taskLocality" : "PROCESS_LOCAL", + "speculative" : false, + "accumulatorUpdates" : [ ], + "taskMetrics" : { + "executorDeserializeTime" : 265, + "executorDeserializeCpuTime" : 69492000, + "executorRunTime" : 1606, + "executorCpuTime" : 508433000, + "resultSize" : 2115, + "jvmGcTime" : 66, + "resultSerializationTime" : 0, + "memoryBytesSpilled" : 0, + "diskBytesSpilled" : 0, + "peakExecutionMemory" : 24040, + "inputMetrics" : { + "bytesRead" : 0, + "recordsRead" : 0 + }, + "outputMetrics" : { + "bytesWritten" : 0, + "recordsWritten" : 0 + }, + "shuffleReadMetrics" : { + "remoteBlocksFetched" : 0, + "localBlocksFetched" : 0, + "fetchWaitTime" : 0, + "remoteBytesRead" : 0, + "remoteBytesReadToDisk" : 0, + "localBytesRead" : 0, + "recordsRead" : 0 + }, + "shuffleWriteMetrics" : { + "bytesWritten" : 0, + "writeTime" : 0, + "recordsWritten" : 0 + } + }, + "executorLogs" : { + "stdout" : "http://127.0.0.1:8081/logPage/?appId=app-20200706201101-0003&executorId=0&logType=stdout", + "stderr" : "http://127.0.0.1:8081/logPage/?appId=app-20200706201101-0003&executorId=0&logType=stderr" + }, + "schedulerDelay" : 68, + "gettingResultTime" : 0 + }, + "45" : { + "taskId" : 45, + "index" : 13, + "attempt" : 0, + "launchTime" : "2020-07-07T03:11:21.129GMT", + "duration" : 1895, + "executorId" : "0", + "host" : "127.0.0.1", + "status" : "SUCCESS", + "taskLocality" : "PROCESS_LOCAL", + "speculative" : false, + "accumulatorUpdates" : [ ], + "taskMetrics" : { + "executorDeserializeTime" : 221, + "executorDeserializeCpuTime" : 54222000, + "executorRunTime" : 1595, + "executorCpuTime" : 495138000, + "resultSize" : 2115, + "jvmGcTime" : 62, + "resultSerializationTime" : 0, + "memoryBytesSpilled" : 0, + "diskBytesSpilled" : 0, + "peakExecutionMemory" : 24040, + "inputMetrics" : { + "bytesRead" : 0, + "recordsRead" : 0 + }, + "outputMetrics" : { + "bytesWritten" : 0, + "recordsWritten" : 0 + }, + "shuffleReadMetrics" : { + "remoteBlocksFetched" : 0, + "localBlocksFetched" : 0, + "fetchWaitTime" : 0, + "remoteBytesRead" : 0, + "remoteBytesReadToDisk" : 0, + "localBytesRead" : 0, + "recordsRead" : 0 + }, + "shuffleWriteMetrics" : { + "bytesWritten" : 0, + "writeTime" : 0, + "recordsWritten" : 0 + } + }, + "executorLogs" : { + "stdout" : "http://127.0.0.1:8081/logPage/?appId=app-20200706201101-0003&executorId=0&logType=stdout", + "stderr" : "http://127.0.0.1:8081/logPage/?appId=app-20200706201101-0003&executorId=0&logType=stderr" + }, + "schedulerDelay" : 79, + "gettingResultTime" : 0 + }, + "44" : { + "taskId" : 44, + "index" : 12, + "attempt" : 0, + "launchTime" : "2020-07-07T03:11:21.126GMT", + "duration" : 1917, + "executorId" : "0", + "host" : "127.0.0.1", + "status" : "SUCCESS", + "taskLocality" : "PROCESS_LOCAL", + "speculative" : false, + "accumulatorUpdates" : [ ], + "taskMetrics" : { + "executorDeserializeTime" : 222, + "executorDeserializeCpuTime" : 51988000, + "executorRunTime" : 1624, + "executorCpuTime" : 498187000, + "resultSize" : 2115, + "jvmGcTime" : 66, + "resultSerializationTime" : 0, + "memoryBytesSpilled" : 0, + "diskBytesSpilled" : 0, + "peakExecutionMemory" : 24040, + "inputMetrics" : { + "bytesRead" : 0, + "recordsRead" : 0 + }, + "outputMetrics" : { + "bytesWritten" : 0, + "recordsWritten" : 0 + }, + "shuffleReadMetrics" : { + "remoteBlocksFetched" : 0, + "localBlocksFetched" : 0, + "fetchWaitTime" : 0, + "remoteBytesRead" : 0, + "remoteBytesReadToDisk" : 0, + "localBytesRead" : 0, + "recordsRead" : 0 + }, + "shuffleWriteMetrics" : { + "bytesWritten" : 0, + "writeTime" : 0, + "recordsWritten" : 0 + } + }, + "executorLogs" : { + "stdout" : "http://127.0.0.1:8081/logPage/?appId=app-20200706201101-0003&executorId=0&logType=stdout", + "stderr" : "http://127.0.0.1:8081/logPage/?appId=app-20200706201101-0003&executorId=0&logType=stderr" + }, + "schedulerDelay" : 71, + "gettingResultTime" : 0 + }, + "39" : { + "taskId" : 39, + "index" : 7, + "attempt" : 0, + "launchTime" : "2020-07-07T03:11:21.109GMT", + "duration" : 1915, + "executorId" : "0", + "host" : "127.0.0.1", + "status" : "SUCCESS", + "taskLocality" : "PROCESS_LOCAL", + "speculative" : false, + "accumulatorUpdates" : [ ], + "taskMetrics" : { + "executorDeserializeTime" : 254, + "executorDeserializeCpuTime" : 64380000, + "executorRunTime" : 1596, + "executorCpuTime" : 539451000, + "resultSize" : 2115, + "jvmGcTime" : 62, + "resultSerializationTime" : 0, + "memoryBytesSpilled" : 0, + "diskBytesSpilled" : 0, + "peakExecutionMemory" : 24040, + "inputMetrics" : { + "bytesRead" : 0, + "recordsRead" : 0 + }, + "outputMetrics" : { + "bytesWritten" : 0, + "recordsWritten" : 0 + }, + "shuffleReadMetrics" : { + "remoteBlocksFetched" : 0, + "localBlocksFetched" : 0, + "fetchWaitTime" : 0, + "remoteBytesRead" : 0, + "remoteBytesReadToDisk" : 0, + "localBytesRead" : 0, + "recordsRead" : 0 + }, + "shuffleWriteMetrics" : { + "bytesWritten" : 0, + "writeTime" : 0, + "recordsWritten" : 0 + } + }, + "executorLogs" : { + "stdout" : "http://127.0.0.1:8081/logPage/?appId=app-20200706201101-0003&executorId=0&logType=stdout", + "stderr" : "http://127.0.0.1:8081/logPage/?appId=app-20200706201101-0003&executorId=0&logType=stderr" + }, + "schedulerDelay" : 65, + "gettingResultTime" : 0 + }, + "35" : { + "taskId" : 35, + "index" : 3, + "attempt" : 0, + "launchTime" : "2020-07-07T03:11:21.091GMT", + "duration" : 1925, + "executorId" : "0", + "host" : "127.0.0.1", + "status" : "SUCCESS", + "taskLocality" : "PROCESS_LOCAL", + "speculative" : false, + "accumulatorUpdates" : [ ], + "taskMetrics" : { + "executorDeserializeTime" : 263, + "executorDeserializeCpuTime" : 62944000, + "executorRunTime" : 1598, + "executorCpuTime" : 502908000, + "resultSize" : 2115, + "jvmGcTime" : 62, + "resultSerializationTime" : 0, + "memoryBytesSpilled" : 0, + "diskBytesSpilled" : 0, + "peakExecutionMemory" : 24040, + "inputMetrics" : { + "bytesRead" : 0, + "recordsRead" : 0 + }, + "outputMetrics" : { + "bytesWritten" : 0, + "recordsWritten" : 0 + }, + "shuffleReadMetrics" : { + "remoteBlocksFetched" : 0, + "localBlocksFetched" : 0, + "fetchWaitTime" : 0, + "remoteBytesRead" : 0, + "remoteBytesReadToDisk" : 0, + "localBytesRead" : 0, + "recordsRead" : 0 + }, + "shuffleWriteMetrics" : { + "bytesWritten" : 0, + "writeTime" : 0, + "recordsWritten" : 0 + } + }, + "executorLogs" : { + "stdout" : "http://127.0.0.1:8081/logPage/?appId=app-20200706201101-0003&executorId=0&logType=stdout", + "stderr" : "http://127.0.0.1:8081/logPage/?appId=app-20200706201101-0003&executorId=0&logType=stderr" + }, + "schedulerDelay" : 64, + "gettingResultTime" : 0 + }, + "43" : { + "taskId" : 43, + "index" : 11, + "attempt" : 0, + "launchTime" : "2020-07-07T03:11:21.123GMT", + "duration" : 1906, + "executorId" : "0", + "host" : "127.0.0.1", + "status" : "SUCCESS", + "taskLocality" : "PROCESS_LOCAL", + "speculative" : false, + "accumulatorUpdates" : [ ], + "taskMetrics" : { + "executorDeserializeTime" : 225, + "executorDeserializeCpuTime" : 48849000, + "executorRunTime" : 1609, + "executorCpuTime" : 502120000, + "resultSize" : 2115, + "jvmGcTime" : 66, + "resultSerializationTime" : 0, + "memoryBytesSpilled" : 0, + "diskBytesSpilled" : 0, + "peakExecutionMemory" : 24040, + "inputMetrics" : { + "bytesRead" : 0, + "recordsRead" : 0 + }, + "outputMetrics" : { + "bytesWritten" : 0, + "recordsWritten" : 0 + }, + "shuffleReadMetrics" : { + "remoteBlocksFetched" : 0, + "localBlocksFetched" : 0, + "fetchWaitTime" : 0, + "remoteBytesRead" : 0, + "remoteBytesReadToDisk" : 0, + "localBytesRead" : 0, + "recordsRead" : 0 + }, + "shuffleWriteMetrics" : { + "bytesWritten" : 0, + "writeTime" : 0, + "recordsWritten" : 0 + } + }, + "executorLogs" : { + "stdout" : "http://127.0.0.1:8081/logPage/?appId=app-20200706201101-0003&executorId=0&logType=stdout", + "stderr" : "http://127.0.0.1:8081/logPage/?appId=app-20200706201101-0003&executorId=0&logType=stderr" + }, + "schedulerDelay" : 72, + "gettingResultTime" : 0 + }, + "40" : { + "taskId" : 40, + "index" : 8, + "attempt" : 0, + "launchTime" : "2020-07-07T03:11:21.112GMT", + "duration" : 1904, + "executorId" : "0", + "host" : "127.0.0.1", + "status" : "SUCCESS", + "taskLocality" : "PROCESS_LOCAL", + "speculative" : false, + "accumulatorUpdates" : [ ], + "taskMetrics" : { + "executorDeserializeTime" : 246, + "executorDeserializeCpuTime" : 69760000, + "executorRunTime" : 1595, + "executorCpuTime" : 510597000, + "resultSize" : 2115, + "jvmGcTime" : 62, + "resultSerializationTime" : 0, + "memoryBytesSpilled" : 0, + "diskBytesSpilled" : 0, + "peakExecutionMemory" : 24040, + "inputMetrics" : { + "bytesRead" : 0, + "recordsRead" : 0 + }, + "outputMetrics" : { + "bytesWritten" : 0, + "recordsWritten" : 0 + }, + "shuffleReadMetrics" : { + "remoteBlocksFetched" : 0, + "localBlocksFetched" : 0, + "fetchWaitTime" : 0, + "remoteBytesRead" : 0, + "remoteBytesReadToDisk" : 0, + "localBytesRead" : 0, + "recordsRead" : 0 + }, + "shuffleWriteMetrics" : { + "bytesWritten" : 0, + "writeTime" : 0, + "recordsWritten" : 0 + } + }, + "executorLogs" : { + "stdout" : "http://127.0.0.1:8081/logPage/?appId=app-20200706201101-0003&executorId=0&logType=stdout", + "stderr" : "http://127.0.0.1:8081/logPage/?appId=app-20200706201101-0003&executorId=0&logType=stderr" + }, + "schedulerDelay" : 63, + "gettingResultTime" : 0 + }, + "36" : { + "taskId" : 36, + "index" : 4, + "attempt" : 0, + "launchTime" : "2020-07-07T03:11:21.095GMT", + "duration" : 1920, + "executorId" : "0", + "host" : "127.0.0.1", + "status" : "SUCCESS", + "taskLocality" : "PROCESS_LOCAL", + "speculative" : false, + "accumulatorUpdates" : [ ], + "taskMetrics" : { + "executorDeserializeTime" : 260, + "executorDeserializeCpuTime" : 112849000, + "executorRunTime" : 1596, + "executorCpuTime" : 503010000, + "resultSize" : 2115, + "jvmGcTime" : 62, + "resultSerializationTime" : 0, + "memoryBytesSpilled" : 0, + "diskBytesSpilled" : 0, + "peakExecutionMemory" : 24040, + "inputMetrics" : { + "bytesRead" : 0, + "recordsRead" : 0 + }, + "outputMetrics" : { + "bytesWritten" : 0, + "recordsWritten" : 0 + }, + "shuffleReadMetrics" : { + "remoteBlocksFetched" : 0, + "localBlocksFetched" : 0, + "fetchWaitTime" : 0, + "remoteBytesRead" : 0, + "remoteBytesReadToDisk" : 0, + "localBytesRead" : 0, + "recordsRead" : 0 + }, + "shuffleWriteMetrics" : { + "bytesWritten" : 0, + "writeTime" : 0, + "recordsWritten" : 0 + } + }, + "executorLogs" : { + "stdout" : "http://127.0.0.1:8081/logPage/?appId=app-20200706201101-0003&executorId=0&logType=stdout", + "stderr" : "http://127.0.0.1:8081/logPage/?appId=app-20200706201101-0003&executorId=0&logType=stderr" + }, + "schedulerDelay" : 64, + "gettingResultTime" : 0 + }, + "47" : { + "taskId" : 47, + "index" : 15, + "attempt" : 0, + "launchTime" : "2020-07-07T03:11:21.136GMT", + "duration" : 1878, + "executorId" : "0", + "host" : "127.0.0.1", + "status" : "SUCCESS", + "taskLocality" : "PROCESS_LOCAL", + "speculative" : false, + "accumulatorUpdates" : [ ], + "taskMetrics" : { + "executorDeserializeTime" : 213, + "executorDeserializeCpuTime" : 47496000, + "executorRunTime" : 1594, + "executorCpuTime" : 1324251000, + "resultSize" : 2115, + "jvmGcTime" : 52, + "resultSerializationTime" : 0, + "memoryBytesSpilled" : 0, + "diskBytesSpilled" : 0, + "peakExecutionMemory" : 24040, + "inputMetrics" : { + "bytesRead" : 0, + "recordsRead" : 0 + }, + "outputMetrics" : { + "bytesWritten" : 0, + "recordsWritten" : 0 + }, + "shuffleReadMetrics" : { + "remoteBlocksFetched" : 0, + "localBlocksFetched" : 0, + "fetchWaitTime" : 0, + "remoteBytesRead" : 0, + "remoteBytesReadToDisk" : 0, + "localBytesRead" : 0, + "recordsRead" : 0 + }, + "shuffleWriteMetrics" : { + "bytesWritten" : 0, + "writeTime" : 0, + "recordsWritten" : 0 + } + }, + "executorLogs" : { + "stdout" : "http://127.0.0.1:8081/logPage/?appId=app-20200706201101-0003&executorId=0&logType=stdout", + "stderr" : "http://127.0.0.1:8081/logPage/?appId=app-20200706201101-0003&executorId=0&logType=stderr" + }, + "schedulerDelay" : 71, + "gettingResultTime" : 0 + } + }, + "executorSummary" : { + "0" : { + "taskTime" : 30596, + "failedTasks" : 0, + "succeededTasks" : 16, + "killedTasks" : 0, + "inputBytes" : 0, + "inputRecords" : 0, + "outputBytes" : 0, + "outputRecords" : 0, + "shuffleRead" : 0, + "shuffleReadRecords" : 0, + "shuffleWrite" : 0, + "shuffleWriteRecords" : 0, + "memoryBytesSpilled" : 0, + "diskBytesSpilled" : 0, + "isBlacklistedForStage" : false, + "peakMemoryMetrics" : { + "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 + }, + "isExcludedForStage" : false + }, + "driver" : { + "taskTime" : 0, + "failedTasks" : 0, + "succeededTasks" : 0, + "killedTasks" : 0, + "inputBytes" : 0, + "inputRecords" : 0, + "outputBytes" : 0, + "outputRecords" : 0, + "shuffleRead" : 0, + "shuffleReadRecords" : 0, + "shuffleWrite" : 0, + "shuffleWriteRecords" : 0, + "memoryBytesSpilled" : 0, + "diskBytesSpilled" : 0, + "isBlacklistedForStage" : false, + "peakMemoryMetrics" : { + "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 + }, + "isExcludedForStage" : false + } + }, + "killedTasksSummary" : { }, + "resourceProfileId" : 0, + "peakExecutorMetrics" : { + "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 + }, + "taskMetricsDistributions" : { + "quantiles" : [ 0.0, 0.25, 0.5, 0.75, 1.0 ], + "executorDeserializeTime" : [ 213.0, 225.0, 254.0, 263.0, 271.0 ], + "executorDeserializeCpuTime" : [ 4.7496E7, 5.4222E7, 5.8152E7, 6.438E7, 1.12849E8 ], + "executorRunTime" : [ 1498.0, 1596.0, 1598.0, 1618.0, 1624.0 ], + "executorCpuTime" : [ 4.1411E8, 4.96683E8, 5.0301E8, 5.10597E8, 1.324251E9 ], + "resultSize" : [ 2115.0, 2115.0, 2115.0, 2115.0, 2158.0 ], + "jvmGcTime" : [ 52.0, 62.0, 62.0, 66.0, 66.0 ], + "resultSerializationTime" : [ 0.0, 0.0, 0.0, 0.0, 11.0 ], + "gettingResultTime" : [ 0.0, 0.0, 0.0, 0.0, 0.0 ], + "schedulerDelay" : [ 62.0, 64.0, 70.0, 71.0, 79.0 ], + "peakExecutionMemory" : [ 24040.0, 24040.0, 24040.0, 24040.0, 24040.0 ], + "memoryBytesSpilled" : [ 0.0, 0.0, 0.0, 0.0, 0.0 ], + "diskBytesSpilled" : [ 0.0, 0.0, 0.0, 0.0, 0.0 ], + "inputMetrics" : { + "bytesRead" : [ 0.0, 0.0, 0.0, 0.0, 0.0 ], + "recordsRead" : [ 0.0, 0.0, 0.0, 0.0, 0.0 ] + }, + "outputMetrics" : { + "bytesWritten" : [ 0.0, 0.0, 0.0, 0.0, 0.0 ], + "recordsWritten" : [ 0.0, 0.0, 0.0, 0.0, 0.0 ] + }, + "shuffleReadMetrics" : { + "readBytes" : [ 0.0, 0.0, 0.0, 0.0, 0.0 ], + "readRecords" : [ 0.0, 0.0, 0.0, 0.0, 0.0 ], + "remoteBlocksFetched" : [ 0.0, 0.0, 0.0, 0.0, 0.0 ], + "localBlocksFetched" : [ 0.0, 0.0, 0.0, 0.0, 0.0 ], + "fetchWaitTime" : [ 0.0, 0.0, 0.0, 0.0, 0.0 ], + "remoteBytesRead" : [ 0.0, 0.0, 0.0, 0.0, 0.0 ], + "remoteBytesReadToDisk" : [ 0.0, 0.0, 0.0, 0.0, 0.0 ], + "totalBlocksFetched" : [ 0.0, 0.0, 0.0, 0.0, 0.0 ] + }, + "shuffleWriteMetrics" : { + "writeBytes" : [ 0.0, 0.0, 0.0, 0.0, 0.0 ], + "writeRecords" : [ 0.0, 0.0, 0.0, 0.0, 0.0 ], + "writeTime" : [ 0.0, 0.0, 0.0, 0.0, 0.0 ] + } + }, + "executorMetricsDistributions" : { + "quantiles" : [ 0.0, 0.25, 0.5, 0.75, 1.0 ], + "taskTime" : [ 0.0, 0.0, 30596.0, 30596.0, 30596.0 ], + "failedTasks" : [ 0.0, 0.0, 0.0, 0.0, 0.0 ], + "succeededTasks" : [ 0.0, 0.0, 16.0, 16.0, 16.0 ], + "killedTasks" : [ 0.0, 0.0, 0.0, 0.0, 0.0 ], + "inputBytes" : [ 0.0, 0.0, 0.0, 0.0, 0.0 ], + "inputRecords" : [ 0.0, 0.0, 0.0, 0.0, 0.0 ], + "outputBytes" : [ 0.0, 0.0, 0.0, 0.0, 0.0 ], + "outputRecords" : [ 0.0, 0.0, 0.0, 0.0, 0.0 ], + "shuffleRead" : [ 0.0, 0.0, 0.0, 0.0, 0.0 ], + "shuffleReadRecords" : [ 0.0, 0.0, 0.0, 0.0, 0.0 ], + "shuffleWrite" : [ 0.0, 0.0, 0.0, 0.0, 0.0 ], + "shuffleWriteRecords" : [ 0.0, 0.0, 0.0, 0.0, 0.0 ], + "memoryBytesSpilled" : [ 0.0, 0.0, 0.0, 0.0, 0.0 ], + "diskBytesSpilled" : [ 0.0, 0.0, 0.0, 0.0, 0.0 ], + "peakMemoryMetrics" : { + "JVMHeapMemory" : [ 0.0, 0.0, 2.13367864E8, 2.13367864E8, 2.13367864E8 ], + "JVMOffHeapMemory" : [ 0.0, 0.0, 1.89011656E8, 1.89011656E8, 1.89011656E8 ], + "OnHeapExecutionMemory" : [ 0.0, 0.0, 0.0, 0.0, 0.0 ], + "OffHeapExecutionMemory" : [ 0.0, 0.0, 0.0, 0.0, 0.0 ], + "OnHeapStorageMemory" : [ 0.0, 0.0, 2133349.0, 2133349.0, 2133349.0 ], + "OffHeapStorageMemory" : [ 0.0, 0.0, 0.0, 0.0, 0.0 ], + "OnHeapUnifiedMemory" : [ 0.0, 0.0, 2133349.0, 2133349.0, 2133349.0 ], + "OffHeapUnifiedMemory" : [ 0.0, 0.0, 0.0, 0.0, 0.0 ], + "DirectPoolMemory" : [ 0.0, 0.0, 282024.0, 282024.0, 282024.0 ], + "MappedPoolMemory" : [ 0.0, 0.0, 0.0, 0.0, 0.0 ], + "ProcessTreeJVMVMemory" : [ 0.0, 0.0, 0.0, 0.0, 0.0 ], + "ProcessTreeJVMRSSMemory" : [ 0.0, 0.0, 0.0, 0.0, 0.0 ], + "ProcessTreePythonVMemory" : [ 0.0, 0.0, 0.0, 0.0, 0.0 ], + "ProcessTreePythonRSSMemory" : [ 0.0, 0.0, 0.0, 0.0, 0.0 ], + "ProcessTreeOtherVMemory" : [ 0.0, 0.0, 0.0, 0.0, 0.0 ], + "ProcessTreeOtherRSSMemory" : [ 0.0, 0.0, 0.0, 0.0, 0.0 ], + "MinorGCCount" : [ 0.0, 0.0, 13.0, 13.0, 13.0 ], + "MinorGCTime" : [ 0.0, 0.0, 115.0, 115.0, 115.0 ], + "MajorGCCount" : [ 0.0, 0.0, 4.0, 4.0, 4.0 ], + "MajorGCTime" : [ 0.0, 0.0, 339.0, 339.0, 339.0 ] + } + } +} diff --git a/core/src/test/scala/org/apache/spark/deploy/history/HistoryServerSuite.scala b/core/src/test/scala/org/apache/spark/deploy/history/HistoryServerSuite.scala index 938eb8d..24640b3 100644 --- a/core/src/test/scala/org/apache/spark/deploy/history/HistoryServerSuite.scala +++ b/core/src/test/scala/org/apache/spark/deploy/history/HistoryServerSuite.scala @@ -181,6 +181,7 @@ class HistoryServerSuite extends SparkFunSuite with BeforeAndAfter with Matchers "multiple resource profiles" -> "applications/application_1578436911597_0052/environment", "stage list with peak metrics" -> "applications/app-20200706201101-0003/stages", "stage with peak metrics" -> "applications/app-20200706201101-0003/stages/2/0", + "stage with summaries" -> "applications/app-20200706201101-0003/stages/2/0?withSummaries=true", "app environment" -> "applications/app-20161116163331-0000/environment", diff --git a/core/src/test/scala/org/apache/spark/ui/StagePageSuite.scala b/core/src/test/scala/org/apache/spark/ui/StagePageSuite.scala index d02d7f8..9f0b73f 100644 --- a/core/src/test/scala/org/apache/spark/ui/StagePageSuite.scala +++ b/core/src/test/scala/org/apache/spark/ui/StagePageSuite.scala @@ -93,7 +93,9 @@ class StagePageSuite extends SparkFunSuite with LocalSparkContext { executorSummary = None, killedTasksSummary = Map.empty, ResourceProfile.DEFAULT_RESOURCE_PROFILE_ID, - peakExecutorMetrics = None + peakExecutorMetrics = None, + taskMetricsDistributions = None, + executorMetricsDistributions = None ) val taskTable = new TaskPagedTable( stageData, diff --git a/docs/monitoring.md b/docs/monitoring.md index 930f91f..8fec5ad 100644 --- a/docs/monitoring.md +++ b/docs/monitoring.md @@ -479,11 +479,27 @@ can be identified by their `[attempt-id]`. In the API listed below, when running <td><code>/applications/[app-id]/stages/[stage-id]</code></td> <td> A list of all attempts for the given stage. + <br><code>?details=true</code> lists all attempts with the task data for the given stage. + <br><code>?withSummaries=true</code> lists task metrics distribution and executor metrics distribution of each attempt. + <br><code>?quantiles=0.1,0.25,0.5,0.75,1.0</code> summarize the metrics with the given quantiles. Query parameter quantiles takes effect only when <code>withSummaries=true</code>. Default value is <code>0.0,0.25,0.5,0.75,1.0</code>. + <br>Example: + <br><code>?details=true</code> + <br><code>?withSummaries=true</code> + <br><code>?details=true&withSummaries=true&quantiles=0.01,0.5,0.99</code> </td> </tr> <tr> <td><code>/applications/[app-id]/stages/[stage-id]/[stage-attempt-id]</code></td> - <td>Details for the given stage attempt.</td> + <td> + Details for the given stage attempt. + <br><code>?details=true</code> lists all task data for the given stage attempt. + <br><code>?withSummaries=true</code> lists task metrics distribution and executor metrics distribution for the given stage attempt. + <br><code>?quantiles=0.1,0.25,0.5,0.75,1.0</code> summarize the metrics with the given quantiles. Query parameter quantiles takes effect only when <code>withSummaries=true</code>. Default value is <code>0.0,0.25,0.5,0.75,1.0</code>. + <br>Example: + <br><code>?details=true</code> + <br><code>?withSummaries=true</code> + <br><code>?details=true&withSummaries=true&quantiles=0.01,0.5,0.99</code> + </td> </tr> <tr> <td><code>/applications/[app-id]/stages/[stage-id]/[stage-attempt-id]/taskSummary</code></td> --------------------------------------------------------------------- To unsubscribe, e-mail: commits-unsubscr...@spark.apache.org For additional commands, e-mail: commits-h...@spark.apache.org