Readability improvements to SortShuffleReader

[sryza]

Some of the changes include:

• Changed ioSortFactor to maxMergeWidth to clarify what it means.
• Changed "partition" to "block" to avoid confusion with Spark's more typical use of the word partition.
• Added header documentation.
• Rearranged and documented the main merge loop.

[jerryshao]

Hi sandy, Seems you use blocksAwaitingMerge.size() to replace numSplits to calculate the merging number, I'm not the semantics is still the same. Since our merging thread is running parallel with fetch thread, so we the size of blocksAwaitingMerge is still increasing, is not a constant value. I'm not sure using this value to calculate the merging number is still the same as before.

[sryza]

Oops, you're definitely right. Will fix this.

[jerryshao]

Hi Sandy, thanks a lot for your work. I need to better understand your patch, seems the semantics may changed compared to the original one, I'm not sure the logic is still correct.

[jerryshao]

I think removing this if case will make the below computation invalid, take block number 200 and maxMergeWidth 10 as a consideration.

[sryza]

What issues do you think would occur?

I tried calculating this out on a spreadsheet, and it seems alright as far as I can tell.
https://docs.google.com/spreadsheets/d/1aTi8NgIGg7wBdNf_bzglYIJyokBe6-iYb6HEWzEpdKw/edit?usp=sharing

[jerryshao]

I think this will make val mergedBlocksHeadroomAfterMerge = (maxMergeWidth - mergedBlocks.size() - 1) to be a negative number, And blocksAwaitingMerge.size() - mergedBlocksHeadroomAfterMerge to be a number bigger than blocksAwaitingMerge.size(). I'm not sure if this will lead to some unexpected values. So the above code is a constraint to make sure we get a positive idealWidth

[sryza]

Oops, forgot to replace blocksAwaitingMerge.size() with remainingToMergeAtLevel. After making that replacement, I'm pretty sure it works out.

Here's my logic:
mergedBlocksHeadroomAfterMerge can be negative, but idealWidth will always be positive. Further, if mergedBlocksHeadroomAfterMerge is negative, then idealWidth will be greater than maxMergeWidth. This is because, if we have made it here, remainingToMergeAtLevel is >= maxMergeWidth, and subtracting a negative number from it will make it > maxMergeWidth. So in all situations where mergedBlocksHeadroomAfterMerge is negative, the merge width will be chosen as maxMergeWidth, which is the correct behavior.

[jerryshao]

OK, I got it :)

[jerryshao]

I still think we may get ZERO numMapBlocks, because this value is got from fetchRawBlocks() and this merging thread is started before calling this function. So I think maybe we can separate getting numMapBlocks and fetchRawBlocks into two functions.

[jerryshao]

Typo from.

[jerryshao]

Hi Sandy, I have a question,

Currently, operators like combineByKey() with map side combine and sortByKey() will use this SortShuffleReader to do shuffle read, but according to my test, this SortShuffleReader has no obvious improvement in combineByKey() operations, also combineByKey() may not always do key sorting within partition when there's no spill, this will potential lead to bug for merge-sorting. So I think can we just enable this shuffle reader only for sortByKey() like operators, for others will follow the code path of HashShuffleReader, what's your opinion?

[jerryshao]

Both have this here and in TieredDiskMerger, can we just remove another one and pass it by parameter.

[sryza]

Good point.

[sryza]

I agree that HashShuffleReader is probably better in many situations, especially for operators with high reduction factors. My guess is that SortShuffleReader could also perform better on operators like groupByKey that don't reduce the data, because it would place less GC pressure by keeping fewer deserialized objects in memory at a time. We'd probably need to run experiments to verify?

[jerryshao]

I agree with you that HashShuffleReader is better in high reduction factor situation. For groupByKey situation, according to my previous test using SparkPerf, the performance is not so good because we just do a another round of unnecessary by key sorting, probably we can also avoid this. I will do some experiments to verify it. Thanks a lot.

[jerryshao]

I'm not sure this solution now is still a optimal or near optimal solution.

[sryza]

Do you have any counterexamples to optimality?

