Lab 7.1 — Debug Shuffle Behavior

Lab type: Reproduce & Instrument Estimated time: 150 min Tez module: tez-runtime-library

Background

Shuffle is where Tez moves data between vertices, and it is the single most common source of Tez bug reports. To debug it you must be able to watch it happen: run a shuffle-heavy DAG, turn up the right log categories, read the fetch/merge/spill log lines back to the source that prints them, and correlate them with the shuffle counters. Then you must be able to break it on purpose — Tez ships a fault injector precisely so you can — and follow the failure up to the AM.

In this lab you run OrderedWordCount (a real three-vertex example from tez-examples) in local mode, instrument its shuffle, read the counters, induce a deterministic fetch failure, and tune the sort buffer while watching the spill counters respond. Every log category, counter, and config key below is quoted from the Tez checkout with a grep so you can confirm it.

Why This Lab Matters for Contributors

"Shuffle is slow" and "I get fetch failures" are the two most common Tez issues, and both are un-triageable without evidence: which counter is high, which log line fired, which class emitted it. A maintainer who can attach SHUFFLE_BYTES, ADDITIONAL_SPILL_COUNT, and a FetcherOrderedGrouped log line to a report turns a vague complaint into a fixable bug. This lab teaches you to generate that evidence on demand — the exact skill a shuffle PR's "how I tested this" section requires.

Prerequisites

• Level 7 index read; you can draw the shuffle path.
• Tez checkout built (mvn -q -DskipTests install).
• The tez-examples jar built (it is a module of the Tez build).
• Hadoop client jars on your classpath for local-mode runs.
• Deep dive: shuffle and sort skimmed.

Step-by-Step Tasks

Step 1 — Confirm the example and its shuffle shape

OrderedWordCount is a real example. Confirm it and its structure:

# The class and its vertices:
grep -n "class OrderedWordCount\|Vertex.create\|Edge.create" \
  tez-examples/src/main/java/org/apache/tez/examples/OrderedWordCount.java

# It's registered in the example driver under the name "orderedwordcount":
grep -n "orderedwordcount\|OrderedWordCount" \
  tez-examples/src/main/java/org/apache/tez/examples/ExampleDriver.java

You will see three vertices — Tokenizer, Summation, Sorter — joined by two OrderedPartitionedKVEdgeConfig edges. That means two shuffles: tokenizer → summation, and summation → sorter. This is exactly the shuffle-heavy DAG you want.

Step 2 — Run it in local mode

TezExampleBase supports a -local flag that sets tez.local.mode=true — no YARN cluster required. Confirm the flag, then run:

# Confirm local mode is a real, supported flag:
grep -n "LOCAL_MODE\|local mode\|TEZ_LOCAL_MODE" \
  tez-examples/src/main/java/org/apache/tez/examples/TezExampleBase.java

# Make a tiny input with repeated words (so shuffle actually groups):
printf 'the cat sat\nthe dog ran\nthe cat ran\n' > /tmp/words.txt

# Run OrderedWordCount in local mode via the example driver jar.
# (Adapt the jar path/version to your build under tez-examples/target/.)
export HADOOP_CLASSPATH=$(find . -name 'tez-*.jar' | tr '\n' ':')
java -cp "$HADOOP_CLASSPATH" org.apache.tez.examples.ExampleDriver \
  orderedwordcount -local -counter file:///tmp/words.txt file:///tmp/wc-out

Expected: the job succeeds and /tmp/wc-out contains word counts. The -counter flag (also from TezExampleBase) prints the DAG counters at the end — you will read those in Step 4.

Note: If you prefer the book's own DAG, the local-mode pattern is the same one used by book/projects/level-3-multi-input (TEZ_LOCAL_MODE=true, fs.defaultFS=file:///). OrderedWordCount is used here because it has two real ordered shuffles, which the multi-input project's single hop does not stress as hard.

Step 3 — Turn up shuffle logging

The shuffle classes log under their fully-qualified class names, so the log4j categories are the packages themselves. Set these to DEBUG/INFO in the log4j.properties (or -Dlog4j... for local mode):

# Consumer side — fetch, schedule, merge:
log4j.logger.org.apache.tez.runtime.library.common.shuffle.orderedgrouped.FetcherOrderedGrouped=DEBUG
log4j.logger.org.apache.tez.runtime.library.common.shuffle.orderedgrouped.ShuffleScheduler=INFO
log4j.logger.org.apache.tez.runtime.library.common.shuffle.orderedgrouped.MergeManager=INFO

