Lab 3.2: Understand the IPO Abstraction

Background

Every task in Tez is the same shape: a Processor reads from zero or more Inputs and writes to zero or more Outputs. That is the Input–Processor–Output (IPO) model, and it is the one abstraction you must own before you can read — let alone change — any runtime code in Tez. A join is a processor with two inputs. A shuffle is what an ordered output and an ordered input do to each other across an edge. A source is a processor whose only "input" is a data source. Learn IPO and the whole runtime stops looking like magic.

This lab is hands-on but not a build-from-scratch lab — that is Lab 3.3. Here you will (1) read the real IPO contracts in tez-api and quote their signatures, then (2) dissect one shipped example end-to-end — OrderedWordCount from tez-examples — line by line: how it wires processors, inputs, outputs, and edges; the initialize/start/getReader/ getWriter lifecycle as that example experiences it; and the memory handshake basics. Then you will modify it: change the partitioner, add a combiner — using the real builder methods, verified against source.

Everything here is verified against a current Apache Tez master checkout. Names are stable; line numbers are not, so you get grep locators. Companion deep dive: ipo-abstractions.md; the submission path that gets you here is Lab 3.1; the Level 3 overview frames both.

Why This Lab Matters for Contributors

IPO is the seam where most Tez contributions live. A new input format, a smarter combiner, a different partitioning strategy, a memory-pressure fix in the sorter — all of them are IPO changes. And IPO is where the subtle bugs hide: an output that is written to before start(), a reader fetched twice, a processor that assumes a single input when the edge routing gave it several. If you cannot recite the lifecycle — who calls initialize, who calls start, when is getWriter legal — you cannot review those patches, and you certainly cannot write one. This lab makes the contract concrete against code that ships in the release.

Prerequisites

• Apache Tez cloned and building; tez-examples compiled.
• You have completed Lab 3.1 — you know how a processor's run(inputs, outputs) gets called (LogicalIOProcessorRuntimeTask.run).
• A scratch file:

mkdir -p ~/tez-notes && : > ~/tez-notes/ipo-3.2.md
export TEZ=~/src/tez && cd "$TEZ"

• Optional but recommended: the ability to run OrderedWordCount in local mode so you can watch the counters it emits.

The IPO Contracts (read these first)

Three abstract base classes in tez-api are the entire user-facing contract. Open all three:

ls tez-api/src/main/java/org/apache/tez/runtime/api/Abstract*.java

AbstractLogicalIOProcessor

tez-api/.../runtime/api/AbstractLogicalIOProcessor.java. The processor you subclass. Its framework-facing run comes from LogicalIOProcessorFrameworkInterface:

public void run(Map<String, LogicalInput> inputs,
    Map<String, LogicalOutput> outputs) throws Exception;

From the interface's own javadoc: inputs is "a map of the source vertex name to LogicalInput — one per incoming edge," and outputs is "a map of the destination vertex name to LogicalOutput — one per outgoing edge." Memorize that keying; it is the crux of the multi-input work in Lab 3.3, and it is exactly what you verified in Lab 3.1 (runInputMap.put(inputSpec.getSourceVertexName(), input)).

The lifecycle methods it forces you to implement come from ProcessorFrameworkInterface: initialize(), handleEvents(List<Event>), close(), plus abort(). Note getContext() returns a ProcessorContext — your handle to counters, user payload, and the memory API.

AbstractLogicalInput

tez-api/.../runtime/api/AbstractLogicalInput.java. Note its two-argument constructor contract, quoted from source:

public AbstractLogicalInput(InputContext inputContext, int numPhysicalInputs) {
    this.inputContext = inputContext;
    this.numPhysicalInputs = numPhysicalInputs;
}
public abstract List<Event> initialize() throws Exception;
public final int getNumPhysicalInputs() { return numPhysicalInputs; }

Two things to write down. First, initialize() returns List<Event> — an input can emit events during init (for example to request data). Second, numPhysicalInputs is "typically determined by Edge routing, and number of upstream tasks" — a single logical input can wrap many physical inputs (one per upstream task). The start() / getReader() methods come from the Input interface.

AbstractLogicalOutput

