Skip to content

[SPARK-26003] Improve SQLAppStatusListener.aggregateMetrics performance #23002

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
mgaido91 wants to merge 2 commits into from
Closed

[SPARK-26003] Improve SQLAppStatusListener.aggregateMetrics performance #23002

Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
7e79041
[SPARK-26003] Improve SQLAppStatusListener.aggregateMetrics performance
mgaido91 Nov 10, 2018
031d512
address comment
mgaido91 Nov 12, 2018
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
5 changes: 2 additions & 3 deletions sql/core/src/main/scala/org/apache/spark/sql/execution/ui/SQLAppStatusListener.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -159,18 +159,17 @@ class SQLAppStatusListener(
}

private def aggregateMetrics(exec: LiveExecutionData): Map[Long, String] = {
val metricIds = exec.metrics.map(_.accumulatorId).sorted
val metricTypes = exec.metrics.map { m => (m.accumulatorId, m.metricType) }.toMap
val metrics = exec.stages.toSeq
.flatMap { stageId => Option(stageMetrics.get(stageId)) }
Copy link
Member

@gengliangwang gengliangwang Nov 12, 2018

Choose a reason for hiding this comment

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

Consider also change the following flatMap / filter / groupBy into while loop

Copy link
Contributor Author

@mgaido91 mgaido91 Nov 12, 2018

Choose a reason for hiding this comment

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

not sure what you mean here. Why should we use a while loop?

Copy link
Member

@gengliangwang gengliangwang Nov 12, 2018

Choose a reason for hiding this comment

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

If the metrics is large, then using a while loop can reduce the number of traversal loops. And it is not complicated to do it in the code here.

Copy link
Member

@gengliangwang gengliangwang Nov 12, 2018

Choose a reason for hiding this comment

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

I am also fine with the current code here.

Copy link
Contributor Author

@mgaido91 mgaido91 Nov 12, 2018

Choose a reason for hiding this comment

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

yes, we can save 1 traversal, but I am not sure it is worth honestly... This approach seems cleaner to me.

.flatMap(_.taskMetrics.values().asScala)
.flatMap { metrics => metrics.ids.zip(metrics.values) }

val aggregatedMetrics = (metrics ++ exec.driverAccumUpdates.toSeq)
.filter { case (id, _) => metricIds.contains(id) }
.filter { case (id, _) => metricTypes.contains(id) }
.groupBy(_._1)
.map { case (id, values) =>
id -> SQLMetrics.stringValue(metricTypes(id), values.map(_._2).toSeq)
id -> SQLMetrics.stringValue(metricTypes(id), values.map(_._2))
}

// Check the execution again for whether the aggregated metrics data has been calculated.
Expand Down