You are viewing a plain text version of this content. The canonical link for it is here.
Posted to reviews@spark.apache.org by kayousterhout <gi...@git.apache.org> on 2014/07/18 04:54:00 UTC
[GitHub] spark pull request: [SPARK-2571] Correctly report shuffle read met...
GitHub user kayousterhout opened a pull request:
https://github.com/apache/spark/pull/1476
[SPARK-2571] Correctly report shuffle read metrics.
Currently, shuffle read metrics are incorrectly reported when stages have multiple shuffle dependencies (they are set to be the metrics from just one of the shuffle dependencies, rather than the accumulated metrics from all of the shuffle dependencies). This fixes that problem, and should probably be back-ported to the 0.9 branch.
Thanks @ryanra for discovering this problem!
cc @rxin @andrewor14
You can merge this pull request into a Git repository by running:
$ git pull https://github.com/kayousterhout/spark-1 join_bug
Alternatively you can review and apply these changes as the patch at:
https://github.com/apache/spark/pull/1476.patch
To close this pull request, make a commit to your master/trunk branch
with (at least) the following in the commit message:
This closes #1476
----
commit 25b7523bca463fd58cc944908111f1a2204c589b
Author: Kay Ousterhout <ka...@gmail.com>
Date: 2014-07-18T02:46:39Z
[SPARK-2571] Correctly report shuffle read metrics.
----
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] spark pull request: [SPARK-2571] Correctly report shuffle read met...
Posted by andrewor14 <gi...@git.apache.org>.
Github user andrewor14 commented on a diff in the pull request:
https://github.com/apache/spark/pull/1476#discussion_r15124263
--- Diff: core/src/main/scala/org/apache/spark/executor/TaskMetrics.scala ---
@@ -75,7 +75,9 @@ class TaskMetrics extends Serializable {
/**
* If this task reads from shuffle output, metrics on getting shuffle data will be collected here
*/
- var shuffleReadMetrics: Option[ShuffleReadMetrics] = None
+ private var _shuffleReadMetrics: Option[ShuffleReadMetrics] = None
+
+ def shuffleReadMetrics = _shuffleReadMetrics
--- End diff --
Yes, I recognize that, but even with a private setter we can still go ahead and change the fields if we wanted to, so it's not really immutable. Not a huge deal.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] spark pull request: [SPARK-2571] Correctly report shuffle read met...
Posted by kayousterhout <gi...@git.apache.org>.
Github user kayousterhout commented on a diff in the pull request:
https://github.com/apache/spark/pull/1476#discussion_r15133729
--- Diff: core/src/main/scala/org/apache/spark/util/JsonProtocol.scala ---
@@ -527,8 +527,9 @@ private[spark] object JsonProtocol {
metrics.resultSerializationTime = (json \ "Result Serialization Time").extract[Long]
metrics.memoryBytesSpilled = (json \ "Memory Bytes Spilled").extract[Long]
metrics.diskBytesSpilled = (json \ "Disk Bytes Spilled").extract[Long]
- metrics.shuffleReadMetrics =
- Utils.jsonOption(json \ "Shuffle Read Metrics").map(shuffleReadMetricsFromJson)
+ Utils.jsonOption(json \ "Shuffle Read Metrics").map { shuffleReadMetrics =>
+ metrics.updateShuffleReadMetrics(shuffleReadMetricsFromJson(shuffleReadMetrics))
--- End diff --
What do you mean here? updateShuffleReadMetrics is the only way to set it right?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] spark pull request: [SPARK-2571] Correctly report shuffle read met...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the pull request:
https://github.com/apache/spark/pull/1476#issuecomment-49394559
QA results for PR 1476:<br>- This patch FAILED unit tests.<br>- This patch merges cleanly<br>- This patch adds no public classes<br><br>For more information see test ouptut:<br>https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/16805/consoleFull
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] spark pull request: [SPARK-2571] Correctly report shuffle read met...
Posted by kayousterhout <gi...@git.apache.org>.
Github user kayousterhout commented on a diff in the pull request:
https://github.com/apache/spark/pull/1476#discussion_r15098545
--- Diff: core/src/test/scala/org/apache/spark/ui/jobs/JobProgressListenerSuite.scala ---
@@ -81,8 +81,6 @@ class JobProgressListenerSuite extends FunSuite with LocalSparkContext with Matc
assert(listener.stageIdToData.size === 1)
// finish this task, should get updated duration
- shuffleReadMetrics.remoteBytesRead = 1000
- taskMetrics.shuffleReadMetrics = Some(shuffleReadMetrics)
--- End diff --
It expects 2000 bytes read because the listener is adding up the total across all tasks in the stage -- so 1000 for the earlier task + 1000 for this task. The lines that I deleted were redundant (remoteBytesRead was already set to 1000 and shuffleReadMetrics had already been set as taskMetrics.shuffleReadMetrics).
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] spark pull request: [SPARK-2571] Correctly report shuffle read met...
Posted by aarondav <gi...@git.apache.org>.
Github user aarondav commented on a diff in the pull request:
https://github.com/apache/spark/pull/1476#discussion_r15097975
--- Diff: core/src/test/scala/org/apache/spark/ui/jobs/JobProgressListenerSuite.scala ---
@@ -81,8 +81,6 @@ class JobProgressListenerSuite extends FunSuite with LocalSparkContext with Matc
assert(listener.stageIdToData.size === 1)
// finish this task, should get updated duration
- shuffleReadMetrics.remoteBytesRead = 1000
- taskMetrics.shuffleReadMetrics = Some(shuffleReadMetrics)
--- End diff --
how does this pass unit tests without calling updateShuffleReadMetrics again? It seems to expect 2000 bytes read
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] spark pull request: [SPARK-2571] Correctly report shuffle read met...
Posted by kayousterhout <gi...@git.apache.org>.
Github user kayousterhout commented on the pull request:
https://github.com/apache/spark/pull/1476#issuecomment-49482886
Thanks for the review @andrewor14 and @aarondav ! Merged into master (I decided not to back port -- I figured since no one has noticed this bug it's not worth back porting -- but happy to go ahead and do that if someone thinks it's worthwhile).
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] spark pull request: [SPARK-2571] Correctly report shuffle read met...
Posted by andrewor14 <gi...@git.apache.org>.
Github user andrewor14 commented on the pull request:
https://github.com/apache/spark/pull/1476#issuecomment-49458944
Couple minor comments, but the changes look good.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] spark pull request: [SPARK-2571] Correctly report shuffle read met...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the pull request:
https://github.com/apache/spark/pull/1476#issuecomment-49390789
QA tests have started for PR 1476. This patch merges cleanly. <br>View progress: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/16805/consoleFull
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] spark pull request: [SPARK-2571] Correctly report shuffle read met...
Posted by kayousterhout <gi...@git.apache.org>.
Github user kayousterhout commented on a diff in the pull request:
https://github.com/apache/spark/pull/1476#discussion_r15133407
--- Diff: core/src/main/scala/org/apache/spark/executor/TaskMetrics.scala ---
@@ -75,7 +75,9 @@ class TaskMetrics extends Serializable {
/**
* If this task reads from shuffle output, metrics on getting shuffle data will be collected here
*/
- var shuffleReadMetrics: Option[ShuffleReadMetrics] = None
+ private var _shuffleReadMetrics: Option[ShuffleReadMetrics] = None
+
+ def shuffleReadMetrics = _shuffleReadMetrics
--- End diff --
I was trying to force people to use updateShuffleReadMetrics() rather than directly updating shuffleReadMetrics to avoid accidentally overwriting existing values...happy to do this in a different way though if you think something else would be better / more readable.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] spark pull request: [SPARK-2571] Correctly report shuffle read met...
Posted by kayousterhout <gi...@git.apache.org>.
Github user kayousterhout commented on a diff in the pull request:
https://github.com/apache/spark/pull/1476#discussion_r15119268
--- Diff: core/src/test/scala/org/apache/spark/ui/jobs/JobProgressListenerSuite.scala ---
@@ -81,8 +81,6 @@ class JobProgressListenerSuite extends FunSuite with LocalSparkContext with Matc
assert(listener.stageIdToData.size === 1)
// finish this task, should get updated duration
- shuffleReadMetrics.remoteBytesRead = 1000
- taskMetrics.shuffleReadMetrics = Some(shuffleReadMetrics)
--- End diff --
The test I changed in SparkListenerSuite tests this problem (since before the test only checked that the value is > 0, and I changed it to test for the correct value -- the updated test fails with the old code). Were you thinking to break this issue out into its own test?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] spark pull request: [SPARK-2571] Correctly report shuffle read met...
Posted by andrewor14 <gi...@git.apache.org>.
Github user andrewor14 commented on a diff in the pull request:
https://github.com/apache/spark/pull/1476#discussion_r15123564
--- Diff: core/src/main/scala/org/apache/spark/executor/TaskMetrics.scala ---
@@ -75,7 +75,9 @@ class TaskMetrics extends Serializable {
/**
* If this task reads from shuffle output, metrics on getting shuffle data will be collected here
*/
- var shuffleReadMetrics: Option[ShuffleReadMetrics] = None
+ private var _shuffleReadMetrics: Option[ShuffleReadMetrics] = None
+
+ def shuffleReadMetrics = _shuffleReadMetrics
--- End diff --
Do we need this? shuffleReadMetrics is a already var so people could just modify it directly anyway
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] spark pull request: [SPARK-2571] Correctly report shuffle read met...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the pull request:
https://github.com/apache/spark/pull/1476#issuecomment-49397551
QA tests have started for PR 1476. This patch merges cleanly. <br>View progress: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/16811/consoleFull
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] spark pull request: [SPARK-2571] Correctly report shuffle read met...
Posted by aarondav <gi...@git.apache.org>.
Github user aarondav commented on a diff in the pull request:
https://github.com/apache/spark/pull/1476#discussion_r15123953
--- Diff: core/src/main/scala/org/apache/spark/executor/TaskMetrics.scala ---
@@ -75,7 +75,9 @@ class TaskMetrics extends Serializable {
/**
* If this task reads from shuffle output, metrics on getting shuffle data will be collected here
*/
- var shuffleReadMetrics: Option[ShuffleReadMetrics] = None
+ private var _shuffleReadMetrics: Option[ShuffleReadMetrics] = None
+
+ def shuffleReadMetrics = _shuffleReadMetrics
--- End diff --
This is effectively a private-setter, public-getter pattern.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] spark pull request: [SPARK-2571] Correctly report shuffle read met...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the pull request:
https://github.com/apache/spark/pull/1476#issuecomment-49402820
QA results for PR 1476:<br>- This patch PASSES unit tests.<br>- This patch merges cleanly<br>- This patch adds no public classes<br><br>For more information see test ouptut:<br>https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/16811/consoleFull
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] spark pull request: [SPARK-2571] Correctly report shuffle read met...
Posted by aarondav <gi...@git.apache.org>.
Github user aarondav commented on a diff in the pull request:
https://github.com/apache/spark/pull/1476#discussion_r15121395
--- Diff: core/src/test/scala/org/apache/spark/ui/jobs/JobProgressListenerSuite.scala ---
@@ -81,8 +81,6 @@ class JobProgressListenerSuite extends FunSuite with LocalSparkContext with Matc
assert(listener.stageIdToData.size === 1)
// finish this task, should get updated duration
- shuffleReadMetrics.remoteBytesRead = 1000
- taskMetrics.shuffleReadMetrics = Some(shuffleReadMetrics)
--- End diff --
Gotcha, no, that's fine. Funny that we had the test already but with relaxed constraints.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] spark pull request: [SPARK-2571] Correctly report shuffle read met...
Posted by aarondav <gi...@git.apache.org>.
Github user aarondav commented on a diff in the pull request:
https://github.com/apache/spark/pull/1476#discussion_r15099340
--- Diff: core/src/test/scala/org/apache/spark/ui/jobs/JobProgressListenerSuite.scala ---
@@ -81,8 +81,6 @@ class JobProgressListenerSuite extends FunSuite with LocalSparkContext with Matc
assert(listener.stageIdToData.size === 1)
// finish this task, should get updated duration
- shuffleReadMetrics.remoteBytesRead = 1000
- taskMetrics.shuffleReadMetrics = Some(shuffleReadMetrics)
--- End diff --
Ahh, gotcha, apologies. Any chance we could get a new unit test for the particular problem in this PR? The changes you added do look correct and verifiable from just reading them, so I'd be fine with this being merged without, but doesn't hurt to ask :)
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] spark pull request: [SPARK-2571] Correctly report shuffle read met...
Posted by andrewor14 <gi...@git.apache.org>.
Github user andrewor14 commented on a diff in the pull request:
https://github.com/apache/spark/pull/1476#discussion_r15123302
--- Diff: core/src/main/scala/org/apache/spark/util/JsonProtocol.scala ---
@@ -527,8 +527,9 @@ private[spark] object JsonProtocol {
metrics.resultSerializationTime = (json \ "Result Serialization Time").extract[Long]
metrics.memoryBytesSpilled = (json \ "Memory Bytes Spilled").extract[Long]
metrics.diskBytesSpilled = (json \ "Disk Bytes Spilled").extract[Long]
- metrics.shuffleReadMetrics =
- Utils.jsonOption(json \ "Shuffle Read Metrics").map(shuffleReadMetricsFromJson)
+ Utils.jsonOption(json \ "Shuffle Read Metrics").map { shuffleReadMetrics =>
+ metrics.updateShuffleReadMetrics(shuffleReadMetricsFromJson(shuffleReadMetrics))
--- End diff --
Not sure if you need to do this here, since we're creating a TaskMetrics from scratch anyway
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] spark pull request: [SPARK-2571] Correctly report shuffle read met...
Posted by andrewor14 <gi...@git.apache.org>.
Github user andrewor14 commented on a diff in the pull request:
https://github.com/apache/spark/pull/1476#discussion_r15133588
--- Diff: core/src/main/scala/org/apache/spark/executor/TaskMetrics.scala ---
@@ -75,7 +75,9 @@ class TaskMetrics extends Serializable {
/**
* If this task reads from shuffle output, metrics on getting shuffle data will be collected here
*/
- var shuffleReadMetrics: Option[ShuffleReadMetrics] = None
+ private var _shuffleReadMetrics: Option[ShuffleReadMetrics] = None
+
+ def shuffleReadMetrics = _shuffleReadMetrics
--- End diff --
That's fine
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] spark pull request: [SPARK-2571] Correctly report shuffle read met...
Posted by asfgit <gi...@git.apache.org>.
Github user asfgit closed the pull request at:
https://github.com/apache/spark/pull/1476
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---