tez-api/.../runtime/api/AbstractLogicalOutput.java. Symmetric:

public AbstractLogicalOutput(OutputContext outputContext, int numPhysicalOutputs) { ... }
public abstract List<Event> initialize() throws Exception;
public final int getNumPhysicalOutputs() { return numPhysicalOutputs; }

The start() / getWriter() methods come from the Output interface. Read that interface's javadoc on start() — it is explicit that the processor is responsible for starting outputs, and that start() "must be written to handle multiple start invocations — typically honoring only the first one." The same is true for Input.start().

The lifecycle, stated exactly

Put the pieces together and record this ordering in your notes — it is the contract every processor depends on:

Tez framework (LogicalIOProcessorRuntimeTask):
  1. constructs the Input/Output with (context, numPhysical...)
  2. calls initialize() on each Input, Output, and the Processor
     -> initialize() may return events; framework routes them
  3. calls processor.run(inputs, outputs)

Inside your Processor.run(...):
  4. YOU call input.start() / output.start()   (idempotent; honor first call)
  5. YOU call input.getReader() / output.getWriter()
  6. read/process/write
Tez framework, after run() returns:
  7. close() on outputs (commit), inputs, processor

The rule that trips people up: getReader()/getWriter() before start() is illegal. The example below relies on this, and Lab 3.3 makes you break it on purpose to see the exception.

Step-by-Step Tasks — Dissect OrderedWordCount

Open tez-examples/src/main/java/org/apache/tez/examples/OrderedWordCount.java. This example extends plain word count by sorting the words by their count, so it has three vertices and two edges — enough to show every IPO idea. Its topology:

Tokenizer ──(Text, IntWritable)──▶ Summation ──(IntWritable, Text)──▶ Sorter ──▶ file
 (TokenProcessor)                   (SumProcessor)                     (NoOpSorter)

Step 1 — The three processors and how they touch IPO

OrderedWordCount reuses WordCount.TokenProcessor and defines SumProcessor and NoOpSorter as inner classes. Read all three and record, for each, how many inputs and outputs it assumes and how it names them.

SumProcessor extends SimpleProcessor. Its run() is the clearest IPO example in the tree:

public void run() throws Exception {
  Preconditions.checkArgument(getInputs().size() == 1);
  Preconditions.checkArgument(getOutputs().size() == 1);
  KeyValueWriter kvWriter = (KeyValueWriter) getOutputs().get(SORTER).getWriter();
  KeyValuesReader kvReader = (KeyValuesReader) getInputs().get(TOKENIZER).getReader();
  while (kvReader.next()) {
    Text word = (Text) kvReader.getCurrentKey();
    int sum = 0;
    for (Object value : kvReader.getCurrentValues()) {
      sum += ((IntWritable) value).get();
    }
    kvWriter.write(new IntWritable(sum), word);   // count as key, word as value
  }
}

Four things to note and write down:

1. Inputs/outputs are fetched by name. getInputs().get(TOKENIZER) — TOKENIZER is the upstream vertex name. getOutputs().get(SORTER) — SORTER is the downstream vertex name. This is the exact keying from the contract above. (TOKENIZER and SORTER are static String fields set to the vertex names.)
2. It casts the reader/writer, not the input/output. The comment in the source spells out why: "the recommended approach is to cast the reader/writer to a specific type instead of casting the input/output. This allows the actual input/output type to be replaced without affecting the semantic guarantees ... represented by the reader and writer." Follow this idiom in your own code.
3. It reads with a KeyValuesReader (plural Values) — because the incoming edge is an ordered grouped edge, so all values for a key arrive grouped. Contrast with a KeyValueReader (singular) used on ungrouped edges.
4. SumProcessor writes count-as-key so the next edge can sort by count. The IPO shape of a vertex is chosen to serve the edge it feeds.

NoOpSorter extends SimpleMRProcessor does no work at all — it copies its grouped input to its output. It can be a no-op because the edge already sorted and grouped the data. Read its run() and confirm it only rewrites (word, sum); the ordering is a property of the edge, not the code.

SimpleProcessor / SimpleMRProcessor are thin conveniences over AbstractLogicalIOProcessor that call start() on all inputs/outputs for you and expose a no-arg run(). Confirm:

