Lab 7.2 — Modify a Processor: Add Deduplication

Lab type: Build-It / Extend Estimated time: 150 min Tez module: tez-runtime-library (base class); local-mode DAG you write

Background

A Processor is the user code at the heart of a Tez vertex: the framework fetches, sorts, and merges data, then hands your processor a reader; your processor consumes it and writes output. In this lab you write a real one — a DedupProcessor that emits each key at most once — and you write it twice, because the correct implementation depends entirely on what kind of input the edge delivers:

• Ordered-grouped input (OrderedGroupedKVInput): the framework has already sorted and grouped by key, so its KeyValuesReader hands you each distinct key exactly once. Deduplication falls out of the grouping — you write almost nothing.
• Unordered input (UnorderedKVInput): records arrive raw, unsorted, and a key may appear many times scattered through the stream. To dedup you must hold a Set of seen keys — O(distinct-keys) memory.

Teaching both is the whole point: it forces you to internalize what the shuffle layer already did for you, and to see that "the same feature" is either free or expensive depending on the edge. That distinction is exactly what a reviewer checks when someone proposes a new processor.

You will extend SimpleProcessor from tez-runtime-library, wire the processor into a two-vertex DAG, run it in local mode, and verify the result with a counter — then look at how shipped processors are tested and what a runtime-library PR must include.

Why This Lab Matters for Contributors

Every runtime-library feature is, at bottom, a processor/input/output plus a test. If you can write a correct Processor, wire it into a DAG, and prove it with a counter in local mode, you can write the code half of nearly any runtime-library PR. And knowing that ordered-grouped input makes grouping-based logic free — while unordered input does not — is the kind of judgment that separates a mergeable design from one a reviewer sends back. This lab is the smallest complete instance of that whole workflow.

Prerequisites

• Level 7 index read; you can name the reader types.
• Lab 7.1 complete; you have run a DAG in local mode.
• Tez checkout built and installed to your local Maven repo.
• Familiarity with book/projects/level-3-multi-input (the local-mode DAG pattern this lab reuses).
• Deep dive: IPO abstractions skimmed.

Step-by-Step Tasks

Step 1 — Read the base class contract

Everything starts from SimpleProcessor. Read it and note the contract you must satisfy:

grep -n "abstract class SimpleProcessor\|public void run\|abstract void run\|preOp\|postOp" \
  tez-runtime-library/src/main/java/org/apache/tez/runtime/library/processor/SimpleProcessor.java

SimpleProcessor is an abstract, @Public @Evolving subclass of AbstractLogicalIOProcessor. It implements the framework's run(Map<String,LogicalInput>, Map<String,LogicalOutput>), which:

1. stores your inputs/outputs,
2. calls preOp() (which starts every input and output for you),
3. calls your run() — the single abstract method you must implement,
4. calls postOp().

So you override exactly one method, run(), and inside it you fetch your reader and writer from getInputs()/getOutputs(). You do not start the inputs yourself — preOp() already did. That is the contract.

Step 2 — Read the two reader contracts

The two dedup implementations differ only because the two readers differ. Confirm each:

# Ordered-grouped input yields a KeyValuesReader (key + Iterable<values>):
grep -n "KeyValuesReader getReader" \
  tez-runtime-library/src/main/java/org/apache/tez/runtime/library/input/OrderedGroupedKVInput.java

# The KeyValuesReader contract: next(), getCurrentKey(), getCurrentValues():
grep -n "abstract .*next\|getCurrentKey\|getCurrentValues" \
  tez-runtime-library/src/main/java/org/apache/tez/runtime/library/api/KeyValuesReader.java

# Unordered input yields a KeyValueReader (one value per next()):
grep -n "KeyValueReader getReader" \
  tez-runtime-library/src/main/java/org/apache/tez/runtime/library/input/UnorderedKVInput.java
grep -n "abstract .*next\|getCurrentKey\|getCurrentValue\b" \
  tez-runtime-library/src/main/java/org/apache/tez/runtime/library/api/KeyValueReader.java

The crucial difference:

Read that table until it is obvious why dedup is free on the first and costs a Set on the second.

Step 3 — Write the ordered-grouped DedupProcessor (dedup is free)

Because OrderedGroupedKVInput's KeyValuesReader yields each key once, "deduplicate by key" is just "count how many times next() returns true." Full working code:

package org.apache.tez.learning.l7;

import org.apache.tez.common.counters.TezCounter;
import org.apache.tez.runtime.api.ProcessorContext;
import org.apache.tez.runtime.library.api.KeyValueWriter;
import org.apache.tez.runtime.library.api.KeyValuesReader;
import org.apache.tez.runtime.library.processor.SimpleProcessor;

