[SPARK-14804][Spark][Graphx] Fix checkpointing of VertexRDD/EdgeRDD #15396
Conversation
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
|
I'm no expert on checkpointing, but the tests look fine to me. You could eliminated duplicated code in the tests by putting the shared code in a helper method. Tested locally to confirm unit tests fail before the fix. |
|
Test build #66538 has finished for PR 15396 at commit
|
|
Can you reference the earlier pull requests here too? |
andrewor14
reviewed
Oct 12, 2016
| @@ -1589,7 +1589,8 @@ abstract class RDD[T: ClassTag]( | |||
| * This is introduced as an alias for `isCheckpointed` to clarify the semantics of the | |||
There was a problem hiding this comment.
need to fix the comment here
There was a problem hiding this comment.
oh wait I just realized this is exactly the same as isCheckpointed. It's kind of confusing why we have two methods that do the same thing. Is it because VertexRDD or something overrides one but not the other?
There was a problem hiding this comment.
@andrewor14 This method left as is in this PR - #15447
andrewor14
reviewed
Oct 12, 2016
| @@ -1589,7 +1589,8 @@ abstract class RDD[T: ClassTag]( | |||
| * This is introduced as an alias for `isCheckpointed` to clarify the semantics of the | |||
| * return value. Exposed for testing. | |||
| */ | |||
| private[spark] def isCheckpointedAndMaterialized: Boolean = isCheckpointed | |||
| private[spark] def isCheckpointedAndMaterialized: Boolean = | |||
| checkpointData.exists(_.isCheckpointed) | |||
There was a problem hiding this comment.
One way to fix the duplicate code is to make this one final and have isCheckpointed call this one, then implementations of RDD can feel free to override isCheckpointed without affecting our internal checkpointing logic.
There was a problem hiding this comment.
what do you mean. if isCheckpointedAndMaterialized = isCheckpointed, and you override isCheckpointed, then that would change the behavior of isCheckpointedAndMaterialized as well.
There was a problem hiding this comment.
No, I meant:
private[spark] final def isCheckpointedAndMaterialized: Boolean = {
checkpointData.exists(_.isCheckpointed)
}
def isCheckpointed: Boolean = isCheckpointedAndMaterialized
There was a problem hiding this comment.
Addressed it. Can you check again?
There was a problem hiding this comment.
looks good, but you forgot to make this final. No big deal
|
Looks good. I left a suggestion that I think will make the code cleaner. |
|
Any update on this? Would love to see it in an upcoming release. |
|
Test build #72002 has finished for PR 15396 at commit
|
|
I am merging this to master and 2.1 |
asfgit
pushed a commit
that referenced
this pull request
Jan 26, 2017
## What changes were proposed in this pull request? EdgeRDD/VertexRDD overrides checkpoint() and isCheckpointed() to forward these to the internal partitionRDD. So when checkpoint() is called on them, its the partitionRDD that actually gets checkpointed. However since isCheckpointed() also overridden to call partitionRDD.isCheckpointed, EdgeRDD/VertexRDD.isCheckpointed returns true even though this RDD is actually not checkpointed. This would have been fine except the RDD's internal logic for computing the RDD depends on isCheckpointed(). So for VertexRDD/EdgeRDD, since isCheckpointed is true, when computing Spark tries to read checkpoint data of VertexRDD/EdgeRDD even though they are not actually checkpointed. Through a crazy sequence of call forwarding, it reads checkpoint data of partitionsRDD and tries to cast it to types in Vertex/EdgeRDD. This leads to ClassCastException. The minimal fix that does not change any public behavior is to modify RDD internal to not use public override-able API for internal logic. ## How was this patch tested? New unit tests. Author: Tathagata Das <[email protected]> Closes #15396 from tdas/SPARK-14804. (cherry picked from commit 47d5d0d) Signed-off-by: Tathagata Das <[email protected]>
asfgit
pushed a commit
that referenced
this pull request
Jan 26, 2017
## What changes were proposed in this pull request? EdgeRDD/VertexRDD overrides checkpoint() and isCheckpointed() to forward these to the internal partitionRDD. So when checkpoint() is called on them, its the partitionRDD that actually gets checkpointed. However since isCheckpointed() also overridden to call partitionRDD.isCheckpointed, EdgeRDD/VertexRDD.isCheckpointed returns true even though this RDD is actually not checkpointed. This would have been fine except the RDD's internal logic for computing the RDD depends on isCheckpointed(). So for VertexRDD/EdgeRDD, since isCheckpointed is true, when computing Spark tries to read checkpoint data of VertexRDD/EdgeRDD even though they are not actually checkpointed. Through a crazy sequence of call forwarding, it reads checkpoint data of partitionsRDD and tries to cast it to types in Vertex/EdgeRDD. This leads to ClassCastException. The minimal fix that does not change any public behavior is to modify RDD internal to not use public override-able API for internal logic. ## How was this patch tested? New unit tests. Author: Tathagata Das <[email protected]> Closes #15396 from tdas/SPARK-14804. (cherry picked from commit 47d5d0d) Signed-off-by: Tathagata Das <[email protected]>
uzadude
pushed a commit
to uzadude/spark
that referenced
this pull request
Jan 27, 2017
## What changes were proposed in this pull request? EdgeRDD/VertexRDD overrides checkpoint() and isCheckpointed() to forward these to the internal partitionRDD. So when checkpoint() is called on them, its the partitionRDD that actually gets checkpointed. However since isCheckpointed() also overridden to call partitionRDD.isCheckpointed, EdgeRDD/VertexRDD.isCheckpointed returns true even though this RDD is actually not checkpointed. This would have been fine except the RDD's internal logic for computing the RDD depends on isCheckpointed(). So for VertexRDD/EdgeRDD, since isCheckpointed is true, when computing Spark tries to read checkpoint data of VertexRDD/EdgeRDD even though they are not actually checkpointed. Through a crazy sequence of call forwarding, it reads checkpoint data of partitionsRDD and tries to cast it to types in Vertex/EdgeRDD. This leads to ClassCastException. The minimal fix that does not change any public behavior is to modify RDD internal to not use public override-able API for internal logic. ## How was this patch tested? New unit tests. Author: Tathagata Das <[email protected]> Closes apache#15396 from tdas/SPARK-14804.
cmonkey
pushed a commit
to cmonkey/spark
that referenced
this pull request
Feb 15, 2017
## What changes were proposed in this pull request? EdgeRDD/VertexRDD overrides checkpoint() and isCheckpointed() to forward these to the internal partitionRDD. So when checkpoint() is called on them, its the partitionRDD that actually gets checkpointed. However since isCheckpointed() also overridden to call partitionRDD.isCheckpointed, EdgeRDD/VertexRDD.isCheckpointed returns true even though this RDD is actually not checkpointed. This would have been fine except the RDD's internal logic for computing the RDD depends on isCheckpointed(). So for VertexRDD/EdgeRDD, since isCheckpointed is true, when computing Spark tries to read checkpoint data of VertexRDD/EdgeRDD even though they are not actually checkpointed. Through a crazy sequence of call forwarding, it reads checkpoint data of partitionsRDD and tries to cast it to types in Vertex/EdgeRDD. This leads to ClassCastException. The minimal fix that does not change any public behavior is to modify RDD internal to not use public override-able API for internal logic. ## How was this patch tested? New unit tests. Author: Tathagata Das <[email protected]> Closes apache#15396 from tdas/SPARK-14804.
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
What changes were proposed in this pull request?
EdgeRDD/VertexRDD overrides checkpoint() and isCheckpointed() to forward these to the internal partitionRDD. So when checkpoint() is called on them, its the partitionRDD that actually gets checkpointed. However since isCheckpointed() also overridden to call partitionRDD.isCheckpointed, EdgeRDD/VertexRDD.isCheckpointed returns true even though this RDD is actually not checkpointed.
This would have been fine except the RDD's internal logic for computing the RDD depends on isCheckpointed(). So for VertexRDD/EdgeRDD, since isCheckpointed is true, when computing Spark tries to read checkpoint data of VertexRDD/EdgeRDD even though they are not actually checkpointed. Through a crazy sequence of call forwarding, it reads checkpoint data of partitionsRDD and tries to cast it to types in Vertex/EdgeRDD. This leads to ClassCastException.
The minimal fix that does not change any public behavior is to modify RDD internal to not use public override-able API for internal logic.
How was this patch tested?
New unit tests.