grep -n "start()\|abstract.*run\|getOutputs\|getInputs" \
  tez-runtime-library/src/main/java/org/apache/tez/runtime/library/processor/SimpleProcessor.java

This is why SumProcessor.run() never calls start() itself — the base class did. Your own raw AbstractLogicalIOProcessor (Lab 3.3) must call start() explicitly.

Step 2 — The edges: OrderedPartitionedKVEdgeConfig

Now the wiring. Read createDAG(...). Each edge is built from an OrderedPartitionedKVEdgeConfig:

OrderedPartitionedKVEdgeConfig summationEdgeConf = OrderedPartitionedKVEdgeConfig
    .newBuilder(Text.class.getName(), IntWritable.class.getName(),
        HashPartitioner.class.getName())
    .setFromConfiguration(tezConf)
    .build();

Read the three arguments to newBuilder — they are, in order, key class, value class, partitioner class. Confirm in the source:

grep -n "public static Builder newBuilder" \
  tez-runtime-library/src/main/java/org/apache/tez/runtime/library/conf/OrderedPartitionedKVEdgeConfig.java

The Tokenizer→Summation edge carries (Text, IntWritable) — words and their partial counts, hashed by word so identical words meet at one Summation task. The Summation→Sorter edge carries (IntWritable, Text) — count as key, so ordering by count is what the edge does for free.

Then the edge property itself:

Edge.create(tokenizerVertex, summationVertex, summationEdgeConf.createDefaultEdgeProperty())

createDefaultEdgeProperty() is the factory that turns the config into a concrete EdgeProperty. Read it:

grep -n "createDefaultEdgeProperty\|DataMovementType\|SCATTER_GATHER" \
  tez-runtime-library/src/main/java/org/apache/tez/runtime/library/conf/OrderedPartitionedKVEdgeConfig.java

It produces a SCATTER_GATHER edge whose output descriptor is OrderedPartitionedKVOutput and whose input descriptor is OrderedGroupedKVInput. That is the payoff of IPO: the config picks which Output class runs on the source side and which Input class runs on the destination side. SumProcessor never names those classes — it just casts the reader/writer.

Step 3 — Build the IPO Map

Fill this in by reading OrderedWordCount.createDAG and createDefaultEdgeProperty:

# Verify the classes the config wires in:
grep -rn "OrderedPartitionedKVOutput\|OrderedGroupedKVInput" \
  tez-runtime-library/src/main/java/org/apache/tez/runtime/library/conf/OrderedPartitionedKVEdgeConfig.java

Step 4 — The write path: OrderedPartitionedKVOutput

When SumProcessor calls kvWriter.write(count, word), where do the bytes go? Open tez-runtime-library/src/main/java/org/apache/tez/runtime/library/output/OrderedPartitionedKVOutput.java and read start() and how it creates its sorter:

grep -n "initialize\|start\|ExternalSorter\|getWriter\|sorter" \
  tez-runtime-library/src/main/java/org/apache/tez/runtime/library/output/OrderedPartitionedKVOutput.java | head

Confirm: getWriter() returns a writer backed by an ExternalSorter (the PipelinedSorter or the default sorter). The write path is: KeyValueWriter.write → sorter buffer → spill to IFile on local disk → merge on close. You do not need to read the sorter internals now; you need to see that the output class owns sorting, which is why the processor can stay dumb.

Note the partitioner: the output partitions each record by key using the HashPartitioner you named in newBuilder. That is why all counts of a given word land in the same downstream partition.

Step 5 — The read path: OrderedGroupedKVInput

On the destination side, open tez-runtime-library/src/main/java/org/apache/tez/runtime/library/input/OrderedGroupedKVInput.java:

grep -n "initialize\|start\|Shuffle\|getReader\|memory" \
  tez-runtime-library/src/main/java/org/apache/tez/runtime/library/input/OrderedGroupedKVInput.java | head

Confirm the shape: initialize() sets up the shuffle machinery; start() kicks off fetching; getReader() returns a KeyValuesReader that merges the fetched, sorted partitions and groups values by key. This is why SumProcessor gets a KeyValuesReader and can iterate getCurrentValues() for each key — the input class did the sort-merge-group, not the processor.

