Shuffle and Sort

Shuffle is where Tez moves data between vertices, and it is where most production performance problems and OOMs live. The stack has two halves, both in tez-runtime-library:

Sort path (producer side): the processor writes (K, V) pairs into an output; the output partitions, sorts, spills, and merges them into IFile segments on local disk, then advertises them. OrderedPartitionedKVOutput → PipelinedSorter / DefaultSorter → IFile.
Shuffle path (consumer side): the input fetches those segments over HTTP (or local disk), merges them, and presents a sorted stream to the processor. OrderedGroupedKVInput → ShuffleScheduler + FetcherOrderedGrouped → MapOutput → MergeManager → ValuesIterator.

Between them sits a NodeManager auxiliary service that serves spilled segments over HTTP. There are two: the Hadoop mapreduce_shuffle handler and Tez's own tez_shuffle handler — this chapter gets that distinction right, because the folklore ("Tez has no shuffle service") is wrong.

After this chapter you can name which input uses which shuffle implementation, read the IFile on-disk format, size the sort and shuffle memory knobs from their real defaults, and diagnose fetch-failure storms and sort-buffer OOMs. This is the physical machinery behind the SCATTER_GATHER edge from logical-physical.md; the IO contracts are in ipo-abstractions.md; the labs lab-01-debug-shuffle.md and lab-02-modify-processor.md exercise it directly.

ls tez-runtime-library/src/main/java/org/apache/tez/runtime/library/common/sort/impl/
ls tez-runtime-library/src/main/java/org/apache/tez/runtime/library/common/shuffle/

The producer side

`OrderedPartitionedKVOutput`

grep -n "requestInitialMemory\|getWriter\|sorter.flush\|generateEvents\|SorterImpl.PIPELINED\|SorterImpl.LEGACY" \
  tez-runtime-library/src/main/java/org/apache/tez/runtime/library/output/OrderedPartitionedKVOutput.java

Lifecycle (the full IO contract is in ipo-abstractions.md):

initialize() — requests tez.runtime.io.sort.mb of buffer through the memory handshake; does no I/O.
start() — reads tez.runtime.sorter.class and instantiates the sorter. Real selection code:

// tez-runtime-library, OrderedPartitionedKVOutput.start()
String sorterClass = conf.get(TezRuntimeConfiguration.TEZ_RUNTIME_SORTER_CLASS,
    TezRuntimeConfiguration.TEZ_RUNTIME_SORTER_CLASS_DEFAULT).toUpperCase(Locale.ENGLISH);
SorterImpl sorterImpl = SorterImpl.valueOf(sorterClass);
// ...
if (sorterImpl.equals(SorterImpl.PIPELINED)) {
  sorter = new PipelinedSorter(getContext(), conf, getNumPhysicalOutputs(),
      memoryUpdateCallbackHandler.getMemoryAssigned());
} else if (sorterImpl.equals(SorterImpl.LEGACY)) {
  sorter = new DefaultSorter(getContext(), conf, getNumPhysicalOutputs(),
      memoryUpdateCallbackHandler.getMemoryAssigned());
}

getWriter() — returns a KeyValuesWriter delegating to the sorter.
close() — sorter.flush() merges spills, then generateEvents() emits one CompositeDataMovementEvent per output plus a VertexManagerEvent (byte stats for auto-parallelism).

Two sorters

find tez-runtime-library/src/main/java -name "PipelinedSorter.java" -o -name "DefaultSorter.java" \
  -o -name "ExternalSorter.java"

Both extend ExternalSorter. SorterImpl (defined in OrderedPartitionedKVOutputConfig) has exactly two values: PIPELINED and LEGACY.

Sorter	`tez.runtime.sorter.class`	Strategy
`PipelinedSorter`	`PIPELINED` (default)	multiple sort spans, a background `sortmaster` thread pool sorts spans while the writer fills the next; can skip the final merge and emit each spill as its own event (pipelined shuffle)
`DefaultSorter`	`LEGACY`	single `kvbuffer` + parallel metadata region, quicksort by `(partition, key)`, spills at `sort.spill.percent`, one final merge of all spills — MapReduce parity

