[SPARK-20776] Fix perf. problems in JobProgressListener caused by TaskMetrics construction #18008

JoshRosen · 2017-05-16T22:39:28Z

What changes were proposed in this pull request?

In

./bin/spark-shell --master=local[64]

I ran

sc.parallelize(1 to 100000, 100000).count()

and profiled the time spend in the LiveListenerBus event processing thread. I discovered that the majority of the time was being spent in TaskMetrics.empty calls in JobProgressListener.onTaskStart. It turns out that we can slightly refactor to remove the need to construct one empty instance per call, greatly improving the performance of this code.

The performance gains here help to avoid an issue where listener events would be dropped because the JobProgressListener couldn't keep up with the throughput.

Before:

After:

How was this patch tested?

Benchmarks described above.

JoshRosen · 2017-05-16T22:43:44Z

core/src/main/scala/org/apache/spark/executor/TaskMetrics.scala



  import InternalAccumulator._
-  @transient private[spark] lazy val nameToAccums = LinkedHashMap(


It looks like the use of LinkedHashMap was added by @cloud-fan in #12612 in order to preserve ordering from the old code. As far as I can tell we don't actually rely on the ordering of the entries in this map, so I didn't preserved the use of LinkedHashMap.

JoshRosen · 2017-05-16T22:47:31Z

core/src/main/scala/org/apache/spark/executor/TaskMetrics.scala

+    testAccum.foreach { accum =>
+      map.put(TEST_ACCUM, accum)
+    }
+    map.asScala


The map + wrapper might consume a little bit of extra memory compared to the old code but it doesn't matter because we don't have that many TaskMetrics resident in the JVM at the same time: in the executor, the only instances are in TaskContexts and in the driver you only have one per stage in the scheduler and some temporary ones in the listener bus queue which are freed as soon as the queue events are processed (which happens faster now, outweighing the extra space usage).

JoshRosen · 2017-05-16T23:07:17Z

Actually, stepping back a second, we might be able to completely remove this bottleneck by simply not constructing tons of empty TaskMetrics objects in JobProgressListener's hot path. Let me see if I can update to do that instead.

This reverts commit 4675b21.

This reverts commit 622951f.

JoshRosen · 2017-05-16T23:44:02Z

core/src/main/scala/org/apache/spark/ui/jobs/JobProgressListener.scala

        updateAggregateMetrics(stageData, info.executorId, m, oldMetrics)
      }

-      val taskData = stageData.taskData.getOrElseUpdate(info.taskId, TaskUIData(info, None))


Important note here: in the old code, the elseUpdate branch would only be taken in rare error cases where we somehow purged the TaskUIData which should have been created when the task launched. It technically doesn't matter what we put in for the Option[Metrics] here since it just gets unconditionally overwritten on line 410 in the old code. So while my new code constructs TaskUIData with default metrics it doesn't actually change the behavior of this block.

cloud-fan · 2017-05-17T00:09:23Z

core/src/main/scala/org/apache/spark/ui/jobs/UIData.scala

