Difference between shuffle() and rebalance() in Apache Flink

7
votes

I am working on my bachelor's final project, which is about the comparison between Apache Spark Streaming and Apache Flink (only streaming) and I have just arrived to "Physical partitioning" in Flink's documentation. The matter is that in this documentation it doesn't explain well how this two transformations work. Directly from the documentation:

shuffle(): Partitions elements randomly according to a uniform distribution.

rebalance(): Partitions elements round-robin, creating equal load per partition. Useful for performance optimisation in the presence of data skew.

Source: https://ci.apache.org/projects/flink/flink-docs-release-1.2/dev/datastream_api.html#physical-partitioning

Both are automatically done, so what I understand is that they both redistribute equally (shuffle() > uniform distribution & rebalance() > round-robin) and randomly the data. Then I deduce that rebalance() distributes the data in a better way ("equal load per partitions") so the tasks have to process the same amount of data, but shuffle() may create bigger and smaller partitions. Then, in which cases might you prefer to use shuffle() than rebalance()?

The only thing that comes to my mind is that probably rebalance()requires some processing time so in some cases it might use more time to do the rebalancing than the time it will improve in the future transformations.

I have been looking for this and nobody has talked about this, only in a mailing list of Flink, but they don't explain how shuffle() works.

Thanks to Sneftel who has helped me to improve my question asking me things to let me rethink about what I wanted to ask; and to Till who answered quite well my question. :D

2
bigdataapache-flinkpartitioningflink-streaming
what made you think that rebalancewas random? - Sneftel
It is not random, it tries to put the elements in a balanced way for the processing to be more efficient, but if this is the only difference, then i find no reason for "shuffle()" to exist as "rebalance()" does the same in a more efficient way. - froblesmartin
what makes you think that rebalance() is more efficient? They're just different approaches. shuffle() is a randomized approach to load-balancing, rebalance() is an explicit greedy one. - Sneftel
From the documentation I understand that shuffle() distributes the elements in a random and uniform way so it may not create same loaded partitions, meanwhile rebalance() tries to create all the partitions with the same load. Then I deduce that rebalance() does the same but in a more efficient way for the work distribution as all the TaskManagers will have approximately the same data to process. Then if rebalance() does better the same job, why will anyone use shuffle()? May the processing needed by rebalance() produce more latency than which it can improve in some cases? Thanks :) - froblesmartin

2 Answers

12
votes

As the documentation states, shuffle will randomly distribute the data whereas rebalance will distribute the data in a round robin fashion. The latter is more efficient since you don't have to compute a random number. Moreover, depending on the randomness, you might end up with some kind of not so uniform distribution.

On the other hand, rebalance will always start sending the first element to the first channel. Thus, if you have only few elements (fewer elements than subtasks), then only some of the subtasks will receive elements, because you always start to send the first element to the first subtask. In the streaming case this should eventually not matter because you usually have an unbounded input stream.

The actual reason why both methods exist is a historically reason. shuffle was introduced first. In order to make the batch an streaming API more similar, rebalance was then introduced.

1
votes

This statement by Flink is misleading:

Useful for performance optimisation in the presence of data skew.

Since it's used to describe rebalance, but not shuffle, it suggests it's the distinguishing factor. My understanding of it was that if some items are slow to process and some fast, the partitioner will use the next free channel to send the item to. But this is not the case, compare the code for rebalance and shuffle. The rebalance just adds to next channel regardless how busy it is.

// rebalance
nextChannelToSendTo = (nextChannelToSendTo + 1) % numberOfChannels;

// shuffle
nextChannelToSendTo = random.nextInt(numberOfChannels);

The statement can be also understood differently: the "load" doesn't mean actual processing time, just the number of items. If your original partitioning has skew (vastly different number of items in partitions), the operation will assign items to partitions uniformly. However in this case it applies to both operations.

My conclusion: shuffle and rebalance do the same thing, but rebalance does it slightly more efficiently. But the difference is so small that it's unlikely that you'll notice it, java.util.Random can generate 70m random numbers in a single thread on my machine.