Verify the default is PIPELINED, not LEGACY:

grep -n "TEZ_RUNTIME_SORTER_CLASS_DEFAULT" \
  tez-runtime-library/src/main/java/org/apache/tez/runtime/library/api/TezRuntimeConfiguration.java

// tez-runtime-library, TezRuntimeConfiguration
public static final String TEZ_RUNTIME_SORTER_CLASS_DEFAULT = SorterImpl.PIPELINED.name();

Sort configuration knobs (real keys + defaults)

grep -n "IO_SORT_MB\|SORT_SPILL_PERCENT\|IO_SORT_FACTOR\|SORTER_CLASS\|PIPELINED_SORTER_MIN_BLOCK\|PIPELINED_SORTER_SORT_THREADS\|COMBINER_CLASS\|TEZ_RUNTIME_COMPRESS\b" \
  tez-runtime-library/src/main/java/org/apache/tez/runtime/library/api/TezRuntimeConfiguration.java

Key	Default	Effect
`tez.runtime.io.sort.mb`	`100`	sort buffer size, MB. Shared by both sorters.
`tez.runtime.sort.spill.percent`	`0.8`	`DefaultSorter` spills when the buffer crosses this.
`tez.runtime.io.sort.factor`	`100`	max segments merged in one pass.
`tez.runtime.sorter.class`	`PIPELINED`	`PIPELINED` or `LEGACY`.
`tez.runtime.pipelined-sorter.sort.threads`	`2`	`PipelinedSorter` sort-thread count.
`tez.runtime.pipelined-sorter.min-block.size.in.mb`	`2000`	block-size cap for lazy allocation.
`tez.runtime.combiner.class`	unset	combiner, run during merge.
`tez.runtime.compress` / `.compress.codec`	`false` / —	per-segment compression.

Warning: DefaultSorter caps the usable buffer. MAX_IO_SORT_MB = 1800, and it clamps availableMemoryMB down to that. Setting tez.runtime.io.sort.mb=4096 does not give a DefaultSorter 4 GB — it silently truncates to 1800 MB and logs it. PipelinedSorter uses multiple blocks and does not hit this cap.
grep -n "MAX_IO_SORT_MB\|availableMemoryMB > MAX_IO_SORT_MB" \
  tez-runtime-library/src/main/java/org/apache/tez/runtime/library/common/sort/impl/dflt/DefaultSorter.java

IFile: the on-disk segment format

IFile is the format both sorters write and every fetcher reads.

grep -n "EOF_MARKER\|RLE_MARKER\|V_END_MARKER\|byte\[\] HEADER\|writeKVPair\|checksumSize" \
  tez-runtime-library/src/main/java/org/apache/tez/runtime/library/common/sort/impl/IFile.java

The header and markers are real constants — do not invent them:

// tez-runtime-library, org.apache.tez.runtime.library.common.sort.impl.IFile
public static final int EOF_MARKER   = -1;  // End of File
public static final int RLE_MARKER   = -2;  // Repeat-same-key marker
public static final int V_END_MARKER = -3;  // End of values marker
static final byte[] HEADER = new byte[] { (byte)'T', (byte)'I', (byte)'F', (byte)0 };

Whole-stream layout: HEADER + <records> + CHECKSUM. The checksum is written by IFileOutputStream over the real data and appended at close; checksumSize comes from IFileOutputStream.getCheckSumSize(). Per record:

+---------------+---------------+----------------+------------------+
| keyLen (VInt) | valLen (VInt) | key bytes (KL) | value bytes (VL) |
+---------------+---------------+----------------+------------------+
end of segment:  keyLen = EOF_MARKER (-1)

writeKVPair writes exactly this, in order:

// tez-runtime-library, IFile.Writer.writeKVPair()
writeValueMarker(out);
WritableUtils.writeVInt(out, keyLength);
WritableUtils.writeVInt(out, valueLength);
out.write(keyData, keyPos, keyLength);
out.write(valueData, valPos, valueLength);