[jerryshao]

I've no specific counterexamples currently, just not sure about it.

Besides, I think setting maxMergeWidth to 10 by default might be a little smaller, previously for MR equal one io.sort.factor often choose to be 100 if I remembered correctly.

[sryza]

I agree that 100 is a better choice.

[jerryshao]

I will merge this PR and rebase to the latest code when this (apache#2753) is merged. Then we can still do some improvements and benchmarks. Really appreciate your work, thanks a lot :).

[sryza]

You're right that we can avoid ser/deser costs by spilling memory blocks to disk directly instead of merging to disk in some situations. In other situations, merging to disk is better because it allows us to avoid future disk to disk merges. I'll think more about what a good rule would be for deciding, as well as update my patch with some of the other changes soon.

I also tried some microbenchmarking of serialization / deserialization cost:

• Serialize a million (String, (Int, String)) tuples with Kryo reference tracking turned on: ~600 ms
• Serialize a million (String, (Int, String)) tuples with Kryo reference tracking turned off: ~300 ms
• Deserialize a million (String, (Int, String)) tuples with Kryo reference tracking turned on: ~400 ms
• Deserialize a million (String, (Int, String)) tuples with Kryo reference tracking turned off: ~300 ms

[jerryshao]

Hi Sandy, I'm going to merge this PR and rebase to the latest master, since NettyBlockTransferService has merged into the master. Then we can still improve our code :).

[jerryshao]

Hi @sryza , through careful thinking, I think this patch may still have several problems:

1. merging strategy is not the best strategy, introduce memory-to-memory merging will lead to many times of merging. Taking block number 30 and mergeWidth 10 as consideration, if the shuffle memory can only fit for 2 blocks, so we have to do 15 memory-to-memory merging, and left 15 disk files for merging. This is much more the the previous strategy (3 times of merging).
2. In Spark shuffle memory is shared for tasks in one executor both in map and reduce side, even if we do memory merge to release the occupied memory, this memory may still be acquired by another task. So do memory merge may not alleviate the overhead.
3. Current merging strategy is not deterministic optimal strategy. Also the merging factor and generated files are related to shuffle memory size, without get the total block number, we can hardly get the global optimal result.

[jerryshao]

I will do some improvements and try to figure out a better strategy.

[sryza]

This is much more the the previous strategy (3 times of merging).

It carries out more merges, but results in less disk IO, right? In the previous strategy, we could end up with 28 segments on disk, which would result in two disk-to-disk merges - one of 10 segments and one of 9 segments. If each segment contains 10 bytes, that would be 190 bytes of disk read and write. With merging the segments to disk, we end up with 14 segments. We would do a single merge of 5 segments, resulting in 100 bytes of disk read / write (because the segments are double the size).

I've been doing some thinking about this and wrote a simulation yesterday that, given a number of blocks on disk and maxMergeWidth, calculates the number of disk-to-disk merges and amount of IO required for them.

import java.util.PriorityQueue

def calcMergeCost(numBlocks: Int, maxMergeWidth: Int): (Int, Int) = {
  val blocks = new PriorityQueue[Int]()
  var merges = 0
  var unitsMerged = 0
  (0 until numBlocks).foreach(x => blocks.offer(1))
  while (blocks.size > maxMergeWidth) {
    val mergeWidth = math.min(blocks.size - maxMergeWidth + 1, maxMergeWidth)
    val unitsThisMerge = (0 until mergeWidth).map(x => blocks.poll()).sum
    blocks.offer(unitsThisMerge)
    unitsMerged += unitsThisMerge
    merges += 1
  }
  (merges, unitsMerged)
}

It looks like, after the number of disk blocks exceeds maxMergeWidth, the amount of IO increases at something like O(n log n) with the number of blocks. So one possible strategy would be to carry out the minimum number of mem-to-disk merges that would keep the number of disk blocks under maxMergeWidth.

[sryza]

One other thing to fix about the approach I posted is that, if we have a total of 11 blocks of size 10MB, and the memory space is 100MB, we'll end up spilling 100MB of data. We should be able to only spill 10MB.