|
If you were logged in you would be able to see more operations.
|
|
|
|
File Attachments:
|
|
|
Issue Links:
|
Blocker
|
|
|
|
This issue is blocked by:
|
|
HADOOP-2943
Compression for intermediate map output is broken
|
|
|
|
|
Incorporates
|
|
This issue incorporates:
|
|
HADOOP-872
map output sorter doesn't compress the outputs before the sort
|
|
|
|
 |
HADOOP-1609
Optimize MapTask.MapOutputBuffer.spill() by not deserialize/serialize keys/values but use appendRaw
|
|
|
|
|
|
HADOOP-287
Speed up SequenceFile sort with memory reduction
|
|
|
|
|
|
|
|
|
|
|
|
| Resolution Date: |
31/Mar/08 10:51 PM
|
Currently, the sort/spill works as follows:
Let r be the number of partitions
For each call to collect(K,V) from map:
- If buffers do not exist, allocate a new DataOutputBuffer to collect K,V bytes, allocate r buffers for collecting K,V offsets
- Write K,V into buffer, noting offsets
- Register offsets with associated partition buffer, allocating/copying accounting buffers if nesc
- Calculate the total mem usage for buffer and all partition collectors by iterating over the collectors
- If total mem usage is greater than half of io.sort.mb, then start a new thread to spill, blocking if another spill is in progress
For each spill (assuming no combiner):
- Save references to our K,V byte buffer and accounting data, setting the former to null (will be recreated on the next call to collect(K,V))
- Open a SequenceFile.Writer for this partition
- Sort each partition separately (the current version of sort reuses, but still requires wrapping, indices in IntWritable objects)
- Build a RawKeyValueIterator of sorted data for the partition
- Deserialize each key and value and call SequenceFile::append(K,V) on the writer for this partition
There are a number of opportunities for reducing the number of copies, creations, and operations we perform in this stage, particularly since growing many of the buffers involved requires that we copy the existing data to the newly sized allocation.
|
|
Description
|
Currently, the sort/spill works as follows:
Let r be the number of partitions
For each call to collect(K,V) from map:
- If buffers do not exist, allocate a new DataOutputBuffer to collect K,V bytes, allocate r buffers for collecting K,V offsets
- Write K,V into buffer, noting offsets
- Register offsets with associated partition buffer, allocating/copying accounting buffers if nesc
- Calculate the total mem usage for buffer and all partition collectors by iterating over the collectors
- If total mem usage is greater than half of io.sort.mb, then start a new thread to spill, blocking if another spill is in progress
For each spill (assuming no combiner):
- Save references to our K,V byte buffer and accounting data, setting the former to null (will be recreated on the next call to collect(K,V))
- Open a SequenceFile.Writer for this partition
- Sort each partition separately (the current version of sort reuses, but still requires wrapping, indices in IntWritable objects)
- Build a RawKeyValueIterator of sorted data for the partition
- Deserialize each key and value and call SequenceFile::append(K,V) on the writer for this partition
There are a number of opportunities for reducing the number of copies, creations, and operations we perform in this stage, particularly since growing many of the buffers involved requires that we copy the existing data to the newly sized allocation. |
Show » |
|
Improvement
Blocker
