IPO Abstractions

Input, Processor, Output — collectively "IPO" — are the three runtime contracts every Tez task is assembled from. A task is exactly one processor, zero or more inputs, and zero or more outputs, wired together by the framework inside a single JVM. The processor is the only place your business logic runs; the inputs and outputs are the framework's plug points for moving bytes across edges. Everything Hive, Pig, and Cascading run on Tez ultimately bottoms out in these interfaces.

This chapter dissects the real interface hierarchy in tez-api, the split between the user-facing and framework-facing halves of each contract, the exact lifecycle ordering enforced by LogicalIOProcessorRuntimeTask, the memory-request handshake, the event flow between an IPO and the AM, and the shipped catalog of inputs/outputs/processors you will subclass or configure in practice.

After this chapter you can open any concrete IPO class in tez-runtime-library, name every method the framework calls on it and in what order, and write a new LogicalInput or LogicalOutput that wires cleanly into an edge.

The interface hierarchy

The single most confusing thing about IPO for newcomers is that each contract is split into two interfaces: a tiny user-facing one and a *FrameworkInterface one. Start by listing them.

cd tez-api/src/main/java/org/apache/tez/runtime/api
ls Input.java Output.java Processor.java \
   LogicalInput.java LogicalOutput.java LogicalIOProcessor.java \
   InputFrameworkInterface.java OutputFrameworkInterface.java ProcessorFrameworkInterface.java \
   AbstractLogicalInput.java AbstractLogicalOutput.java AbstractLogicalIOProcessor.java \
   MergedLogicalInput.java
grep -n "public interface\|public abstract class\|extends" Input.java LogicalInput.java \
   InputFrameworkInterface.java LogicalInputFrameworkInterface.java AbstractLogicalInput.java

Here is what you will find, for the input side (output and processor mirror it):

Why the split? The two halves have different callers. start() and getReader() are called by the processor (your code, processor.run(...)). initialize(), handleEvents(), and close() are called by the framework (LogicalIOProcessorRuntimeTask). Tez keeps them in separate interfaces so the processor's view of an input (Input) is narrower than the framework's view — the processor literally cannot call close() on its inputs because that method is not on the type it holds.

// tez-api, org.apache.tez.runtime.api.Input
public interface Input {
  public void start() throws Exception;
  public Reader getReader() throws Exception;
}

// tez-api, org.apache.tez.runtime.api.InputFrameworkInterface
public interface InputFrameworkInterface {
  public List<Event> initialize() throws Exception;
  public void handleEvents(List<Event> inputEvents) throws Exception;
  public List<Event> close() throws Exception;
}

Note: LogicalInput, LogicalOutput, and LogicalIOProcessor are empty marker interfaces. Do not go looking for methods on them — run grep -c "public" LogicalInput.java and you will get zero declarations. The word "Logical" means "the processor sees one of these per edge, regardless of how many physical connections back it." Contrast with the physical inputs discussed in logical-physical.md.

AbstractLogicalInput — the base you extend

You almost never implement LogicalInput directly. You extend AbstractLogicalInput, which stores the context and the physical-input count for you.

sed -n '40,90p' tez-api/src/main/java/org/apache/tez/runtime/api/AbstractLogicalInput.java

// tez-api, org.apache.tez.runtime.api.AbstractLogicalInput
public abstract class AbstractLogicalInput
    implements LogicalInput, LogicalInputFrameworkInterface {

  private final int numPhysicalInputs;
  private final InputContext inputContext;

  public AbstractLogicalInput(InputContext inputContext, int numPhysicalInputs) {
    this.inputContext = inputContext;
    this.numPhysicalInputs = numPhysicalInputs;
  }
  // ...
  public final int getNumPhysicalInputs() { return numPhysicalInputs; }
  public final InputContext getContext()  { return inputContext; }
  public float getProgress() { return 0.0f; }   // override for real progress
}

Two things worth burning in:

1. The constructor signature is a hard contract. Tez instantiates your input reflectively and requires a public two-arg (InputContext, int) constructor. The class Javadoc says so verbatim. Miss it and you get a NoSuchMethodException at task init, not compile time.
2. numPhysicalInputs is fixed at construction and comes from edge routing. A SCATTER_GATHER input into a task sees numPhysicalInputs == source vertex parallelism; a BROADCAST input sees the same. That number is what your handleEvents will receive one DataMovementEvent per, over the life of the task. See logical-physical.md for who computes it.

AbstractLogicalOutput is the mirror image — constructor (OutputContext, int numPhysicalOutputs), plus getNumPhysicalOutputs() and getContext().

grep -n "public AbstractLogicalOutput\|getNumPhysicalOutputs\|abstract.*initialize" \
  tez-api/src/main/java/org/apache/tez/runtime/api/AbstractLogicalOutput.java

Warning: OutputFrameworkInterface.handleEvents(List<Event>) returns void and does not declare throws Exception, whereas InputFrameworkInterface.handleEvents throws. This asymmetry is real — confirm it with grep -n handleEvents tez-api/src/main/java/org/apache/tez/runtime/api/*FrameworkInterface.java. An output that needs to fail on a bad event must wrap and rethrow as an unchecked exception.

The processor contract

The processor is where you actually consume readers and drive writers.

grep -n "run(" tez-api/src/main/java/org/apache/tez/runtime/api/LogicalIOProcessorFrameworkInterface.java
sed -n '30,80p' tez-api/src/main/java/org/apache/tez/runtime/api/AbstractLogicalIOProcessor.java

// tez-api, org.apache.tez.runtime.api.LogicalIOProcessorFrameworkInterface
public void run(Map<String, LogicalInput> inputs,
                Map<String, LogicalOutput> outputs) throws Exception;

// tez-api, org.apache.tez.runtime.api.AbstractLogicalIOProcessor
public abstract class AbstractLogicalIOProcessor
    implements LogicalIOProcessor, LogicalIOProcessorFrameworkInterface {
  private final ProcessorContext context;
  public AbstractLogicalIOProcessor(ProcessorContext context) { this.context = context; }
  @Override public abstract void initialize() throws Exception;
  public final ProcessorContext getContext() { return context; }
  @Override public void abort() { }   // default no-op
}

The map keys are source/destination vertex names — the same strings you used when you built the DAG edges. run receives Map<String, LogicalInput> (not AbstractLogicalInput), so inside run you can only call start() and getReader() on an input — exactly the user-facing half.

Tip: ProcessorFrameworkInterface declares abort() as @Unstable. It is the framework's way of telling a running processor "wrap up, you are being killed." The default AbstractLogicalIOProcessor.abort() does nothing — override it if your processor holds resources a kill must release promptly. The kill path itself is in tez-runtime.md via TaskRunner2Callable.

For the common case of "read all inputs, produce output, done," extend SimpleProcessor instead of AbstractLogicalIOProcessor — it implements the input/output start dance for you and calls a single run() you override.

Lifecycle: the exact ordering

The framework, not your code, drives the lifecycle. The authoritative sequence lives in LogicalIOProcessorRuntimeTask.initialize() / run() / close() (covered in depth in tez-runtime.md). Read it:

sed -n '234,340p' \
  tez-runtime-internals/src/main/java/org/apache/tez/runtime/LogicalIOProcessorRuntimeTask.java

The ordering it enforces:

1. construct processor, each input, each output (reflection, 2-arg ctors)
2. processor.initialize()           ─┐  submitted to a thread pool;
3. input.initialize()  (each)        ├─ ALL run in parallel,
4. output.initialize() (each)       ─┘  framework blocks until all return
5. memoryDistributor.makeInitialAllocations()   ← callbacks fire here
6. input.start()  (auto-started by framework, in parallel)
7. processor.run(inputs, outputs)   ← YOUR code; calls getReader()/getWriter()
8. input.close() / output.close()   (each)  → return completion events
9. processor.close()

Critical facts that trip people up:

• initialize() for all IOs is parallel. The framework submits one InitializeInputCallable / InitializeOutputCallable per IO to a pool sized numInputs + numOutputs, then blocks on a completion service. There is no ordering guarantee among IOs — a multi-input join's inputs all initialize concurrently. Never assume input A initialized before input B.
• start() is auto-invoked by the framework just after allocations, for every input not already started. This is why Input.start() "must be non-blocking" (the interface Javadoc says so twice) — the framework calls it on a pool thread and must not wedge. A start() that blocks on I/O deadlocks task startup.
• start() must tolerate multiple invocations. The processor may also call start(). Both OrderedGroupedKVInput and OrderedPartitionedKVOutput guard with an AtomicBoolean isStarted and no-op on the second call. Do the same.

// tez-runtime-library, OrderedGroupedKVInput.start()
public synchronized void start() throws IOException {
  if (!isStarted.get()) {
    memoryUpdateCallbackHandler.validateUpdateReceived();
    shuffle = createShuffle();
    shuffle.run();
    // ... drain events that arrived before start
    isStarted.set(true);
  }
}

The memory-request handshake

An IO does not allocate its big buffers directly. It asks a broker for memory and resizes when the broker replies. This is the single most important framework interaction for any buffer-heavy IO, because it is what keeps a task with a 512 MB sort buffer and two 256 MB shuffle inputs from OOMing a 1 GB container.

grep -n "requestInitialMemory\|getTotalMemoryAvailableToTask" \
  tez-api/src/main/java/org/apache/tez/runtime/api/TaskContext.java
grep -n "memoryAssigned" \
  tez-api/src/main/java/org/apache/tez/runtime/api/MemoryUpdateCallback.java

// tez-api, org.apache.tez.runtime.api.TaskContext  (InputContext/OutputContext extend it)
public void requestInitialMemory(long size, MemoryUpdateCallback callbackHandler);
public long getTotalMemoryAvailableToTask();

// tez-api, org.apache.tez.runtime.api.MemoryUpdateCallback
public abstract class MemoryUpdateCallback {
  public abstract void memoryAssigned(long assignedSize);
}

The protocol, per IO:

1. In initialize(), call getContext().requestInitialMemory(desired, callback). OrderedPartitionedKVOutput requests tez.runtime.io.sort.mb worth; OrderedGroupedKVInput requests Shuffle.getInitialMemoryRequirement(...).
2. The framework collects all requests, then in MemoryDistributor.makeInitialAllocations() runs an InitialMemoryAllocator (default WeightedScalingMemoryDistributor) to scale the sum down to fit the container.
3. The distributor calls callback.memoryAssigned(actualBytes) on each IO.
4. Your IO uses actualBytes — not the number you asked for — to size buffers.

// tez-runtime-library, OrderedGroupedKVInput.initialize()  (abbreviated)
long initialMemoryRequest = Shuffle.getInitialMemoryRequirement(conf,
    getContext().getTotalMemoryAvailableToTask());
this.memoryUpdateCallbackHandler = new MemoryUpdateCallbackHandler();
getContext().requestInitialMemory(initialMemoryRequest, memoryUpdateCallbackHandler);

An IO that requests zero (e.g. OrderedGroupedKVInput with zero physical inputs) passes a null callback and immediately marks itself ready via getContext().inputIsReady(). The full distributor mechanics — weights, reserve fraction, the IllegalStateException you get for requesting after allocations ran — are in tez-runtime.md.

Warning: If you skip the handshake and new byte[256<<20] directly in initialize(), nothing stops you at compile time, but you have opted out of scaling. Two such IOs in one container and you OOM. Always route buffers through memoryAssigned.

Events flowing between AM and IO

Tez has no side-channel event bus. Every event to or from an IO rides the umbilical heartbeat (see tez-runtime.md and event-routing.md). Enumerate the shipped event types:

ls tez-api/src/main/java/org/apache/tez/runtime/api/events/

There are eleven event classes here, not the "seven" a stale mental model might suggest — ls the directory yourself, the set grows over releases.

The most consequential IO method for events is close() on an output. Its return value is what the downstream vertex learns from:

// tez-runtime-library, OrderedPartitionedKVOutput.close()
public synchronized List<Event> close() throws IOException {
  List<Event> returnEvents = Lists.newLinkedList();
  if (sorter != null) {
    sorter.flush();
    returnEvents.addAll(sorter.close());
    returnEvents.addAll(generateEvents());   // CompositeDataMovementEvent + VertexManagerEvent
    sorter = null;
  } else {
    returnEvents = generateEmptyEvents();     // still emit, so downstream isn't stuck
  }
  return returnEvents;
}

generateEvents() produces one CompositeDataMovementEvent describing all partitions (the AM later expand()s it into per-destination DataMovementEvents via the edge manager) plus a VertexManagerEvent carrying partition byte-size stats. An output that returns an empty list when it actually wrote data starves the downstream vertex — note how the code above emits empty events rather than nothing even when the sorter never started.

sequenceDiagram
  participant SO as Source LogicalOutput
  participant AM as AM (VertexImpl + EdgeManager)
  participant DI as Dest LogicalInput
  SO->>SO: close() -> sorter.flush()
  SO-->>AM: CompositeDataMovementEvent (count=numPartitions) + VertexManagerEvent
  AM->>AM: EdgeManager.routeCompositeDataMovementEventToDestination(...)
  AM-->>DI: DataMovementEvent (per source attempt, per partition)
  DI->>DI: handleEvents([...]) -> schedule a fetch
  Note over DI: fetch failure?
  DI-->>AM: InputReadErrorEvent -> AM re-runs source attempt

The shipped catalog

You will rarely write an IO from scratch; you will pick from the library. Here is the full shipped set — enumerate it yourself, it is short:

ls tez-runtime-library/src/main/java/org/apache/tez/runtime/library/input/
ls tez-runtime-library/src/main/java/org/apache/tez/runtime/library/output/
ls tez-runtime-library/src/main/java/org/apache/tez/runtime/library/processor/

Root inputs (MRInput, MROutput) live in tez-mapreduce, not tez-runtime-library:

find tez-mapreduce/src/main/java -name "MRInput.java" -o -name "MROutput.java" \
  -o -name "MRInputAMSplitGenerator.java"

MRInput is the canonical root input: its AM-side MRInputAMSplitGenerator calls InputFormat.getSplits(...) and pushes InputDataInformationEvents to tasks. Intermediate inputs like OrderedGroupedKVInput have no initializer — their data descriptors arrive as DataMovementEvents from upstream close().

Auto-wiring: edge configs

You do not hand-build the OutputDescriptor/InputDescriptor pair and the EdgeProperty for a shuffle by hand. The *EdgeConfig builders in tez-runtime-library/.../conf/ do it, guaranteeing the two sides agree.

ls tez-runtime-library/src/main/java/org/apache/tez/runtime/library/conf/
grep -n "newBuilder\|createDefaultEdgeProperty\|createDefaultCustomEdgeProperty" \
  tez-runtime-library/src/main/java/org/apache/tez/runtime/library/conf/OrderedPartitionedKVEdgeConfig.java

// tez-runtime-library, OrderedPartitionedKVEdgeConfig
OrderedPartitionedKVEdgeConfig edgeConf =
    OrderedPartitionedKVEdgeConfig
        .newBuilder("org.apache.hadoop.io.Text",
                    "org.apache.hadoop.io.IntWritable",
                    HashPartitioner.class.getName())
        .setFromConfiguration(tezConf)
        .build();

Edge e = Edge.create(mapVertex, reduceVertex, edgeConf.createDefaultEdgeProperty());

createDefaultEdgeProperty() returns a fully-formed EdgeProperty:

// tez-runtime-library, OrderedPartitionedKVEdgeConfig.createDefaultEdgeProperty()
EdgeProperty edgeProperty = EdgeProperty.create(
    EdgeProperty.DataMovementType.SCATTER_GATHER,
    EdgeProperty.DataSourceType.PERSISTED,
    EdgeProperty.SchedulingType.SEQUENTIAL,
    OutputDescriptor.create(getOutputClassName()).setUserPayload(getOutputPayload()),
    InputDescriptor.create(getInputClassName()).setUserPayload(getInputPayload()));

Note the guarantee: the config knows its output class is OrderedPartitionedKVOutput and its input class is OrderedGroupedKVInput, and it serializes the same key/value/partitioner config into both payloads. That is why you use the builder instead of wiring descriptors by hand — hand-wiring is the classic way to ship a partitioner to the output that the input never hears about. There is also createDefaultCustomEdgeProperty(edgeManagerDescriptor) for CUSTOM movement (see logical-physical.md).

The three edge-config families:

MergedLogicalInput

When a vertex has several physical inputs that should look like one stream to the processor (a vertex-group union, a multi-way concat), the framework wraps them in a MergedLogicalInput.

sed -n '40,90p' tez-api/src/main/java/org/apache/tez/runtime/api/MergedLogicalInput.java
grep -rln "extends MergedLogicalInput" tez-runtime-library/src/main/java

// tez-api, org.apache.tez.runtime.api.MergedLogicalInput
public abstract class MergedLogicalInput implements LogicalInput {
  private List<Input> inputs;
  public MergedLogicalInput(MergedInputContext context, List<Input> inputs) { ... }
  public final List<Input> getInputs() { return inputs; }

  @Override
  public final void start() throws Exception {     // starts all constituents once
    if (!isStarted.getAndSet(true)) {
      for (Input input : inputs) { input.start(); }
    }
  }
}

The processor still calls getReader() exactly once; your MergedLogicalInput subclass combines the constituent readers. Concrete subclasses: ConcatenatedMergedKeyValueInput, ConcatenatedMergedKeyValuesInput, OrderedGroupedMergedKVInput. Note start() is final and fans out to every constituent — you cannot forget to start one.

Reading exercise

# The two halves of each contract
sed -n '30,60p' tez-api/src/main/java/org/apache/tez/runtime/api/Input.java
sed -n '40,75p' tez-api/src/main/java/org/apache/tez/runtime/api/InputFrameworkInterface.java

# Real implementations to model
grep -rln "extends AbstractLogicalInput"  tez-runtime-library/src/main/java
grep -rln "extends AbstractLogicalOutput" tez-runtime-library/src/main/java

# Lifecycle authority
sed -n '234,340p' \
  tez-runtime-internals/src/main/java/org/apache/tez/runtime/LogicalIOProcessorRuntimeTask.java

# Events
ls tez-api/src/main/java/org/apache/tez/runtime/api/events/

Answer:

1. Which methods can a processor call on an input it received in run(...), and which can it not? Justify from the type run's map is parameterized on.
2. Is there any ordering guarantee between initialize() of input A and input B in the same task? Cite the loop in LogicalIOProcessorRuntimeTask that proves your answer.
3. Why must Input.start() be non-blocking? What breaks if it blocks on a socket read? Trace the caller.
4. Find OrderedPartitionedKVOutput.close(). Name the two event types it returns and the downstream effect of each.
5. In the memory handshake, what value should an IO size its buffer to — the one it requested, or the one from memoryAssigned? What happens if it uses the request?
6. Distinguish InputInitializerEvent from InputDataInformationEvent: who emits each, in which direction, and for which class of input?

Common bugs and symptoms

Validation: prove you understand this

1. Write a minimal AbstractLogicalInput whose getReader() yields 100 fixed strings. Give it the correct constructor, a no-op handleEvents, and empty initialize/close. Wire it into a one-vertex DAG on MiniTezCluster (see local-mode.md) and run it.
2. From OrderedGroupedKVInput, quote the exact lines where handleEvents queues events that arrive before start() and where start() drains them.
3. List all eleven classes in org.apache.tez.runtime.api.events and mark the direction of each (AM→task, task→AM, or AM-internal).
4. Diagram the event path from one upstream LogicalOutput.close() to a downstream LogicalInput.handleEvents(), naming the EdgeManagerPlugin method in the middle. Cross-check against logical-physical.md.
5. Explain, in terms of callers, why Tez splits each contract into a user-facing interface and a *FrameworkInterface. What would break if it were one interface?
6. Given a vertex with a MergedLogicalInput over three constituents, how many times is getReader() called by the processor, and how many start() calls reach the constituents? Cite the final methods that enforce your answer.