Run-length encoding is applied when consecutive keys repeat: instead of re-writing the key, IFile writes RLE_MARKER and terminates the value run with V_END_MARKER. This is why the reducer-facing API is (key, Iterable<value>) — the format itself groups repeats.

Note: When compression is enabled, the record framing lives inside the compressed stream; HEADER and the trailing checksum are outside it. A fetcher that reads a truncated compressed segment sees a Premature EOF from inputStream, not a checksum error.

Each output produces one data file plus a TezSpillRecord index describing each partition:

grep -n "startOffset\|rawLength\|partLength" \
  tez-runtime-library/src/main/java/org/apache/tez/runtime/library/common/sort/impl/TezIndexRecord.java

// tez-runtime-library, org.apache.tez.runtime.library.common.sort.impl.TezIndexRecord
private long startOffset;   // offset of this partition within the data file
private long rawLength;     // raw (uncompressed) length
private long partLength;    // on-disk length (includes checksum + compression)

The ShuffleHandler reads this index to answer "give me partition p of source attempt (vertex, task, attempt)" without scanning the data file.

Spill and merge

sequenceDiagram
  participant P as Processor
  participant W as KeyValuesWriter
  participant S as Sorter
  participant D as Local disk
  P->>W: write(K,V) x N
  W->>S: collect into sort buffer
  S->>S: buffer crosses sort.spill.percent
  S->>D: spill_0.out (partitioned, sorted) + spill_0.out.index
  Note over S: PipelinedSorter keeps accepting writes into the next span
  P->>W: close()
  W->>S: flush()
  S->>D: merge spill_0..spill_N -> file.out + file.out.index
  S-->>P: CompositeDataMovementEvent (per partition) + VertexManagerEvent

The combiner (tez.runtime.combiner.class) runs during the merge, not during accumulation, over already-sorted runs. Tez gives no guarantee on how many merge passes invoke it, so a combiner must be associative and commutative — a non-idempotent combiner produces wrong counts across passes.

Inside `PipelinedSorter` — spans, blocks, and pipelined shuffle

grep -n "class SortSpan\|sortmaster\|MIN_BLOCK_SIZE\|pipelinedShuffle\|isFinalMergeEnabled" \
  tez-runtime-library/src/main/java/org/apache/tez/runtime/library/common/sort/impl/PipelinedSorter.java

PipelinedSorter divides its buffer into SortSpans. While the writer fills one span, a fixed sortmaster thread pool (tez.runtime.pipelined-sorter.sort.threads, default 2) sorts the previous span — that overlap is the entire point, and it is why the sorter cuts wall-clock time without reducing CPU work. Blocks are sized between the requested memory and MIN_BLOCK_SIZE:

// tez-runtime-library, PipelinedSorter (buffer setup)
MIN_BLOCK_SIZE = ((256 << 20) - 64);   // clamped by tez.runtime.pipelined-sorter.min-block.size.in.mb
// ...
pipelinedShuffle = !isFinalMergeEnabled() && confPipelinedShuffle;

That last line is the edge case that bites people. PipelinedSorter will emit each spill as its own DataMovementEvent — skipping the final merge — only when both tez.runtime.pipelined-shuffle.enabled (default false) is set and tez.runtime.enable.final-merge.in.output (default true) is turned off. Enabling pipelined shuffle while leaving final merge on silently disables the final merge and logs it; and pipelined shuffle refuses to run on DefaultSorter at all (Preconditions.checkArgument(sorterImpl == PIPELINED)).

Warning — pipelined-shuffle edge case: with pipelined shuffle on, a source attempt emits multiple events (one per spill) instead of one final event. If that attempt is later re-run, the consumer may already hold spills from the dead attempt — Tez must track and discard them by attempt number. Mis-tuning here (enabling pipelined shuffle on a DAG with heavy speculative execution) multiplies fetch bookkeeping and can regress the very latency it was meant to improve. Treat it as a targeted optimization, not a default.

