[SPARK-25299] Use the shuffle writer plugin for the SortShuffleWriter. #532
Conversation
This file contains hidden or 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
Because compressed output streams don't like it.
ifilonenko
reviewed
Apr 9, 2019
| private def writeEmptyPartition(mapOutputWriter: ShuffleMapOutputWriter): Unit = { | ||
| var partitionWriter: ShufflePartitionWriter = null | ||
| try { | ||
| partitionWriter = mapOutputWriter.getNextPartitionWriter |
There was a problem hiding this comment.
Would you need to do partitonWriter.toStream(), as in UnsafeShuffleWriter, to ensure that the outputFileStream is created and an empty file exists. It seems be expected by the UnsafeShuffleWriterSuite, idk if it is the same here
Author
There was a problem hiding this comment.
That shouldn't be necessary, the writer.close() should properly know what to do if a stream was never created.
yifeih
reviewed
Apr 11, 2019
| // The iterator may have stopped short of opening a writer for every partition. So fill in the | ||
| // remaining empty partitions. | ||
| for (emptyPartition <- nextPartitionId until numPartitions) { | ||
| writeEmptyPartition(mapOutputWriter) |
There was a problem hiding this comment.
Hmm wait why do we need this? Shouldn't new long[numPartitions] fill the array with default value of 0?
Author
There was a problem hiding this comment.
It depends on the contract we want to present to other plugin writers. I.e. do we make a contract that we open a writer for strictly every partition, even empty ones? Or do we say we open for the first N partitions where N is the last non-empty partition? My take is that we should have the contract that we always open a writer for every partition, empty or not, from 0 through numPartitions - 1. But, again, this shows the limitation of presenting an API that doesn't include the partition identifier explicitly when getting partition writers.
There was a problem hiding this comment.
Ohh i see hmm yea I'm ok keeping it like this then. It does show more consistency for plugin implementers
core/src/main/scala/org/apache/spark/util/collection/ShufflePartitionPairsWriter.scala
Outdated
Show resolved
Hide resolved
|
looks good to me! |
Author
|
Ok I'm going to merge this - we have some incoming suggestions for the shuffle writer API but we should get the implementations with the existing APIs merged first. Thanks everyone who reviewed! |
svc-spark-25299
left a comment
svc-spark-25299
left a comment
There was a problem hiding this comment.
================================================================================================
BlockStoreShuffleReader reader
================================================================================================
Java HotSpot(TM) 64-Bit Server VM 1.8.0_131-b11 on Linux 4.15.0-1014-gcp
Intel(R) Xeon(R) CPU @ 2.30GHz
no aggregation or sorting: Best Time(ms) Avg Time(ms) Stdev(ms) Rate(M/s) Per Row(ns) Relative
------------------------------------------------------------------------------------------------------------------------
local fetch 11006 11081 123 0.9 1100.6 1.0X
remote rpc fetch 10901 10977 48 0.9 1090.1 1.0X
Java HotSpot(TM) 64-Bit Server VM 1.8.0_131-b11 on Linux 4.15.0-1014-gcp
Intel(R) Xeon(R) CPU @ 2.30GHz
with aggregation: Best Time(ms) Avg Time(ms) Stdev(ms) Rate(M/s) Per Row(ns) Relative
------------------------------------------------------------------------------------------------------------------------
local fetch 28171 28412 217 0.1 14085.4 1.0X
remote rpc fetch 28239 28689 249 0.1 14119.5 1.0X
Java HotSpot(TM) 64-Bit Server VM 1.8.0_131-b11 on Linux 4.15.0-1014-gcp
Intel(R) Xeon(R) CPU @ 2.30GHz
with sorting: Best Time(ms) Avg Time(ms) Stdev(ms) Rate(M/s) Per Row(ns) Relative
------------------------------------------------------------------------------------------------------------------------
local fetch 30290 30609 280 0.1 15145.2 1.0X
remote rpc fetch 30122 30697 369 0.1 15060.9 1.0X
Java HotSpot(TM) 64-Bit Server VM 1.8.0_131-b11 on Linux 4.15.0-1014-gcp
Intel(R) Xeon(R) CPU @ 2.30GHz
with seek: Best Time(ms) Avg Time(ms) Stdev(ms) Rate(M/s) Per Row(ns) Relative
------------------------------------------------------------------------------------------------------------------------
seek to last record 1 1 1 2797.3 0.4 1.0X
================================================================================================
BypassMergeSortShuffleWriter write
================================================================================================
Java HotSpot(TM) 64-Bit Server VM 1.8.0_131-b11 on Linux 4.15.0-1014-gcp
Intel(R) Xeon(R) CPU @ 2.30GHz
BypassMergeSortShuffleWrite without spill: Best Time(ms) Avg Time(ms) Stdev(ms) Rate(M/s) Per Row(ns) Relative
------------------------------------------------------------------------------------------------------------------------
small dataset without disk spill 2 4 2 0.5 2030.5 1.0X
Java HotSpot(TM) 64-Bit Server VM 1.8.0_131-b11 on Linux 4.15.0-1014-gcp
Intel(R) Xeon(R) CPU @ 2.30GHz
BypassMergeSortShuffleWrite with spill: Best Time(ms) Avg Time(ms) Stdev(ms) Rate(M/s) Per Row(ns) Relative
------------------------------------------------------------------------------------------------------------------------
without transferTo 7244 7359 105 0.9 1079.4 1.0X
with transferTo 7252 7311 64 0.9 1080.6 1.0X
================================================================================================
SortShuffleWriter writer
================================================================================================
Java HotSpot(TM) 64-Bit Server VM 1.8.0_131-b11 on Linux 4.15.0-1014-gcp
Intel(R) Xeon(R) CPU @ 2.30GHz
SortShuffleWriter without spills: Best Time(ms) Avg Time(ms) Stdev(ms) Rate(M/s) Per Row(ns) Relative
------------------------------------------------------------------------------------------------------------------------
small dataset without spills 11 17 5 0.1 10518.0 1.0X
Java HotSpot(TM) 64-Bit Server VM 1.8.0_131-b11 on Linux 4.15.0-1014-gcp
Intel(R) Xeon(R) CPU @ 2.30GHz
SortShuffleWriter with spills: Best Time(ms) Avg Time(ms) Stdev(ms) Rate(M/s) Per Row(ns) Relative
------------------------------------------------------------------------------------------------------------------------
no map side combine 14347 14510 129 0.5 2137.9 1.0X
with map side aggregation 14200 14277 59 0.5 2115.9 1.0X
with map side sort 14162 14267 82 0.5 2110.3 1.0X
================================================================================================
UnsafeShuffleWriter write
================================================================================================
Java HotSpot(TM) 64-Bit Server VM 1.8.0_131-b11 on Linux 4.15.0-1014-gcp
Intel(R) Xeon(R) CPU @ 2.30GHz
UnsafeShuffleWriter without spills: Best Time(ms) Avg Time(ms) Stdev(ms) Rate(M/s) Per Row(ns) Relative
------------------------------------------------------------------------------------------------------------------------
small dataset without spills 20 24 5 0.1 19894.4 1.0X
Java HotSpot(TM) 64-Bit Server VM 1.8.0_131-b11 on Linux 4.15.0-1014-gcp
Intel(R) Xeon(R) CPU @ 2.30GHz
UnsafeShuffleWriter with spills: Best Time(ms) Avg Time(ms) Stdev(ms) Rate(M/s) Per Row(ns) Relative
------------------------------------------------------------------------------------------------------------------------
without transferTo 15869 15965 81 0.8 1182.3 1.0X
with transferTo 15893 16023 144 0.8 1184.1 1.0X
mccheah
added a commit
that referenced
this pull request
Jun 27, 2019
#532) * [SPARK-25299] Use the shuffle writer plugin for the SortShuffleWriter. * Remove unused * Handle empty partitions properly. * Adjust formatting * Don't close streams twice. Because compressed output streams don't like it. * Clarify comment
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
4 participants
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.
No description provided.