Lab 1.4: Project — Number Pipeline DAG

Background

This is the capstone of Level 1. In Lab 1.3 you ran someone else's DAG and watched it live. Now you build one from scratch, in code you own, and assert its result. You will construct a three-vertex Tez pipeline — generator → multiplier → sink — that runs entirely in local mode (no YARN, no HDFS, no Docker) and proves, via a Tez counter, that it computed the right answer.

Generator (2 tasks)          emits integers 0–99
    │  SCATTER_GATHER (OrderedPartitionedKV, HashPartitioner)
    ▼
Multiplier (2 tasks)         value * 2
    │  SCATTER_GATHER (OrderedPartitionedKV, HashPartitioner)
    ▼
Sink (1 task)                sums values → counter NumberPipeline/TotalSum

Numbers 0–99 flow through the pipeline. The final sum is deterministic and hand-checkable: sum(0..99) * 2 = 4950 * 2 = 9900. That is the whole point of using integers instead of a text corpus — you can verify the math without trusting the framework. This pipeline is the same shape as OrderedWordCount from Lab 1.3, stripped to its skeleton so nothing hides the DAG API.

A complete, runnable companion project already exists at book/projects/level-1-number-pipeline/. This lab makes you build it, run it, read every line, break it deliberately, and then extend it. Read the code; do not skim it. Every symbol you touch maps to a real class in the Tez source you built in Lab 1.1.

Why This Lab Matters for Contributors

• Writing a DAG from scratch is the fastest way to internalize the DAG / Vertex / Edge / processor / I/O model that the rest of Tez is built on.
• The "break it and understand it" experiments teach you the failure modes — missing start(), key skew, parallelism assumptions — that show up in real bug reports.
• Counters are how a processor reports results back to the driver; you will read them in nearly every diagnostic task from here on.
• The extension exercise (adding a vertex) mirrors real patch work: inserting a stage into an existing DAG without breaking the edges around it.

Prerequisites

• Lab 1.1 complete: mvn install -DskipTests -Pnoui installed the Tez artifacts into your local ~/.m2.
• Your build's project version string (mvn help:evaluate -Dexpression=project.version -q -DforceStdout inside your tez clone — e.g. 1.0.0-SNAPSHOT on master).
• JDK and Maven on PATH (same versions as Lab 1.1).
• You have skimmed OrderedWordCount (Lab 1.3) so the three-vertex shape is familiar.

Project Layout

book/projects/
├── pom.xml                              ← parent; pins tez.version + hadoop.version
└── level-1-number-pipeline/
    ├── pom.xml                          ← builds a fat "jar-with-dependencies"
    └── src/main/java/org/apache/tez/learning/l1/
        ├── GeneratorProcessor.java      ← no inputs; emits integers
        ├── MultiplierProcessor.java     ← one input, one output; value * 2
        ├── SinkProcessor.java           ← sums values; publishes a counter
        ├── FilterProcessor.java         ← exercise skeleton (incomplete on purpose)
        └── NumberPipelineDAG.java       ← main class: configures, builds, submits the DAG

The parent pom.xml uses maven-assembly-plugin to emit a fat jar with mainClass org.apache.tez.learning.l1.NumberPipelineDAG, so you can run the whole thing with java -jar.

Step 1: Align the Tez and Hadoop Versions

The project compiles against the Tez artifacts you installed in Lab 1.1, so its tez.version must match what is in your local ~/.m2. Read both and reconcile them:

# The version you actually built/installed:
cd /path/to/tez
mvn help:evaluate -Dexpression=project.version -q -DforceStdout ; echo
mvn help:evaluate -Dexpression=hadoop.version   -q -DforceStdout ; echo

Open book/projects/pom.xml and set the two properties to match. On current master that means:

<properties>
  <tez.version>1.0.0-SNAPSHOT</tez.version>   <!-- match your `mvn install` output -->
  <hadoop.version>3.4.2</hadoop.version>       <!-- match Tez's hadoop.version -->
  ...
</properties>

Warning: A tez.version mismatch is the single most common failure here — Maven will report Could not resolve dependency org.apache.tez:tez-api:<x>. The fix is always: make tez.version equal the version present under ~/.m2/repository/org/apache/tez/tez-api/. Keeping the project's hadoop.version equal to Tez's avoids subtle classpath conflicts between two Hadoop lines.

Step 2: Compile

cd /path/to/opensource-engineer-and-contributor/book/projects

# Build only the level-1 module (fast; the parent also aggregates other levels):
mvn -pl level-1-number-pipeline package -q