The unordered outputs use a different, simpler buffer: tez.runtime.unordered.output.buffer.size-mb (default 100), with no sort at all.

The consumer side

Two shuffle implementations — which input uses which

This is the fact people get wrong. There are two entirely separate shuffle packages, and the input class picks one:

grep -n "import org.apache.tez.runtime.library.common.shuffle" \
  tez-runtime-library/src/main/java/org/apache/tez/runtime/library/input/OrderedGroupedKVInput.java \
  tez-runtime-library/src/main/java/org/apache/tez/runtime/library/input/UnorderedKVInput.java

Input	Shuffle package	Key classes
`OrderedGroupedKVInput`	`...common.shuffle.orderedgrouped`	`Shuffle`, `ShuffleScheduler`, `FetcherOrderedGrouped`, `MergeManager`, `MapOutput`, `MapHost`
`UnorderedKVInput`	`...common.shuffle.impl` (+ shared `...common.shuffle`)	`ShuffleManager`, `ShuffleInputEventHandlerImpl`, `Fetcher`, `SimpleFetchedInputAllocator`, `FetchedInput`

The ordered path merges and sorts (reduce-side grouping). The unordered path fetches and concatenates, no merge — used for hash joins and broadcast. Do not mix the class names across packages; Fetcher (unordered) and FetcherOrderedGrouped (ordered) are different classes.

OrderedGroupedKVInput.start() builds the Shuffle, which owns the scheduler, the fetchers, and the merge manager:

// tez-runtime-library, OrderedGroupedKVInput.start()
public synchronized void start() throws IOException {
  if (!isStarted.get()) {
    memoryUpdateCallbackHandler.validateUpdateReceived();
    shuffle = createShuffle();       // ShuffleScheduler + FetcherOrderedGrouped[] + MergeManager
    shuffle.run();
    // drain events queued before start
    isStarted.set(true);
  }
}

Fetcher and the shuffle URL

A fetcher connects over HTTP to the ShuffleHandler on the source task's node. The URL is built in ShuffleUtils — read the real constructor, do not guess the query string:

grep -n "mapOutput?job=\|&dag=\|&reduce=\|&map=\|constructBaseURIForShuffleHandler" \
  tez-runtime-library/src/main/java/org/apache/tez/runtime/library/common/shuffle/ShuffleUtils.java

// tez-runtime-library, ShuffleUtils.constructBaseURIForShuffleHandler()
sb.append("mapOutput?job=");
sb.append(appId.replace("application", "job"));
sb.append("&dag=");   sb.append(dagIdentifier);
sb.append("&reduce="); sb.append(partition);   // "-<partition+count-1>" if partitionCount>1
sb.append("&map=");                              // then comma-separated path components

The response streams the requested attempts back-to-back, each prefixed with a ShuffleHeader:

// tez-runtime-library, orderedgrouped.ShuffleHeader
String mapId;
long uncompressedLength;
long compressedLength;
int  forReduce;

The fetcher reads the header, asks the merge manager to reserve(size), and either buffers in memory or streams to disk.

Local-disk fetch shortcut

When the source output lives on the same node, HTTP is wasteful. The fetcher detects this and reads the local file directly:

grep -n "localDiskFetchEnabled\|isLocalFetch\|setupLocalDiskFetch\|optimize.local.fetch" \
  tez-runtime-library/src/main/java/org/apache/tez/runtime/library/common/shuffle/Fetcher.java

// tez-runtime-library, common.shuffle.Fetcher.callInternal()
boolean isLocalFetch =
    localDiskFetchEnabled && host.equals(localHostname) && port == shufflePort;
if (isLocalFetch) {
  hostFetchResult = setupLocalDiskFetch();   // read the IFile off local disk, no HTTP
}

Controlled by tez.runtime.optimize.local.fetch (default true). Disabling it forces every fetch through the ShuffleHandler even for co-located outputs — occasionally useful when debugging the handler, terrible for throughput.

Shuffle configuration knobs (real keys + defaults)

