In Hadoop , when do reduce tasks start?
Reducers start based on a threshold of percentage of mappers that have finished.
There are 3 phases of reduce: shuffle, sort, reduce.
- Shuffle is only a data transfer where the data is collected by the reducer from each mapper which can happen while mappers are generating data.
- Sort and reduce can only start once all the mappers are done.
The reduce phase has 3 steps: shuffle, sort, reduce. Shuffle is where the data is collected by the reducer from each mapper.
This can happen while mappers are generating data since it is only a data transfer.
On the other hand, sort and reduce can only start once all the mappers are done.
Reducers start based on a threshold of percentage of mappers that have finished. You can change the parameter to get reducers to start sooner or later.
Why is starting the reducers early a good thing?
Because it spreads out the data transfer from the mappers to the reducers over time, which is a good thing if your network is the bottleneck.
Why is starting the reducers early a bad thing? Because they “hog up” reduce slots while only copying data.
Another job that starts later that will actually use the reduce slots now can’t use them.
You can customize when the reducers startup by changing the default value of mapred.reduce.slowstart.completed.maps in mapred-site.xml. A value of 1.00 will wait for all the mappers to finish before starting the reducers. A value of 0.0 will start the reducers right away. A value of 0.5 will start the reducers when half of the mappers are complete. You can also change mapred.reduce.slowstart.completed.maps on a job-by-job basis.
Typically, I like to keep mapred.reduce.slowstart.completed.maps above 0.9 if the system ever has multiple jobs running at once. This way the job doesn’t hog up reducers when they aren’t doing anything but copying data. If you only ever have one job running at a time, doing 0.1 would probably be appropriate
When a reduce task starts, its input is scattered in many files across all the nodes where map tasks ran. If run in distributed mode these need to be first copied to the local filesystem in a copy phase.
Once all the data is available locally it is appended to one file in an append phase. The file is then merge sorted so that the key-value pairs for a given key are contiguous (sort phase). This makes the actual reduce operation simple: the file is read sequentially and the values are passed to the reduce method with an iterator reading the input file until the next key value is encountered.
At the end, the output will consist of one output file per executed reduce task. The format of the files can be specified with JobConf.setOutputFormat. If SequentialOutputFormat is used then the output key and value classes must also be specified.