-      private var _metrics: Option[TaskMetricsUIData]) {
+  class TaskUIData private(private var _taskInfo: TaskInfo) {
+
+    private[this] var _metrics: Option[TaskMetricsUIData] = Some(TaskMetricsUIData.EMPTY)


when will this be None?

The only way for this to become None is if updateTaskMetrics is called with None.

updateTaskMetrics is called in two places:

In JobProgressListener.onTaskEnd, where the metrics are from Option(taskEnd.taskMetrics), where taskEnd.taskMetrics can be null in case the task has failed (according to docs).

In JobProgressListener.onExecutorMetricsUpdate, where the metrics are guaranteed to be defined / non-None.

cloud-fan · 2017-05-17T00:16:11Z

LGTM

SparkQA · 2017-05-17T00:45:28Z

Test build #76984 has finished for PR 18008 at commit 4675b21.

This patch fails Spark unit tests.
This patch merges cleanly.
This patch adds no public classes.

SparkQA · 2017-05-17T02:04:56Z

Test build #76988 has finished for PR 18008 at commit 6e66b80.

This patch passes all tests.
This patch merges cleanly.
This patch adds no public classes.

SparkQA · 2017-05-17T02:29:26Z

Test build #76990 has finished for PR 18008 at commit 1c62909.

This patch passes all tests.
This patch merges cleanly.
This patch adds no public classes.

SparkQA · 2017-05-17T02:44:10Z

Test build #76991 has finished for PR 18008 at commit feda785.

This patch passes all tests.
This patch merges cleanly.
This patch adds no public classes.

cloud-fan · 2017-05-17T05:05:05Z

thanks, merging to master/2.2!

…kMetrics construction ## What changes were proposed in this pull request? In ``` ./bin/spark-shell --master=local[64] ``` I ran ``` sc.parallelize(1 to 100000, 100000).count() ``` and profiled the time spend in the LiveListenerBus event processing thread. I discovered that the majority of the time was being spent in `TaskMetrics.empty` calls in `JobProgressListener.onTaskStart`. It turns out that we can slightly refactor to remove the need to construct one empty instance per call, greatly improving the performance of this code. The performance gains here help to avoid an issue where listener events would be dropped because the JobProgressListener couldn't keep up with the throughput. **Before:** ![image](https://cloud.githubusercontent.com/assets/50748/26133095/95bcd42a-3a59-11e7-8051-a50550e447b8.png) **After:** ![image](https://cloud.githubusercontent.com/assets/50748/26133070/7935e148-3a59-11e7-8c2d-73d5aa5a2397.png) ## How was this patch tested? Benchmarks described above. Author: Josh Rosen <[email protected]> Closes #18008 from JoshRosen/nametoaccums-improvements. (cherry picked from commit 30e0557) Signed-off-by: Wenchen Fan <[email protected]>

witgo · 2017-05-17T06:21:44Z

@JoshRosen , what's the tool in your screenshot?

JoshRosen · 2017-05-17T06:24:26Z

@witgo, I'm using YourKit Java Profiler 2016.02. In these screenshots I enabled CPU sampling then took a performance snapshot and used the per-thread view, focusing on the time taken in the live listener bus thread by right-clicking on the subtree and choosing "focus subtree" from the context menu.

witgo · 2017-05-17T06:35:05Z

@JoshRosen I see, Thank you.

…kMetrics construction ## What changes were proposed in this pull request? In ``` ./bin/spark-shell --master=local[64] ``` I ran ``` sc.parallelize(1 to 100000, 100000).count() ``` and profiled the time spend in the LiveListenerBus event processing thread. I discovered that the majority of the time was being spent in `TaskMetrics.empty` calls in `JobProgressListener.onTaskStart`. It turns out that we can slightly refactor to remove the need to construct one empty instance per call, greatly improving the performance of this code. The performance gains here help to avoid an issue where listener events would be dropped because the JobProgressListener couldn't keep up with the throughput. **Before:** ![image](https://cloud.githubusercontent.com/assets/50748/26133095/95bcd42a-3a59-11e7-8051-a50550e447b8.png) **After:** ![image](https://cloud.githubusercontent.com/assets/50748/26133070/7935e148-3a59-11e7-8c2d-73d5aa5a2397.png) ## How was this patch tested? Benchmarks described above. Author: Josh Rosen <[email protected]> Closes apache#18008 from JoshRosen/nametoaccums-improvements.

JoshRosen added 2 commits May 16, 2017 12:44

TaskMetrics nameToAccums improvements.

622951f

Add comment.

4675b21

JoshRosen commented May 16, 2017

View reviewed changes

JoshRosen added 2 commits May 16, 2017 16:12

Revert "Add comment."

782b772

This reverts commit 4675b21.

Revert "TaskMetrics nameToAccums improvements."

214559e

This reverts commit 622951f.

JoshRosen changed the title ~~[SPARK-20776] Fix perf. problems in TaskMetrics.nameToAccums map initialization~~ [SPARK-20776] Fix perf. problems in JobProgressListener caused by TaskMetrics construction May 16, 2017

JoshRosen added 2 commits May 16, 2017 16:20

Don't construct empty TaskMetrics.

6e66b80

Initialize TaskUIData with empty TaskMetricsUIData.

1c62909

JoshRosen commented May 16, 2017

View reviewed changes

Roll back unnecessary change to AccumulatorV2

feda785

cloud-fan reviewed May 17, 2017

View reviewed changes

JoshRosen mentioned this pull request May 17, 2017

[SPARK-20715] Store MapStatuses only in MapOutputTracker, not ShuffleMapStage #17955

Closed

asfgit closed this in 30e0557 May 17, 2017

JoshRosen deleted the nametoaccums-improvements branch May 17, 2017 05:07



		import InternalAccumulator._
		@transient private[spark] lazy val nameToAccums = LinkedHashMap(

[SPARK-20776] Fix perf. problems in JobProgressListener caused by TaskMetrics construction #18008

[SPARK-20776] Fix perf. problems in JobProgressListener caused by TaskMetrics construction #18008

Uh oh!

Conversation

JoshRosen commented May 16, 2017 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

What changes were proposed in this pull request?

How was this patch tested?

Uh oh!

JoshRosen May 16, 2017 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

JoshRosen May 16, 2017

Choose a reason for hiding this comment

Uh oh!

JoshRosen commented May 16, 2017

Uh oh!

JoshRosen May 16, 2017

Choose a reason for hiding this comment

Uh oh!

cloud-fan May 17, 2017

Choose a reason for hiding this comment

Uh oh!

JoshRosen May 17, 2017

Choose a reason for hiding this comment

Uh oh!

cloud-fan commented May 17, 2017

Uh oh!

SparkQA commented May 17, 2017

Uh oh!

SparkQA commented May 17, 2017

Uh oh!

SparkQA commented May 17, 2017

Uh oh!

SparkQA commented May 17, 2017

Uh oh!

cloud-fan commented May 17, 2017

Uh oh!

witgo commented May 17, 2017

Uh oh!

JoshRosen commented May 17, 2017

Uh oh!

witgo commented May 17, 2017

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

4 participants

JoshRosen commented May 16, 2017 •

edited

Loading

JoshRosen May 16, 2017 •

edited

Loading