grep -n "SHUFFLE_PARALLEL_COPIES\|FETCH_MAX_TASK_OUTPUT_AT_ONCE\|SHUFFLE_FETCH_BUFFER_PERCENT\|SHUFFLE_MEMORY_LIMIT_PERCENT\|SHUFFLE_MERGE_PERCENT\|SHUFFLE_READ_TIMEOUT\|MEMTOMEM" \
  tez-runtime-library/src/main/java/org/apache/tez/runtime/library/api/TezRuntimeConfiguration.java

Key	Default	Effect
`tez.runtime.shuffle.parallel.copies`	`20`	fetcher threads per task.
`tez.runtime.shuffle.fetch.max.task.output.at.once`	`20`	max attempts per HTTP request.
`tez.runtime.shuffle.fetch.buffer.percent`	`0.90`	fraction of heap the merge manager may use.
`tez.runtime.shuffle.memory.limit.percent`	`0.25`	max fraction of that budget a single input may occupy in memory before going to disk.
`tez.runtime.shuffle.merge.percent`	`0.90`	in-memory usage that triggers an in-memory merge.
`tez.runtime.shuffle.read.timeout`	`180000` (3 min)	HTTP socket read timeout.
`tez.runtime.shuffle.memory-to-memory.enable`	`false`	enable the memory-to-memory merger.

MergeManager: three merge tracks

grep -n "IntermediateMemoryToMemoryMerger\|InMemoryMerger\|OnDiskMerger\|maxSingleShuffleLimit\|mergeThreshold\|postMergeMemLimit" \
  tez-runtime-library/src/main/java/org/apache/tez/runtime/library/common/shuffle/orderedgrouped/MergeManager.java

// tez-runtime-library, orderedgrouped.MergeManager (fields)
private final IntermediateMemoryToMemoryMerger memToMemMerger;   // optional
private final InMemoryMerger inMemoryMerger;                     // in-mem -> disk
final OnDiskMerger onDiskMerger;                                 // disk -> disk
private final long maxSingleShuffleLimit;   // biggest input allowed to stay in memory
private final long mergeThreshold;          // in-mem usage that kicks a merge
private final long postMergeMemLimit;       // budget for the final merged output

The three tracks:

Memory-to-memory (off by default): merges several in-memory inputs into one, avoiding a disk round-trip. Enabled by tez.runtime.shuffle.memory-to-memory.enable.
In-memory merge (InMemoryMerger): when in-memory inputs cross mergeThreshold, merge them and spill the result to disk.
On-disk merge (OnDiskMerger): when on-disk segments accumulate, merge io.sort.factor at a time into fewer, larger segments.

At processor-pull time, a final merge combines the remaining in-memory and on-disk inputs into one sorted KeyValuesReader (a ValuesIterator), which presents (key, Iterable<value>) — the classic reducer API. Where an input lands is decided by MapOutput.Type:

// tez-runtime-library, orderedgrouped.MapOutput.Type
WAIT, MEMORY, DISK, DISK_DIRECT

An input larger than maxSingleShuffleLimit goes straight to DISK; a hard precondition enforces maxSingleShuffleLimit < mergeThreshold so a single giant input can never wedge the in-memory budget.

sequenceDiagram
  participant SM as ShuffleScheduler
  participant F as FetcherOrderedGrouped
  participant NM as Source NM (ShuffleHandler)
  participant MM as MergeManager
  participant T as Processor
  SM->>F: assign (source attempt, partition)
  F->>NM: GET /mapOutput?job=..&dag=..&reduce=p&map=attempt1,attempt2
  NM-->>F: ShuffleHeader + IFile bytes (per attempt)
  F->>MM: reserve(size)
  alt fits & < maxSingleShuffleLimit
    MM-->>F: MapOutput(MEMORY)
  else too big
    MM-->>F: MapOutput(DISK)
  end
  F->>MM: commit
  MM->>MM: InMemoryMerger / OnDiskMerger when thresholds crossed
  T->>SM: getReader() (blocks until all inputs done)
  SM->>MM: finalMerge()
  MM-->>T: ValuesIterator (key, Iterable<value>)