# Producer side — sort and spill:
log4j.logger.org.apache.tez.runtime.library.common.sort.impl.PipelinedSorter=INFO
log4j.logger.org.apache.tez.runtime.library.common.sort.impl.dflt.DefaultSorter=INFO

Confirm these are the real loggers before trusting them:

grep -rn "LoggerFactory.getLogger" \
  tez-runtime-library/src/main/java/org/apache/tez/runtime/library/common/shuffle/orderedgrouped/FetcherOrderedGrouped.java \
  tez-runtime-library/src/main/java/org/apache/tez/runtime/library/common/shuffle/orderedgrouped/ShuffleScheduler.java \
  tez-runtime-library/src/main/java/org/apache/tez/runtime/library/common/shuffle/orderedgrouped/MergeManager.java

Step 4 — Read the fetch, merge, and spill log lines against the source

Rerun with logging up and capture the output. You are matching real log lines to the code that prints them. Here are the exact log strings to look for, each quoted from its class in the checkout:

Fetch progress — from ShuffleScheduler.logProgress:

copy(<N> (spillsFetched=<n>) of <total>. Transfer rate (CumulativeDataFetched/TimeSinceInputStarted)) <x> MB/s)

Fetcher waiting for merge memory — from FetcherOrderedGrouped:

fetcher#<id> - MergerManager returned Status.WAIT ...

In-memory merge kicking off — from MergeManager:

Initiating in-memory merge with <noInMemorySegments> segments...

Spill — from DefaultSorter (or a Spilling to <file> line from PipelinedSorter):

<vertex>: Spilling map output. bufstart=... bufend=... ; kvstart=... kvend=... length=.../<maxRec>

Verify a couple against source so you know they are not invented:

grep -n "MergerManager returned Status.WAIT" \
  tez-runtime-library/src/main/java/org/apache/tez/runtime/library/common/shuffle/orderedgrouped/FetcherOrderedGrouped.java
grep -n "Initiating in-memory merge with" \
  tez-runtime-library/src/main/java/org/apache/tez/runtime/library/common/shuffle/orderedgrouped/MergeManager.java
grep -n "Spilling map output" \
  tez-runtime-library/src/main/java/org/apache/tez/runtime/library/common/sort/impl/dflt/DefaultSorter.java
grep -n "Transfer rate" \
  tez-runtime-library/src/main/java/org/apache/tez/runtime/library/common/shuffle/orderedgrouped/ShuffleScheduler.java

Build a table: log line → class → method → what it means. That table is the Rosetta Stone for every future shuffle bug you triage.

Step 5 — Read the shuffle counters

With -counter (Step 2) the DAG counters print. These are real TaskCounter enum members — verify the full set:

grep -nE "SPILLED_RECORDS|NUM_SHUFFLED_INPUTS|NUM_FAILED_SHUFFLE_INPUTS|MERGED_MAP_OUTPUTS|ADDITIONAL_SPILL_COUNT|ADDITIONAL_SPILLS_BYTES_WRITTEN|ADDITIONAL_SPILLS_BYTES_READ|SHUFFLE_BYTES|SHUFFLE_BYTES_TO_MEM|SHUFFLE_BYTES_TO_DISK|SHUFFLE_BYTES_DECOMPRESSED|NUM_MEM_TO_DISK_MERGES|NUM_DISK_TO_DISK_MERGES|SHUFFLE_PHASE_TIME|MERGE_PHASE_TIME" \
  tez-api/src/main/java/org/apache/tez/common/counters/TaskCounter.java

Record, for the Summation and Sorter (reduce-side) vertices:

NUM_FAILED_SHUFFLE_INPUTS should be 0 on a clean run — you are about to make it non-zero.

Step 6 — Induce a fetch failure deterministically

Tez has a built-in fetch-failure injector so you don't have to yank cables. Confirm it, then turn it on:

# The two config keys (enable flag + failure spec):
grep -nE "shuffle.fetch.testing.errors" \
  tez-runtime-library/src/main/java/org/apache/tez/runtime/library/api/TezRuntimeConfiguration.java

# The class that decides whether a given fetch should fail:
grep -n "shouldFail\|probabilityPercent" \
  tez-runtime-library/src/main/java/org/apache/tez/runtime/library/common/shuffle/FetcherErrorTestingConfig.java

