[SPARK-23989][SQL] exchange should copy data before non-serialized shuffle #21101
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
|
Test build #89513 has finished for PR 21101 at commit
|
JoshRosen
reviewed
Apr 19, 2018
| // If we're using the original SortShuffleManager and the number of output partitions is | ||
| // sufficiently small, then Spark will fall back to the hash-based shuffle write path, which | ||
| // doesn't buffer deserialized records. | ||
| // Note that we'll have to remove this case if we fix SPARK-6026 and remove this bypass. | ||
| false | ||
| } else if (serializer.supportsRelocationOfSerializedObjects) { | ||
| } else if (numParts <= SortShuffleManager.MAX_SHUFFLE_OUTPUT_PARTITIONS_FOR_SERIALIZED_MODE) { |
There was a problem hiding this comment.
I was almost going to suggest that we should we check for both conditions with an && here just as future-proofing in case serializer was changed, but I can now see why that isn't a huge risk in the current codebase: we always use an UnsafeRowSerializer here now. It was only in the pre-Tungsten era that we could use either UnsafeRowSerializer or SparkSqlSerializer here.
|
Merging to master and 2.3. Let me know if more further backports are needed. |
asfgit
pushed a commit
that referenced
this pull request
Apr 19, 2018
…uffle ## What changes were proposed in this pull request? In Spark SQL, we usually reuse the `UnsafeRow` instance and need to copy the data when a place buffers non-serialized objects. Shuffle may buffer objects if we don't make it to the bypass merge shuffle or unsafe shuffle. `ShuffleExchangeExec.needToCopyObjectsBeforeShuffle` misses the case that, if `spark.sql.shuffle.partitions` is large enough, we could fail to run unsafe shuffle and go with the non-serialized shuffle. This bug is very hard to hit since users wouldn't set such a large number of partitions(16 million) for Spark SQL exchange. TODO: test ## How was this patch tested? todo. Author: Wenchen Fan <wenchen@databricks.com> Closes #21101 from cloud-fan/shuffle. (cherry picked from commit 6e19f76) Signed-off-by: Herman van Hovell <hvanhovell@databricks.com>
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?
In Spark SQL, we usually reuse the
UnsafeRowinstance and need to copy the data when a place buffers non-serialized objects.Shuffle may buffer objects if we don't make it to the bypass merge shuffle or unsafe shuffle.
ShuffleExchangeExec.needToCopyObjectsBeforeShufflemisses the case that, ifspark.sql.shuffle.partitionsis large enough, we could fail to run unsafe shuffle and go with the non-serialized shuffle.This bug is very hard to hit since users wouldn't set such a large number of partitions(16 million) for Spark SQL exchange.
TODO: test
How was this patch tested?
todo.