Step 6 — The memory handshake basics

Sorting outputs and shuffling inputs both need large buffers, and many run in the same JVM. Tez does not let each grab memory blindly — it brokers memory through the context. Look at how an IPO member requests memory:

grep -rn "requestInitialMemory\|MemoryUpdateCallback\|getTotalMemoryAvailableToTask" \
  tez-runtime-library/src/main/java/org/apache/tez/runtime/library/output/OrderedPartitionedKVOutput.java \
  tez-runtime-library/src/main/java/org/apache/tez/runtime/library/input/OrderedGroupedKVInput.java | head

The pattern, which you should record: during initialize(), an input/output calls getContext().requestInitialMemory(size, callback). The framework's memory distributor decides how much each requester actually gets and invokes the MemoryUpdateCallback with the grant. Read the callback interface:

grep -n "memoryAssigned" tez-api/src/main/java/org/apache/tez/runtime/api/MemoryUpdateCallback.java

So the memory a sorter buffer or shuffle buffer uses is negotiated, not hard-coded — which is why two IPO members in one task JVM don't blow the heap. That negotiation happening in initialize() (before start()) is another reason the lifecycle ordering is strict.

Step 7 — Trace one record end-to-end

Tie it together in prose in your notes. Follow a single word, say "tez", from Tokenizer to file:

1. TokenProcessor reads a line from its MRInput (getInputs().get(INPUT) — a data source, keyed by the source name, not a vertex) and writes ("tez", 1) to its OrderedPartitionedKVOutput.
2. The output hashes "tez" by HashPartitioner, sorts, and spills to IFile.
3. SumProcessor's OrderedGroupedKVInput shuffles the partition holding "tez", merges, and presents ("tez", [1,1,1,...]) to the KeyValuesReader.
4. SumProcessor sums to ("tez", 12) then writes (12, "tez") — count as key — to its output.
5. The Summation→Sorter edge sorts by count. NoOpSorter copies (12, "tez") → ("tez", 12) to MROutput, which writes the final file.

Every hop is an IPO boundary. The processors are small; the input/output classes carry the weight.

Reader Exercises — Modify the Wiring

Now change the example using real builder methods. Make a scratch copy of OrderedWordCount under a new class name so you don't disturb the shipped example.

Exercise A — Swap the partitioner

The third newBuilder argument is the partitioner class name. Change the Summation edge to use a different partitioner and observe the effect on which task gets which key.

# Confirm the available partitioners in the library:
ls tez-runtime-library/src/main/java/org/apache/tez/runtime/library/partitioner/

Replace HashPartitioner.class.getName() with a partitioner of your own. Confirm the contract you must implement by reading the shipped one:

grep -rn "int getPartition" \
  tez-runtime-library/src/main/java/org/apache/tez/runtime/library/partitioner/HashPartitioner.java

getPartition(key, value, numPartitions) returning an int in [0, numPartitions) is the whole contract. Write one that sends short words to partition 0 and long words to partition 1 (with the destination vertex parallelism set to 2), rebuild, and confirm from counters/output that the split happened. Record what changed and what did not (the sum is still correct — partitioning changes placement, not values).

Exercise B — Add a combiner

A combiner runs map-side, on the output, to pre-aggregate before the shuffle. It is set via the output side of the edge config's fluent builder. Find the method — it is on the output specific builder, reached via configureOutput():

grep -n "configureOutput\|setCombiner" \
  tez-runtime-library/src/main/java/org/apache/tez/runtime/library/conf/OrderedPartitionedKVEdgeConfig.java \
  tez-runtime-library/src/main/java/org/apache/tez/runtime/library/conf/OrderedPartitionedKVOutputConfig.java

You will find configureOutput() returns an OrderedPartitionedKVOutputConfig.SpecificBuilder, which declares:

public SpecificBuilder<E> setCombiner(String combinerClassName);
public SpecificBuilder<E> setCombiner(String combinerClassName, Map<String, String> combinerConf);

Wire a combiner onto the Tokenizer→Summation edge like so (fill in a real combiner class name):