The ShuffleHandler — Tez has its own

The consumer fetches from a NodeManager auxiliary service. Two exist, and the distinction matters for cluster setup.

find tez-plugins/tez-aux-services/src/main/java -name "ShuffleHandler.java"
grep -n "TEZ_SHUFFLE_SERVICEID\|SHUFFLE_PORT_CONFIG_KEY\|DEFAULT_SHUFFLE_PORT\|STATE_DB_NAME" \
  tez-plugins/tez-aux-services/src/main/java/org/apache/tez/auxservices/ShuffleHandler.java

Tez ships org.apache.tez.auxservices.ShuffleHandler in the tez-plugins/tez-aux-services module:

// tez-plugins/tez-aux-services, org.apache.tez.auxservices.ShuffleHandler
public static final String TEZ_SHUFFLE_SERVICEID   = "tez_shuffle";
public static final String SHUFFLE_PORT_CONFIG_KEY = "tez.shuffle.port";
public static final int    DEFAULT_SHUFFLE_PORT    = 13563;
private static final String STATE_DB_NAME          = "tez_shuffle_state";

But by default Tez still targets the Hadoop handler. The AM's aux-service id defaults to mapreduce_shuffle:

// tez-api, org.apache.tez.dag.api.TezConfiguration / TezConstants
// tez.am.shuffle.auxiliary-service.id  default:
public static final String TEZ_AM_SHUFFLE_AUXILIARY_SERVICE_ID_DEFAULT =
    TezConstants.TEZ_SHUFFLE_HANDLER_SERVICE_ID;   // = "mapreduce_shuffle"

So the correct statement is: Tez runs on the Hadoop mapreduce_shuffle handler out of the box, but ships its own tez_shuffle handler you can deploy as an aux service and point tez.am.shuffle.auxiliary-service.id at. Either way, the NodeManager loads it via yarn-site.xml:

<property>
  <name>yarn.nodemanager.aux-services</name>
  <value>mapreduce_shuffle</value>   <!-- or tez_shuffle -->
</property>

A missing or mis-named aux service is the number-one cause of java.net.ConnectException in the fetcher.

Failure handling: fetch failures → `InputReadErrorEvent`

When a fetch fails, the scheduler decides whether it is a transient blip or a signal that the source output is gone. Read the decision:

grep -n "copyFailed\|maxFetchFailuresBeforeReporting\|reportReadErrorImmediately\|informAM\|penalties" \
  tez-runtime-library/src/main/java/org/apache/tez/runtime/library/common/shuffle/orderedgrouped/ShuffleScheduler.java

// tez-runtime-library, orderedgrouped.ShuffleScheduler.copyFailed()  (abbreviated)
boolean shouldInformAM =
    (reportReadErrorImmediately && (readError || connectError))
        || ((failures % maxFetchFailuresBeforeReporting) == 0);   // default threshold 5
if (shouldInformAM) {
  informAM(fetchFailure);   // sends InputReadErrorEvent to the AM
}

informAM sends an InputReadErrorEvent (see the event catalog in ipo-abstractions.md):

// tez-runtime-library, orderedgrouped.ShuffleScheduler.informAM()
failedEvents.add(InputReadErrorEvent.create(
    "Fetch failure for " + taskAttemptIdentifier + " to jobtracker.",
    srcAttempt.getInputIdentifier(), srcAttempt.getAttemptNumber(),
    fetchFailure.isLocalFetch(), fetchFailure.isDiskErrorAtSource(), localHostname));
inputContext.sendEvents(failedEvents);

The AM correlates InputReadErrorEvents and, past a threshold, re-runs the source attempt (the AM side is in failure-handling.md). Failed hosts go on a penalties DelayQueue so the scheduler backs off before retrying — this is the mechanism that keeps a single slow NodeManager from starving all fetcher threads.