/**
 * Deduplicates by key on an ORDERED-GROUPED input.
 *
 * The framework (OrderedGroupedKVInput) has already sorted+grouped by key, so
 * the KeyValuesReader hands us each distinct key exactly once. We emit the key
 * with a single representative value (the first of its grouped values) and count
 * distinct keys. No Set required — O(1) extra memory.
 */
public class OrderedDedupProcessor extends SimpleProcessor {

  public static final String GROUP = "Dedup";
  public static final String DISTINCT_KEYS = "DistinctKeys";
  public static final String INPUT_EDGE = "in";
  public static final String OUTPUT_EDGE = "out";

  public OrderedDedupProcessor(ProcessorContext context) {
    super(context);
  }

  @Override
  public void run() throws Exception {
    // preOp() already started inputs/outputs for us.
    KeyValuesReader reader =
        (KeyValuesReader) getInputs().get(INPUT_EDGE).getReader();
    KeyValueWriter writer =
        (KeyValueWriter) getOutputs().get(OUTPUT_EDGE).getWriter();

    TezCounter distinct =
        getContext().getCounters().findCounter(GROUP, DISTINCT_KEYS);

    while (reader.next()) {
      Object key = reader.getCurrentKey();
      // Grouping guarantees this key is seen exactly once. Emit one
      // representative value (the first in the group) and move on.
      Object firstValue = reader.getCurrentValues().iterator().next();
      writer.write(key, firstValue);
      distinct.increment(1);
    }
  }
}

The dedup logic is the absence of logic: there is no Set, no contains check. The shuffle layer did the grouping; your processor rides on top of it. Note in your writeup: memory is O(1) regardless of how many duplicate keys existed upstream.

Step 4 — Write the unordered DedupProcessor (dedup needs a Set)

Now the contrast. On an UnorderedKVInput, records arrive raw and a key may recur anywhere in the stream, so you must remember which keys you've emitted:

package org.apache.tez.learning.l7;

import org.apache.hadoop.io.IntWritable;
import org.apache.tez.common.counters.TezCounter;
import org.apache.tez.runtime.api.ProcessorContext;
import org.apache.tez.runtime.library.api.KeyValueReader;
import org.apache.tez.runtime.library.api.KeyValueWriter;
import org.apache.tez.runtime.library.processor.SimpleProcessor;

import java.util.HashSet;
import java.util.Set;

/**
 * Deduplicates by key on an UNORDERED input.
 *
 * UnorderedKVInput's KeyValueReader yields raw (key,value) pairs in no order, so
 * a key can appear many times scattered through the stream. To emit each key
 * once we must hold a Set of seen keys — O(distinct-keys) memory. This is the
 * cost the ordered-grouped variant avoids entirely.
 */
public class UnorderedDedupProcessor extends SimpleProcessor {

  public static final String GROUP = "Dedup";
  public static final String DISTINCT_KEYS = "DistinctKeys";
  public static final String DUP_SKIPPED = "DuplicatesSkipped";
  public static final String INPUT_EDGE = "in";
  public static final String OUTPUT_EDGE = "out";

  public UnorderedDedupProcessor(ProcessorContext context) {
    super(context);
  }

  @Override
  public void run() throws Exception {
    KeyValueReader reader =
        (KeyValueReader) getInputs().get(INPUT_EDGE).getReader();
    KeyValueWriter writer =
        (KeyValueWriter) getOutputs().get(OUTPUT_EDGE).getWriter();

    TezCounter distinct =
        getContext().getCounters().findCounter(GROUP, DISTINCT_KEYS);
    TezCounter skipped =
        getContext().getCounters().findCounter(GROUP, DUP_SKIPPED);

    // Copy keys before storing: readers may reuse the Writable instance
    // across next() calls, so keeping the live object would alias.
    Set<IntWritable> seen = new HashSet<>();

    while (reader.next()) {
      IntWritable key = (IntWritable) reader.getCurrentKey();
      Object value = reader.getCurrentValue();
      if (seen.add(new IntWritable(key.get()))) {  // add() is false if already present
        writer.write(key, value);
        distinct.increment(1);
      } else {
        skipped.increment(1);
      }
    }
  }
}

Two subtleties worth a note in your writeup, because they are the kind of thing reviewers flag:

• Writable reuse. Readers frequently return the same Writable object each next() and mutate it in place. Storing the live key in the Set would alias every entry to one object. Hence new IntWritable(key.get()).
• Unbounded memory. The Set grows with the number of distinct keys. On a billion-key stream this OOMs the task — which is precisely why, when you control the DAG, you route dedup through an ordered-grouped edge instead.

Step 5 — Wire it into a two-vertex DAG (local mode)

Follow the local-mode pattern from book/projects/level-3-multi-input: a source vertex that emits (key, value) pairs with duplicate keys, connected by an OrderedPartitionedKVEdgeConfig (SCATTER_GATHER, sorted + grouped) to the OrderedDedupProcessor. Full DAG driver:

package org.apache.tez.learning.l7;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.io.IntWritable;
import org.apache.tez.client.TezClient;
import org.apache.tez.dag.api.*;
import org.apache.tez.dag.api.client.DAGClient;
import org.apache.tez.dag.api.client.DAGStatus;
import org.apache.tez.dag.api.client.StatusGetOpts;
import org.apache.tez.runtime.library.conf.OrderedPartitionedKVEdgeConfig;
import org.apache.tez.runtime.library.partitioner.HashPartitioner;

import java.util.EnumSet;

public class DedupDAG {

  public static void main(String[] args) throws Exception {
    TezConfiguration tezConf = new TezConfiguration(new Configuration());
    // Local mode: no YARN cluster required (same pattern as level-3-multi-input).
    tezConf.setBoolean(TezConfiguration.TEZ_LOCAL_MODE, true);
    tezConf.set("fs.defaultFS", "file:///");
    tezConf.setBoolean(TezConfiguration.TEZ_LOCAL_MODE_WITHOUT_NETWORK, true);

    TezClient client = TezClient.create("dedup-dag", tezConf);
    client.start();
    try {
      DAG dag = buildDAG();
      DAGClient dagClient = client.submitDAG(dag);
      DAGStatus status =
          dagClient.waitForCompletion(EnumSet.of(StatusGetOpts.GET_COUNTERS));

      if (status.getState() != DAGStatus.State.SUCCEEDED) {
        System.err.println("DAG FAILED: " + status);
        System.exit(1);
      }
      long distinct = status.getDAGCounters()
          .findCounter(OrderedDedupProcessor.GROUP,
                       OrderedDedupProcessor.DISTINCT_KEYS)
          .getValue();
      System.out.println("DistinctKeys = " + distinct);   // expect 3 (see DupKeySource)
    } finally {
      client.stop();
    }
  }

  static DAG buildDAG() {
    OrderedPartitionedKVEdgeConfig edgeCfg =
        OrderedPartitionedKVEdgeConfig
            .newBuilder(IntWritable.class.getName(),
                        IntWritable.class.getName(),
                        HashPartitioner.class.getName())
            .build();

    Vertex source = Vertex.create(
        "DupKeySource",
        ProcessorDescriptor.create(DupKeySource.class.getName()), 1);

    Vertex dedup = Vertex.create(
        "Dedup",
        ProcessorDescriptor.create(OrderedDedupProcessor.class.getName()), 1);

    // The edge name must match the processor's INPUT_EDGE constant.
    Edge e = Edge.create(source, dedup, edgeCfg.createDefaultEdgeProperty())
                 .setDestinationEdgeName(OrderedDedupProcessor.INPUT_EDGE);

    return DAG.create("DedupDAG")
        .addVertex(source)
        .addVertex(dedup)
        .addEdge(e);
  }
}

The source emits keys with deliberate duplicates so the dedup has something to do:

package org.apache.tez.learning.l7;

import org.apache.hadoop.io.IntWritable;
import org.apache.tez.runtime.api.ProcessorContext;
import org.apache.tez.runtime.library.api.KeyValueWriter;
import org.apache.tez.runtime.library.processor.SimpleProcessor;

/** Emits (key,value) with duplicate keys: keys {1,1,2,3,3,3} → 3 distinct. */
public class DupKeySource extends SimpleProcessor {
  public DupKeySource(ProcessorContext context) { super(context); }

  @Override
  public void run() throws Exception {
    KeyValueWriter w =
        (KeyValueWriter) getOutputs().values().iterator().next().getWriter();
    int[] keys = {1, 1, 2, 3, 3, 3};
    for (int k : keys) {
      w.write(new IntWritable(k), new IntWritable(k * 10));
    }
  }
}

Note: The Dedup vertex here has no downstream output edge, so its getOutputs() is empty — drop the writer lines from OrderedDedupProcessor for the counter-only run, or give Dedup a DataSinkDescriptor (MROutput) the way OrderedWordCount's sorter vertex does if you want the deduped stream on disk. For verification, the counter alone is enough.