On success the fat jar is at (note the 1.0-SNAPSHOT in the name is the learning project's version — the parent pom's <version>, not Tez's):

level-1-number-pipeline/target/level-1-number-pipeline-1.0-SNAPSHOT-jar-with-dependencies.jar

Step 3: Run

java -jar level-1-number-pipeline/target/level-1-number-pipeline-1.0-SNAPSHOT-jar-with-dependencies.jar

Expected output (the many framework INFO lines are elided):

TezClient started (local mode).
Submitting DAG...
[SinkProcessor] task=0  partialSum=9900

=== NumberPipeline Result ===
  Expected : 9900
  Actual   : 9900
  Result   : PASS

Note: A large volume of Tez INFO logging in local mode is normal — the important lines are [SinkProcessor] and the === NumberPipeline Result === block.

Local-mode runtime dependency (important): When tez.local.mode=true, TezClient reflectively loads org.apache.tez.client.LocalClient — and that class lives in the tez-dag module, not in tez-api/tez-runtime-library. If java -jar fails at submit time with a ClassNotFoundException/NoClassDefFoundError naming org.apache.tez.client.LocalClient or another org.apache.tez.dag.* class, the fat jar is missing the local-mode runtime. Fix it by adding these to level-1-number-pipeline/pom.xml and rebuilding:

<dependency>
  <groupId>org.apache.tez</groupId>
  <artifactId>tez-dag</artifactId>
  <version>${tez.version}</version>
</dependency>
<dependency>
  <groupId>org.apache.tez</groupId>
  <artifactId>tez-runtime-internals</artifactId>
  <version>${tez.version}</version>
</dependency>

This is a genuine Tez lesson, not a workaround: local mode is tez-dag. The tez-api classes let you describe a DAG; tez-dag is what runs it.

Step 4: Read Every Source File

Before you modify anything, read each file and answer its questions. The answers are in the Tez source you built — use the grep commands to find them; never trust a line number.

NumberPipelineDAG.java — the driver

The main method is the whole lifecycle in one place. Note the exact call sequence:

TezConfiguration tezConf = new TezConfiguration();
tezConf.setBoolean(TezConfiguration.TEZ_LOCAL_MODE, true);
tezConf.set("fs.defaultFS", "file:///");
tezConf.setBoolean("tez.local.mode.without.network", true);

TezClient tezClient = TezClient.create("NumberPipelineDAG", tezConf);
tezClient.start();
DAGClient dagClient = tezClient.submitDAG(buildDAG());
DAGStatus status = dagClient.waitForCompletion();

The DAG itself is built with the public API — Vertex.create, DAG.create, Edge.create, and the OrderedPartitionedKVEdgeConfig builder:

OrderedPartitionedKVEdgeConfig edgeConf =
    OrderedPartitionedKVEdgeConfig
        .newBuilder(IntWritable.class.getName(),   // key class
                    IntWritable.class.getName(),   // value class
                    HashPartitioner.class.getName())
        .build();

Vertex generator  = Vertex.create("generator",
    ProcessorDescriptor.create(GeneratorProcessor.class.getName()), 2);
// ...multiplier (2), sink (1)...

DAG.create("NumberPipeline")
   .addVertex(generator).addVertex(multiplier).addVertex(sink)
   .addEdge(Edge.create(generator,  multiplier, edgeConf.createDefaultEdgeProperty()))
   .addEdge(Edge.create(multiplier, sink,       edgeConf.createDefaultEdgeProperty()));

Questions to answer (with source references):

1. What does tez.local.mode=true actually change about task execution? (Grep TEZ_LOCAL_MODE usage in tez-dag; find where LocalClient is chosen.)
2. OrderedPartitionedKVEdgeConfig.newBuilder(keyClass, valueClass, partitionerClass) — what is HashPartitioner doing, and where does the partition count come from? (Hint: it is the downstream vertex's parallelism, not a builder argument.)
3. dagClient.waitForCompletion() — does it block the calling thread, or return immediately? (Read the method in DAGClient.)
4. EnumSet.of(StatusGetOpts.GET_COUNTERS) — why is this needed to read the counter? Why aren't counters always attached to DAGStatus?

   grep -rn "GET_COUNTERS" /path/to/tez/tez-api/src/main/java/org/apache/tez/dag/api/client/
   

GeneratorProcessor.java — a source vertex (no inputs)

Extends AbstractLogicalIOProcessor. It has no inputs and one output. Each of the 2 tasks generates 50 integers from its task index:

LogicalOutput logicalOutput = outputs.values().iterator().next();
logicalOutput.start();                                   // REQUIRED before getWriter()
KeyValueWriter writer = ((OrderedPartitionedKVOutput) logicalOutput).getWriter();

int rangeStart = getContext().getTaskIndex() * 50;       // task 0 → 0–49, task 1 → 50–99
for (int n = rangeStart; n < rangeStart + 50; n++) {
    writer.write(new IntWritable(n), new IntWritable(n));
}

Questions:

1. Which Tez interface does it implement, and in which module does that class live?

   find /path/to/tez -name AbstractLogicalIOProcessor.java -not -path '*/target/*'
   

2. Why is output.start() called before getWriter()? What happens if you remove it? (Break 1 below makes you find out.)
3. How does the processor know which range to generate? (getContext().getTaskIndex().)
4. Key and value are the same integer n. Why? When would you want them to differ? (The key feeds HashPartitioner; the value is the payload.)

MultiplierProcessor.java — a transform vertex (input + output)

One input (from generator), one output (to sink). It reads the grouped, sorted input and writes value * FACTOR:

logicalInput.start();    // triggers shuffle fetch + merge of upstream data
logicalOutput.start();
KeyValueReader reader = ((OrderedGroupedKVInput)   logicalInput ).getReader();
KeyValueWriter writer = ((OrderedPartitionedKVOutput) logicalOutput).getWriter();
while (reader.next()) {
    IntWritable value = (IntWritable) reader.getCurrentValue();
    writer.write((IntWritable) reader.getCurrentKey(),
                 new IntWritable(value.get() * FACTOR));   // FACTOR = 2
}

Questions:

1. OrderedGroupedKVInput is the input side; OrderedPartitionedKVOutput is the output side. Why the different names? (One groups fetched data by key; one partitions outgoing data.)
2. What does input.start() actually trigger? Read OrderedGroupedKVInput.start():

   find /path/to/tez -path '*runtime/library/input/OrderedGroupedKVInput.java' -not -path '*/target/*'
   

3. FACTOR = 2 is hardcoded. The Javadoc explains passing it via UserPayload. How many bytes is an int in a ByteBuffer? (You will need this in the FilterProcessor exercise.)

SinkProcessor.java — a terminal vertex (input, no output)

parallelism = 1. It sums all values and publishes the total as a counter:

getContext().getCounters()
    .findCounter(COUNTER_GROUP, COUNTER_NAME)   // "NumberPipeline" / "TotalSum"
    .increment(partialSum);

Questions:

1. What is the type of getContext().getCounters()? (A TezCounters.)
2. findCounter(group, name) — what happens the first time the counter doesn't exist? (It is created lazily and returned.)
3. There is one sink task. If you set parallelism to 2, is the counter still correct? Why? (Yes — the AM aggregates each task's counter; Break 2 proves it.)

Step 5: Break It and Understand It

Make each change, rebuild (mvn -pl level-1-number-pipeline package -q), run, observe, then revert.

Break 1: Remove output.start()

In GeneratorProcessor.run(), comment out logicalOutput.start();.

Expected: a NullPointerException/IllegalStateException from the Tez runtime when getWriter() is called on an uninitialized output. Tez I/O objects are lazily initialized — start() allocates buffers and (for inputs) kicks off the shuffle fetch. Forgetting start() is a classic first-patch mistake.

Break 2: Change the sink parallelism

Change the sink vertex's parallelism from 1 to 3 in buildDAG(), rebuild, run.

Expected: the total counter is still 9900, but now you see three [SinkProcessor] lines with different partialSum values that add up to 9900. The AM sums the per-task counters automatically. This is the counter-aggregation model you will rely on for diagnostics.

Break 3: Force key skew

In GeneratorProcessor, change writer.write(new IntWritable(n), new IntWritable(n)) to writer.write(new IntWritable(0), new IntWritable(n)) — every record now has key 0.

Expected: HashPartitioner routes all records to a single multiplier task (the owner of partition 0); the other multiplier task gets nothing. The result is still 9900 (correct), but the work is completely skewed. Add a counter in MultiplierProcessor that counts records per task to see the skew. Key skew is one of the most common real-world Tez/MapReduce performance problems — this makes it visible in 20 lines.

Step 6: The FilterProcessor Exercise

FilterProcessor.java is a deliberately-incomplete skeleton. Your task: insert a filter vertex between multiplier and sink that keeps only values divisible by a threshold you pass via UserPayload, then verify the new expected sum.

6a — Implement the processor

1. Add a private int threshold; field.
2. In initialize(), read it from the payload:

   byte[] bytes = getContext().getUserPayload().deepCopyAsArray();
   this.threshold = java.nio.ByteBuffer.wrap(bytes).getInt();
   

3. In run(), replace if (true) with if (value.get() % threshold == 0); keep the FilteredOut counter increment on the drop path.

6b — Wire it into the DAG

In NumberPipelineDAG.buildDAG(), create the vertex with a 4-byte UserPayload and re-chain the edges so data flows generator → multiplier → filter → sink:

Vertex filter = Vertex.create("filter",
    ProcessorDescriptor.create(FilterProcessor.class.getName())
        .setUserPayload(UserPayload.create(
            (java.nio.ByteBuffer) java.nio.ByteBuffer.allocate(4).putInt(4).flip())),  // threshold = 4
    2);   // same parallelism as multiplier

DAG.create("NumberPipeline")
   .addVertex(generator).addVertex(multiplier).addVertex(filter).addVertex(sink)
   .addEdge(Edge.create(generator,  multiplier, edgeConf.createDefaultEdgeProperty()))
   .addEdge(Edge.create(multiplier, filter,     edgeConf.createDefaultEdgeProperty()))
   .addEdge(Edge.create(filter,     sink,       edgeConf.createDefaultEdgeProperty()));

6c — Compute the new expected sum and assert it

After * 2, values are 0, 2, 4, …, 198. Keep only those divisible by 4: 0, 4, 8, …, 196. Their sum is 4 * (0 + 1 + … + 49) = 4 * 1225 = 4900. Update expectedSum() to return 4900, rebuild, and confirm the run prints PASS. If it prints FAIL, your filter predicate or the edge chain is wrong — read the FilteredOut counter to see how many records were actually dropped.

Step 7: Connect Every Class to the Tez Source

Each class you used maps to a real module. The table below is verified against the current checkout — note that the runtime API classes live in tez-api (package org.apache.tez.runtime.api), not in a separate tez-runtime-api module, and TezConfiguration lives in tez-api, not tez-common:

For each row: find the method you called, read what it actually does, then find its unit test (usually under the same package in src/test/java/, e.g. TestOrderedPartitionedKVEdgeConfig).

Step 8: Find Related JIRA Issues

Your pipeline exercises OrderedPartitionedKVOutput and the ordered shuffle. Search the Tez JIRA for real work in that area (use the tracker UI at issues.apache.org/jira; do not trust any specific issue number cited from memory):

project = TEZ AND component = "runtime-library" AND status in (Open, "Patch Available")
ORDER BY priority DESC

For each open issue: can you understand the description, locate the relevant code in the source you just mapped, and tell whether a failing test already exists? That triage skill is exactly what Level 2 formalizes.

Deliverables

• tez.version and hadoop.version in book/projects/pom.xml match your Lab 1.1 build.
• The module compiles: mvn -pl level-1-number-pipeline package -q (no errors).
• Running the fat jar prints PASS with result 9900.
• You can answer every question in Step 4 with a source reference (found via grep/find).
• You ran all three "Break It" experiments and can explain each observed behavior.
• FilterProcessor is implemented and the pipeline prints PASS with result 4900.
• You opened each source file in the Step 7 table and located its unit test.
• You found at least two open runtime-library JIRA issues you could plausibly work on.

Troubleshooting

Stretch Goals

1. Payload-drive the multiplier. Do for MultiplierProcessor.FACTOR what the filter exercise did for the threshold: pass it via UserPayload and read it in initialize(). Update expectedSum() accordingly.

2. Add a real output. Give sink an MROutput (TextOutputFormat) writing to a local directory (file:///tmp/...) in addition to the counter, then diff the file against the counter total. This is how OrderedWordCount writes its result.

3. Unordered vs. ordered edges. Swap one edge from OrderedPartitionedKVEdgeConfig to UnorderedPartitionedKVEdgeConfig (find it under .../runtime/library/conf/). What breaks, and what does the sink read differently? Explain in terms of the shuffle-sort deep-dive.

4. Watch the transitions. Re-run with the impl package at INFO and grep transitioned from (Lab 1.3, Step 5). Reconcile the vertex lifecycle you see against vertex-lifecycle.

5. Write a JUnit test. Add a test under src/test/java/ that runs the pipeline in local mode and asserts the counter equals 9900, so a regression is caught automatically. This is the habit every Tez patch must demonstrate.

Validation / Self-check

You are done when you can answer these without notes:

1. What are the three things NumberPipelineDAG.main does with tezConf to enable local mode, and which module provides the runtime that actually executes the DAG?
2. Where does the partition count for HashPartitioner come from — a builder argument, or something else?
3. Why must you call start() on an input/output before getting its reader/writer, and what does input.start() trigger that output.start() does not?
4. How does a single counter end up correct when the sink runs with parallelism > 1?
5. Why does Break 3 (fixed key = 0) still produce the right sum but terrible work distribution, and what real-world problem is that a model of?
6. Which module do the runtime API classes (AbstractLogicalIOProcessor, LogicalOutput) live in, and which module has TezConfiguration — and why is it not tez-common?
7. What is the new expected sum after inserting the divisible-by-4 filter, and how did you derive it by hand?

Where to go next: you have now built, run, broken, and extended a real Tez DAG. That is the whole of Level 1 — see the Level 1 index for how the four labs fit together. The concepts you exercised are dissected in the dag-model, local-mode, and ipo-abstractions deep-dives. When you can do everything in this lab from memory, you are ready to make your first real contribution in Level 2.