• tez.runtime.shuffle.fetch.testing.errors.enable — turn on the injector (default false).
• tez.runtime.shuffle.fetch.testing.errors.config — the failure spec, format maphost#mapvertex#probability#features (default *#50). For example *#*#100 fails every fetch from every host; *#Summation#100 fails only fetches whose source vertex is Summation.

Rerun with, e.g., -Dtez.runtime.shuffle.fetch.testing.errors.enable=true -Dtez.runtime.shuffle.fetch.testing.errors.config='*#*#100'. In the log you will see the injector announce itself and throw:

Initialized FetcherOrderedGroupedWithInjectableErrors with config: [FetcherErrorTestingConfig: ...]
FetcherOrderedGroupedWithInjectableErrors tester made failure for host: <h>, input attempt: <n>

Then follow the failure upward:

1. NUM_FAILED_SHUFFLE_INPUTS increments.
2. The reduce task raises an InputReadErrorEvent — confirm the class:

   grep -rn "InputReadErrorEvent" \
     tez-runtime-internals/src/main/java/org/apache/tez/runtime/api/events/InputReadErrorEvent.java
   

3. The AM's TaskAttemptImpl receives it and blames the producer attempt; confirm the blame logic exists:

   grep -n "blamed for read error\|downstreamBlamingHosts\|tooManyDownstreamHostsBlamed" \
     tez-dag/src/main/java/org/apache/tez/dag/app/dag/impl/TaskAttemptImpl.java
   

4. If failures exceed tez.runtime.shuffle.fetch.failures.limit (default 5) the producer output is declared lost and its map attempt may be re-run.

Write down the chain: injector throws → NUM_FAILED_SHUFFLE_INPUTS++ → InputReadErrorEvent → AM blames producer → re-run or fail. This is the canonical fetch-failure story every bug report is a variation of.

Step 7 — Tune the sort buffer and watch spills respond

The producer sort buffer is tez.runtime.io.sort.mb (default 100). A smaller buffer forces more spills; a larger one fewer. Confirm the key and default:

grep -n "TEZ_RUNTIME_IO_SORT_MB\b\|io.sort.mb" \
  tez-runtime-library/src/main/java/org/apache/tez/runtime/library/api/TezRuntimeConfiguration.java

Run three times, low → default → higher, and record the spill counters each time (use a bigger input so spills actually happen — repeat /tmp/words.txt a few thousand times):

You should see spill counts fall as io.sort.mb rises — until you exceed the task heap, at which point you get an OOM instead. Explain that trade-off in one sentence; it is the exact reasoning a tuning PR must include.

Deliverables

• A successful local-mode OrderedWordCount run with -counter output saved.
• The log line → class → method → meaning table (Step 4), each line confirmed against source.
• The shuffle-counter table (Step 5) from a clean run.
• The fetch-failure chain (Step 6): injector config used, the failure log line, and the InputReadErrorEvent → AM-blame path.
• The io.sort.mb tuning table (Step 7) showing spill counters responding, plus your trade-off sentence.

Troubleshooting

Stretch Goals

1. Compare the two sorters. Set tez.runtime.sorter.class=DefaultSorter and rerun; diff the spill log lines and counters against the default PipelinedSorter. Note which class emits which spill message.
2. Fail one vertex only. Use *#Summation#100 so only the summation→sorter fetch fails. Confirm from NUM_FAILED_SHUFFLE_INPUTS per vertex that the other shuffle stayed clean — a precision the FetcherErrorTestingConfig spec makes possible.
3. Read MergeManager's memory math. Find the MergerManager: memoryLimit=... line and trace back the fraction of task heap given to the shuffle buffer (tez.runtime.shuffle.fetch.buffer.percent, default 0.90). Predict the merge counters and check.

Validation

1. Which class prints the copy(... Transfer rate ...) progress line, and what does spillsFetched count?
2. Where does the sorter run — producer or consumer — and which counter tells you it spilled?
3. Name three TaskCounter members that move when data crosses a shuffle edge, and say which increases on a failed fetch.
4. What two config keys turn on the fetch-failure injector, and what does the spec *#Summation#100 mean?
5. When a fetch fails, what event does the reduce task raise, and what does the AM do to the producer attempt?
6. What config controls how many fetch failures are tolerated before the producer output is declared lost, and what is its default?
7. You raised tez.runtime.io.sort.mb and spills fell — what stops you from raising it arbitrarily high?

When you can generate and read this evidence on demand, you can debug the shuffle; move to Lab 7.2 — Modify a Processor: Add Deduplication, where you write runtime-library code yourself. For the theory, revisit Deep dive: shuffle and sort.