Warning — fetch-failure storms: if a source NodeManager dies, every consumer's fetchers fail on its partitions simultaneously, each firing InputReadErrorEvents. The AM re-runs the source, but until the replacement completes, consumers spin against the penalty queue. Symptoms: a long shuffle "plateau" with rising FailedShuffleCounter. Mitigation is source-side output durability (PERSISTED_RELIABLE) plus tuned maxFetchFailuresBeforeReporting, not more fetcher threads.

Reading exercise

# Verify the EOF sentinel and header
grep -n "EOF_MARKER\|byte\[\] HEADER\|writeKVPair" \
  tez-runtime-library/src/main/java/org/apache/tez/runtime/library/common/sort/impl/IFile.java

# Which sorter file is larger, and why?
wc -l tez-runtime-library/src/main/java/org/apache/tez/runtime/library/common/sort/impl/PipelinedSorter.java \
      tez-runtime-library/src/main/java/org/apache/tez/runtime/library/common/sort/impl/dflt/DefaultSorter.java

# Every read site for the sort buffer knobs
grep -rn "TEZ_RUNTIME_IO_SORT_MB\|TEZ_RUNTIME_SORT_SPILL_PERCENT" tez-runtime-library/src/main/java

# The exact shuffle URL construction
grep -n "mapOutput?job=\|&reduce=\|&map=" \
  tez-runtime-library/src/main/java/org/apache/tez/runtime/library/common/shuffle/ShuffleUtils.java

# Where the combiner runs
grep -n "combiner\|runCombineProcessor" \
  tez-runtime-library/src/main/java/org/apache/tez/runtime/library/common/sort/impl/dflt/DefaultSorter.java

Answer:

Verify EOF_MARKER == -1 and the HEADER bytes. What breaks if a reader sees the wrong header?
Which sorter's source file is larger, and what does that tell you about their relative complexity?
Which input class imports the orderedgrouped package and which imports impl? What functional difference follows?
Read constructBaseURIForShuffleHandler. Reconstruct the full URL for partition 7 of a single-partition request.
At what phase does the combiner run, and why does that force associativity?
Find where isLocalFetch is computed. Under exactly what three conditions does a fetch skip HTTP?

Common bugs and symptoms

Symptom	Likely cause
`Fetcher: java.net.ConnectException`	aux service (`mapreduce_shuffle`/`tez_shuffle`) not configured or NM down
`Premature EOF from inputStream`	source wrote a partial `IFile` (killed mid-spill); consumer retries another attempt
OOM during sort	`tez.runtime.io.sort.mb` too high vs container heap; remember `DefaultSorter` caps at 1800 MB
OOM during shuffle	`shuffle.fetch.buffer.percent` too high, or one input under `memory.limit.percent` starves heap
Wrong reducer output count	combiner not associative/idempotent across merge passes
`OnDiskMerger` thrashing	`io.sort.factor` too low → many tiny segments → many merge passes
Long shuffle plateau, rising `FailedShuffleCounter`	fetch-failure storm from a dead source NM; check penalty queue and re-run activity
`maxSingleShuffleLimit should be less than mergeThreshold`	`memory.limit.percent` set so high a single input exceeds the merge threshold

Validation: prove you understand this

Sketch the byte layout of an IFile segment with 3 records in one partition. Show each keyLen/valLen VInt and the EOF_MARKER. Note where the HEADER and checksum sit.
A reducer reads from 200 mappers with parallel.copies=20 and fetch.max.task.output.at.once=20. Compute the minimum number of HTTP requests the fetcher pool must issue. Justify.
Explain why PipelinedSorter cuts wall-clock time but not CPU time, in terms of the sortmaster thread pool and the overlap it creates.
A 10 GB shuffle lands in a 4 GB-heap reducer with fetch.buffer.percent=0.90 and memory.limit.percent=0.25. Compute the single-input memory cap and predict which inputs go to disk.
Using only grep, find the exact file and method where the ?reduce=&map= URL is constructed on the Tez fetcher side. Cite the module path.
Correct the statement "Tez has no NodeManager shuffle service." Name Tez's handler class, its service id, its default port, and the config key that selects it. Cross-check against yarn-integration.md.

Open-Source Engineer & Contributor