Step 6 — Run and verify with a counter

Build and run exactly like the book's other local-mode projects:

cd book/projects
# (add a level-7-dedup module mirroring level-3-multi-input's pom, then:)
mvn -pl level-7-dedup package -q
java -jar level-7-dedup/target/level-7-dedup-1.0-SNAPSHOT-jar-with-dependencies.jar

Expected:

DistinctKeys = 3

The source emitted six records over keys {1,1,2,3,3,3}; after grouping, the ordered dedup sees keys {1,2,3} — three distinct. If you swap in the unordered variant on an UnorderedKVInput edge, DistinctKeys is still 3 and DuplicatesSkipped is 3 (the three extra records). Same result, different memory profile — that is the lab's thesis, now proven by counters.

Step 7 — Write the unit test a PR would ship

Shipped Tez processors are tested with mock inputs/outputs, not by standing up a cluster. Look at how the runtime-library does it, then mirror the pattern:

# Real processor/output tests to model your test on:
ls tez-runtime-library/src/test/java/org/apache/tez/runtime/library/output/
ls tez-runtime-library/src/test/java/org/apache/tez/runtime/library/input/

For your processor, a focused unit test mocks the reader and asserts on the counters — the same Mockito style as the book's TestMultiInputProcessors:

@Test
public void unorderedDedup_skipsRepeatedKeysOnce() throws Exception {
  // Mock a KeyValueReader returning (1,10),(1,10),(2,20)
  // Drive UnorderedDedupProcessor.run()
  // Assert: DistinctKeys == 2, DuplicatesSkipped == 1
}

@Test
public void orderedDedup_countsEachGroupedKeyOnce() throws Exception {
  // Mock a KeyValuesReader whose next() returns keys 1,2,3 once each
  // Assert: DistinctKeys == 3
}

Then note what a real runtime-library PR must include, because this is the checklist a reviewer applies:

• The feature (the processor/input/output) with the right audience/stability annotations (@Public/@Private, @Evolving/@Stable), like SimpleProcessor.
• A unit test that exercises the logic with mocked IPO — like the existing output//input/ tests — not just a happy-path cluster run.
• A counter or observable effect so the behavior is verifiable (yours has DistinctKeys/DuplicatesSkipped).
• Config keys, if any, added to TezRuntimeConfiguration with defaults and a @ConfigurationProperty annotation.
• No line-number-fragile assumptions; tests assert on behavior/counters.

That is the difference between "I wrote a processor" and "I wrote a runtime-library PR."

Deliverables

• OrderedDedupProcessor.java — dedup via grouping, O(1) memory, compiling.
• UnorderedDedupProcessor.java — dedup via Set, with the Writable-copy and DuplicatesSkipped counter.
• DedupDAG.java + DupKeySource.java wired and running in local mode, printing DistinctKeys = 3.
• The counter output proving both variants produce the same distinct count.
• The two unit tests (ordered + unordered) mocking the readers.
• A short writeup contrasting the memory profiles and listing the runtime-library PR checklist.

Troubleshooting

Stretch Goals

1. Prove the memory claim. Feed the unordered variant a large synthetic stream of many distinct keys and watch task heap climb; feed the ordered variant the same data through a sorted edge and watch it stay flat. Capture both.
2. Emit all grouped values, not just the first. Change OrderedDedupProcessor to write every value in the group under its (single) key — the difference between "distinct keys" and "grouped by key" semantics. Explain how this relates to what a reducer normally does.
3. Add a config key. Introduce dedup.emit.first.only in a config, default true; when false, the unordered variant emits the last value seen per key instead of the first. Add it the way TezRuntimeConfiguration adds keys, and test both branches.

Validation

1. Which single method must a SimpleProcessor subclass implement, and what has preOp() already done to your inputs before it runs?
2. Why does the ordered-grouped dedup need no Set? Name the framework step that makes it unnecessary.
3. On the unordered path, why must you copy the key before putting it in the Set? What bug appears if you don't?
4. What is the memory complexity of each variant in terms of distinct keys, and which one can OOM a task?
5. Given only a DistinctKeys counter, how do you confirm both variants agree on the same input?
6. How are shipped Tez processors tested — cluster or mocks — and which existing test packages did you model yours on?
7. List three things a runtime-library PR must include beyond the code itself.

When both processors compile, the DAG prints DistinctKeys = 3 in local mode, and your tests pass, you have written and validated a runtime-library feature the way a contributor does. That closes Level 7 — proceed to Level 8: Reproduce and Fix a Real Issue.