OrderedPartitionedKVEdgeConfig summationEdgeConf = OrderedPartitionedKVEdgeConfig
    .newBuilder(Text.class.getName(), IntWritable.class.getName(),
        HashPartitioner.class.getName())
    .setFromConfiguration(tezConf)
    .configureOutput()
        .setCombiner(MyIntSumCombiner.class.getName())
        .done()          // return to the parent Builder
    .build();

Confirm the done() return-to-parent method exists on the specific builder (that is how these nested fluent builders hand control back). Then check what a combiner must implement:

find tez-runtime-library/src/main/java -name "Combiner.java"
grep -rn "combine(" tez-runtime-library/src/main/java/org/apache/tez/runtime/library/api/ 2>/dev/null | head

Rebuild and confirm from counters that fewer bytes crossed the shuffle (compare OUTPUT_BYTES/SHUFFLE_BYTES-family counters before and after). Record the delta — this is the observable payoff of a combiner and exactly the kind of before/after evidence a committer puts in a PR description.

Exercise C — Set a custom key comparator (bonus)

Ordering is a property of the ordered edge, controlled by the key comparator. The edge config exposes it directly on the top-level builder:

grep -n "setKeyComparatorClass" \
  tez-runtime-library/src/main/java/org/apache/tez/runtime/library/conf/OrderedPartitionedKVEdgeConfig.java

Note it sets the comparator on both the input and output builders (read the body — it calls outputBuilder.setKeyComparatorClass(...) and inputBuilder.setKeyComparatorClass(...)). Set a reverse comparator on the Summation→Sorter edge and confirm the output file's ordering flips.

Deliverables

• Quoted signatures for AbstractLogicalIOProcessor.run, AbstractLogicalInput/Output constructors and initialize(), and the Input/Output start()/getReader()/getWriter() methods, in ~/tez-notes/ipo-3.2.md.
• The written-out IPO lifecycle ordering (who calls initialize, start, getWriter, close).
• The completed IPO Map table for OrderedWordCount.
• The end-to-end trace of one word (Step 7), in your own words.
• Exercise A: a working custom partitioner + a note on what changed vs. what stayed correct.
• Exercise B: a combiner wired via configureOutput().setCombiner(...) + before/after shuffle byte counts.

Troubleshooting

Stretch Goals

1. UnorderedKVOutput/UnorderedKVInput. Find the unordered edge config and compare it to the ordered one. When would you choose an unordered (broadcast-style) edge over SCATTER_GATHER?

   ls tez-runtime-library/src/main/java/org/apache/tez/runtime/library/conf/ | grep -i unordered
   

2. DataMovementEvent. The output tells the input where its data is via a DataMovementEvent. Find the class and record what it carries and why its payload is an opaque byte array.

   find . -path '*/api/events/DataMovementEvent.java' | grep -v test
   

3. EdgeManagerPlugin. Read the interface; note the methods a custom edge manager must implement and when you would use one instead of SCATTER_GATHER.

   find tez-api/src/main/java -name "EdgeManagerPlugin.java"
   

4. Broadcast in a join. Read HashJoinExample in tez-examples. It broadcasts the small side. Explain, in IPO terms, why a broadcast input is the right shape for the hash side of a join.

Validation / Self-check

Answer without looking back:

1. In Processor.run(inputs, outputs), what is each map keyed by, and where in the runtime is that key assigned?
2. Why does SumProcessor cast getReader()/getWriter() rather than casting the Input/Output object it fetched? Quote the reason from the source comment.
3. State the IPO lifecycle ordering: which of initialize, start, getWriter, close does the framework call, and which does your processor call?
4. OrderedWordCount's SumProcessor writes the count as the key. Why — what does that buy the next edge?
5. Which class actually sorts and spills the output, and which class shuffles and merges the input? Name both, and explain why the processor can stay ignorant of them.
6. Where do a sorter buffer and a shuffle buffer in the same task JVM get their memory, and which lifecycle method performs the request?
7. To add a combiner, which builder method do you call and on which side of the edge? To change key ordering, which method, and why does it touch both input and output builders?

When you can answer all seven and both modification exercises run green, you have completed Lab 3.2. Continue to Lab 3.3: Build It — Multi-Input Union DAG.