DAG Model

A Tez DAG is an immutable plan for a distributed computation. Everything else in this book — the AM, the schedulers, the shuffle — exists to execute the object graph you build with the classes in this chapter. The model lives entirely in tez-api, which is deliberately free of YARN server dependencies: a DAG you construct in a unit test is byte-for-byte the same plan the DAGAppMaster will execute on a 2,000-node cluster.

After this chapter you should be able to: write a small DAG by hand without an IDE; recite the three EdgeProperty enums and predict which EdgeManager the AM will instantiate for any edge; enumerate the checks DAG.verify() performs and construct a one-line DAG that fails each of them; and explain how the API object graph is flattened into the DAGPlan protobuf that crosses the wire.

This chapter pairs with tez-client.md (which ships the plan) and dag-app-master.md (which executes it). The hands-on companion is Lab 3.1: trace a DAG submission.

The classes you actually call from a client

ls tez-api/src/main/java/org/apache/tez/dag/api/
grep -n "^public " tez-api/src/main/java/org/apache/tez/dag/api/DAG.java | head -40

The public surface is small and deliberate:

Class	Role
`DAG`	The container: named vertices, edges, vertex groups, DAG-scoped local resources, credentials, ACLs.
`Vertex`	One logical processing stage: a `ProcessorDescriptor`, a parallelism, resources, environment, data sources/sinks.
`Edge`	A directed connection between two vertices, carrying an `EdgeProperty`.
`EdgeProperty`	The three-axis classification of an edge plus the `OutputDescriptor`/`InputDescriptor` pair that implements it.
`InputDescriptor` / `OutputDescriptor` / `ProcessorDescriptor`	Class name + `UserPayload` for user code instantiated in the task JVM. All extend `EntityDescriptor`.
`UserPayload`	An opaque, versioned `ByteBuffer` wrapper — the only way to pass configuration to your plugin classes.
`DataSourceDescriptor` / `DataSinkDescriptor`	Root inputs and leaf outputs: how a first/last vertex reads from or commits to the outside world.
`VertexGroup` / `GroupInputEdge`	Union several vertices' output into one logical input on a consumer vertex.
`VertexLocationHint` / `TaskLocationHint`	Optional placement hints, per vertex and per task.
`PreWarmVertex`	A special vertex used only to pre-allocate session containers (see tez-client.md).

Construction is by static factory, never new:

DAG dag = DAG.create("wordcount");
Vertex tokenizer = Vertex.create("Tokenizer",
    ProcessorDescriptor.create(TokenProcessor.class.getName()), numMappers);
Vertex summation = Vertex.create("Summation",
    ProcessorDescriptor.create(SumProcessor.class.getName()), numReducers);
dag.addVertex(tokenizer).addVertex(summation)
   .addEdge(Edge.create(tokenizer, summation, edgeProperty));

Everything is mutable while you build, frozen at submission. The mutation API is addVertex, addEdge, addTaskLocalFiles, createVertexGroup, addURIsForCredentials — all synchronized methods on DAG (grep "public synchronized DAG" in DAG.java). After TezClient.submitDAG() the only sanctioned way to change the plan is at runtime inside the AM via VertexManagerPlugin callbacks (see vertex-lifecycle.md and logical-physical.md).

Note: Vertex.setParallelism(int) is package-private on master. You fix parallelism at Vertex.create(name, processor, parallelism) time, or you leave it at -1 and let an InputInitializer or VertexManagerPlugin decide it in the AM. The validation rules for -1 are covered below — they are a common interview-grade question.

Descriptors and UserPayload — how user code travels

grep -n "class EntityDescriptor" tez-api/src/main/java/org/apache/tez/dag/api/EntityDescriptor.java
grep -n "public static UserPayload create" tez-api/src/main/java/org/apache/tez/dag/api/UserPayload.java

Every piece of user code in Tez — processor, input, output, committer, initializer, vertex manager, edge manager — is described the same way: a fully qualified class name plus an optional UserPayload. EntityDescriptor is the base class; InputDescriptor.create(className), OutputDescriptor.create(className), and ProcessorDescriptor.create(className) are thin subclasses that exist so the type system stops you from wiring an output class into a processor slot.

UserPayload is a wrapper over ByteBuffer with an int version:

// tez-api: org.apache.tez.dag.api.UserPayload
public static UserPayload create(@Nullable ByteBuffer payload) {
  return new UserPayload(payload, 0);
}

The idiomatic way to fill it is TezUtils.createUserPayloadFromConf(conf) (org.apache.tez.common.TezUtils), which serializes a Hadoop Configuration. The payload is embedded verbatim into the plan protobuf, so a multi-megabyte payload bloats every SubmitDAGRequestProto; keep it to configuration, not data.

Distinguish carefully:

Concept	Class	Lives during
Plan-time root-input definition	`DataSourceDescriptor` (input + optional `InputInitializerDescriptor` + credentials)	Client + AM planning
Plan-time leaf-output definition	`DataSinkDescriptor` (output + optional `OutputCommitterDescriptor`)	Client + AM planning/commit
Runtime input in the task JVM	`LogicalInput` implementation named by the `InputDescriptor`	Task execution

A DataSourceDescriptor may carry an InputInitializer that the AM runs before the vertex starts (e.g. MRInputAMSplitGenerator computing splits). Its output arrives at tasks as InputDataInformationEvents — see event-routing.md and ipo-abstractions.md. The task never sees the DataSourceDescriptor itself.

EdgeProperty — three orthogonal axes

EdgeProperty is the most consequential class in the API. Read the enums in full; their Javadoc is normative:

grep -n "enum " tez-api/src/main/java/org/apache/tez/dag/api/EdgeProperty.java

From tez-api, org.apache.tez.dag.api.EdgeProperty:

public enum DataMovementType {
  /** Output produced by the i-th source task is available to the i-th destination task. */
  ONE_TO_ONE,
  /** Output produced by any source task is available to all destination tasks. */
  BROADCAST,
  /** The i-th output produced by all source tasks is available to the same
   *  destination task. Source tasks scatter their outputs and they are
   *  gathered by designated destination tasks. */
  SCATTER_GATHER,
  /** Custom routing defined by the user. */
  CUSTOM
}

public enum DataSourceType {
  /** Data produced by the source is persisted and available even when the
   *  task is not running. The data may become unavailable and may cause the
   *  source task to be re-executed. */
  PERSISTED,
  /** Source data is stored reliably and will always be available. This is not supported yet. */
  @Unstable
  PERSISTED_RELIABLE,
  /** Data produced by the source task is available only while the source task
   *  is running. This requires the destination task to run concurrently. */
  @Unstable
  EPHEMERAL
}

public enum SchedulingType {
  /** Destination task is eligible to run after one or more of its source tasks
   *  have started or completed. */
  SEQUENTIAL,
  /** Destination task must run concurrently with the source task. Development in progress. */
  @Unstable
  CONCURRENT
}

There is a fourth enum, ConcurrentEdgeTriggerType, used only with CONCURRENT scheduling. Note that on master, PERSISTED_RELIABLE, EPHEMERAL, CONCURRENT, and everything around them is still @Unstable. In practice every production edge you will debug is (SCATTER_GATHER|BROADCAST|ONE_TO_ONE|CUSTOM, PERSISTED, SEQUENTIAL).

The factory ties movement to the I/O pair:

EdgeProperty.create(DataMovementType.SCATTER_GATHER,
    DataSourceType.PERSISTED, SchedulingType.SEQUENTIAL,
    outputDescriptor,   // e.g. OrderedPartitionedKVOutput
    inputDescriptor);   // e.g. OrderedGroupedKVInput

There are two more create overloads taking an EdgeManagerPluginDescriptor for CUSTOM movement — grep "public static EdgeProperty create".

How the AM picks the EdgeManager

The plan-time enum becomes a runtime router inside the AM. This is not in tez-api — it is org.apache.tez.dag.app.dag.impl.Edge in tez-dag:

grep -n "createEdgeManager" -A 30 tez-dag/src/main/java/org/apache/tez/dag/app/dag/impl/Edge.java

// tez-dag: org.apache.tez.dag.app.dag.impl.Edge
private void createEdgeManager() throws TezException {
  switch (edgeProperty.getDataMovementType()) {
    case ONE_TO_ONE:
      if (conf.getBoolean(TezConfiguration.TEZ_AM_ONE_TO_ONE_ROUTING_USE_ON_DEMAND_ROUTING,
          TezConfiguration.TEZ_AM_ONE_TO_ONE_ROUTING_USE_ON_DEMAND_ROUTING_DEFAULT)) {
        edgeManager = new OneToOneEdgeManagerOnDemand(edgeManagerContext);
      } else {
        edgeManager = new OneToOneEdgeManager(edgeManagerContext);
      }
      break;
    case BROADCAST:
      edgeManager = new BroadcastEdgeManager(edgeManagerContext);
      break;
    case SCATTER_GATHER:
      edgeManager = new ScatterGatherEdgeManager(edgeManagerContext);
      break;
    case CUSTOM:
      // reflectively instantiate the user's EdgeManagerPlugin from the
      // EdgeManagerPluginDescriptor, passing its UserPayload
      // ...
  }
}

Movement	Runtime `EdgeManager` (tez-dag `dag/impl`)	Typical use
`SCATTER_GATHER`	`ScatterGatherEdgeManager`	Map → Reduce shuffle
`BROADCAST`	`BroadcastEdgeManager`	Small-side join broadcast
`ONE_TO_ONE`	`OneToOneEdgeManager` (or `OneToOneEdgeManagerOnDemand` when `tez.am.one-to-one.routing.use.on-demand-routing=true`; default `false`)	Pipelined same-parallelism stages
`CUSTOM`	Your `EdgeManagerPlugin` subclass	Hive cartesian product, custom partitioning

Event routing across these managers — including on-demand routing, which computes routes lazily instead of materializing full routing tables — is the subject of event-routing.md.

Vertex parallelism and location hints

grep -n "public static Vertex create\|setLocationHint\|VertexLocationHint\|TaskLocationHint" \
  tez-api/src/main/java/org/apache/tez/dag/api/Vertex.java | head

A vertex's parallelism is either a concrete positive int or -1 ("decide later, in the AM"). Placement is advisory: VertexLocationHint.create(List<TaskLocationHint>), where each TaskLocationHint.createTaskLocationHint(Set<String> hosts, Set<String> racks) names preferred hosts/racks for one task index, and a second form (createTaskLocationHint(String vertexName, int taskIndex)) expresses affinity to another vertex's task for container reuse. Hints feed the scheduler (scheduler.md); they are never guarantees.

The remaining per-vertex knobs mirror per-DAG ones: addTaskLocalFiles, setTaskEnvironment, setTaskLaunchCmdOpts, setConf(property, value), and setExecutionContext (which selects named scheduler/launcher/communicator plugins per vertex — see the plugin section of dag-app-master.md).

VertexGroup and GroupInputEdge

A VertexGroup makes several producer vertices look like a single logical producer to one consumer — the standard encoding for a union (Hive's UNION ALL compiles to exactly this; see hive-integration.md).

grep -n "createVertexGroup" tez-api/src/main/java/org/apache/tez/dag/api/DAG.java
grep -n "public static GroupInputEdge create" -A 6 tez-api/src/main/java/org/apache/tez/dag/api/GroupInputEdge.java

// tez-api: org.apache.tez.dag.api.GroupInputEdge
public static GroupInputEdge create(VertexGroup inputVertexGroup,
                                    Vertex outputVertex,
                                    EdgeProperty edgeProperty,
                                    InputDescriptor mergedInput) {
  return new GroupInputEdge(inputVertexGroup, outputVertex, edgeProperty, mergedInput);
}

Rules worth memorizing:

Groups are created through the DAG (dag.createVertexGroup(name, v1, v2)), not free-standing, so the DAG can police name collisions between groups and vertices.
The mergedInput descriptor names a merged-input class (e.g. ConcatenatedMergedKeyValueInput) that presents N physical inputs as one logical input to the consumer's processor — see ipo-abstractions.md.
VertexGroup.addDataSink lets all members share one output/committer, which is how "N vertices commit to one table" becomes atomic-ish at DAG commit.
In the wire format, groups do not add plan edges; a GroupInputEdge is expanded into ordinary EdgePlans per member plus PlanVertexGroupInfo metadata (grep PlanVertexGroupInfo in DAGApiRecords.proto).

Validation — what `DAG.verify()` actually checks

grep -n "Deque<String> verify(boolean restricted)" tez-api/src/main/java/org/apache/tez/dag/api/DAG.java
grep -n "detectCycles\|strongConnect\|checkAndInferOneToOneParallelism" \
  tez-api/src/main/java/org/apache/tez/dag/api/DAG.java

verify(true) runs as the first step of createDag(...), i.e. at submission time, client-side. The checks, in source order (all throw IllegalStateException unless noted):

Non-empty: "Invalid dag containing 0 vertices".
Unique vertex names: "DAG contains multiple vertices with name: ...".
Input/Output names must not collide with vertex names, nor with the names of adjacent vertices — four distinct loops in the source, e.g. "Vertex: X contains an incoming vertex and Input with the same name: ...". This matters because at runtime a task identifies each of its inputs by name, and an input fed by an edge is named after the source vertex.
Acyclicity via Tarjan's strongly-connected-components algorithm (detectCycles/strongConnect), which doubles as the producer of the topological order the AM consumes later:

// tez-api: org.apache.tez.dag.api.DAG.strongConnect (trimmed)
if (av.lowlink == av.index) {
  AnnotatedVertex pop = stack.pop();
  if (pop != av) {
    // strongly connected component detected -> a cycle
    StringBuilder message = new StringBuilder();
    message.append(av.v.getName()).append(" <- ");
    for (; pop != av; pop = stack.pop()) {
      message.append(pop.v.getName()).append(" <- ");
      pop.onstack = false;
    }
    message.append(av.v.getName());
    throw new IllegalStateException("DAG contains a cycle: " + message);
  } else {
    // detect self-cycle ...
    throw new IllegalStateException("DAG contains a self-cycle on vertex:" + pop.v.getName());
  }
}

One-to-one parallelism inference and consistency (checkAndInferOneToOneParallelism). First it propagates known parallelism across ONE_TO_ONE edges (logging "Inferring parallelism for vertex..."), then it enforces consistency — mismatch throws TezUncheckedException("1-1 Edge. Destination vertex parallelism must match source vertex. ...") — and finally it polices -1:

// tez-api: org.apache.tez.dag.api.DAG.checkAndInferOneToOneParallelism (trimmed)
// vertices with -1 parallelism, currently only 3 cases are allowed:
//   1. has input initializers
//   2. 1-1 uninited sources
//   3. has custom vertex manager
for (Vertex vertex : vertices.values()) {
  if (vertex.getParallelism() == -1) {
    // ... hasInputInitializer? numShards known from client-side splits? continue
    // ... has1to1UninitedSources? continue
    if (vertex.getVertexManagerPlugin() != null) { continue; }
    throw new IllegalStateException(vertex.getName() +
        " has -1 tasks but does not have input initializers, " +
        "1-1 uninited sources or custom vertex manager to set it at runtime");
  }
}

Restricted-mode edge check: with restricted=true, every edge's DataSourceType must be PERSISTED or EPHEMERAL — anything else is "Unsupported source type on edge." (this is where PERSISTED_RELIABLE is rejected today).

Separately, verifyLocalResources(tezConf) rejects a vertex-level local resource that conflicts (same name, different content) with a DAG-level one.

Tip: The block comment above verify() in DAG.java is a small design document: it categorizes illegal, legal, and "not yet categorized" DAG shapes. Read it before proposing any validator change — several seemingly odd omissions (orphan vertices, parallel edges between the same pair) are deliberate.

From object graph to DAGPlan protobuf

The wire format is defined in tez-api/src/main/proto/DAGApiRecords.proto:

grep -n "^message " tez-api/src/main/proto/DAGApiRecords.proto

// tez-api: DAGApiRecords.proto
message DAGPlan {
  required string name = 1;
  repeated VertexPlan vertex = 2;
  repeated EdgePlan edge = 3;
  optional ConfigurationProto dagConf = 4;
  optional bytes credentials_binary = 5;
  repeated PlanVertexGroupInfo vertex_groups = 6;
  repeated PlanLocalResource local_resource = 7;
  optional string dag_info = 8;
  optional VertexExecutionContextProto default_execution_context = 9;
  optional CallerContextProto caller_context = 10;
  optional ACLInfo aclInfo = 11;
}

Supporting messages you will meet while debugging: VertexPlan (processor, parallelism, location hints, root inputs/leaf outputs, in/out edge IDs), EdgePlan (the two vertex names plus the serialized edge property), TezEntityDescriptorProto (class_name + tez_user_payload + optional history_text — the serialized form of every descriptor), RootInputLeafOutputProto, and AMPluginDescriptorProto (the serialized ServicePluginsDescriptor — see dag-app-master.md).

The conversion happens in DAG.createDag(...):

// tez-api: org.apache.tez.dag.api.DAG
public synchronized DAGPlan createDag(Configuration tezConf, Credentials extraCredentials,
    Map<String, LocalResource> tezJarResources, LocalResource binaryConfig,
    boolean tezLrsAsArchive, ServicePluginsDescriptor servicePluginsDescriptor,
    JavaOptsChecker javaOptsChecker) {
  Deque<String> topologicalVertexStack = verify(true);
  verifyLocalResources(tezConf);
  DAGPlan.Builder dagBuilder = DAGPlan.newBuilder();
  dagBuilder.setName(this.name);
  // ... vertices, edges, groups, credentials, ACLs ...
}

Note what that signature reveals: the plan embeds the Tez framework jar resources, the binary configuration local resource, credentials, and the service-plugin descriptors — the plan is self-sufficient for the AM to run a brand-new DAG without consulting the client again.

The leaf-level to/from proto helpers live in DagTypeConverters (tez-api, ~36 KB of static methods — e.g. convertFromLocalResources, convertCallerContextToProto). One kitchen-sink class instead of per-class toProto() methods keeps protobuf types out of the public API signatures and keeps the generated classes an implementation detail:

grep -n "public static" tez-api/src/main/java/org/apache/tez/dag/api/DagTypeConverters.java | head -20

End-to-end

flowchart LR
    A["User code: DAG.create()"] --> B["addVertex / addEdge /\ncreateVertexGroup / addDataSource"]
    B --> C["TezClient.submitDAG(dag)"]
    C --> D["DAG.verify(true)\nname/cycle/parallelism checks"]
    D -->|ok| E["DAG.createDag(...)\n→ DAGPlan protobuf"]
    E --> F["session: submitDAG RPC\nnon-session: tez-dag.pb local resource"]
    F --> G["DAGAppMaster: DAGImpl\nfrom DAGPlan"]
    G --> H["VertexImpl per VertexPlan"]
    H --> I["dag.impl.Edge per EdgePlan\nEdgeManager selected"]

  tez-api (client JVM)                     tez-dag (AM JVM)
 ┌──────────────────────────┐            ┌───────────────────────────┐
 │ DAG ──┬─ Vertex          │  DAGPlan   │ DAGImpl ──┬─ VertexImpl   │
 │       ├─ Edge(EdgeProp)  │  (proto)   │           ├─ Edge         │
 │       ├─ VertexGroup     │ ─────────► │           │  └ EdgeManager│
 │       └─ DataSource/Sink │            │           └─ TaskImpl...  │
 └──────────────────────────┘            └───────────────────────────┘
   names + descriptors only                live state machines

The two Edge classes are a classic trap: org.apache.tez.dag.api.Edge (tez-api) is three fields and no behavior; org.apache.tez.dag.app.dag.impl.Edge (tez-dag) owns routing state and the EdgeManager. When someone says "look at Edge," always ask which one.

Reading exercise

# The API surface, top-down
sed -n '1,140p' tez-api/src/main/java/org/apache/tez/dag/api/DAG.java
sed -n '1,120p' tez-api/src/main/java/org/apache/tez/dag/api/Vertex.java

# Every verify() throw in one screen
grep -n "throw new IllegalStateException\|throw new TezUncheckedException" \
  tez-api/src/main/java/org/apache/tez/dag/api/DAG.java

# Where the plan is built
grep -n "DAGPlan.newBuilder\|dagBuilder\." \
  tez-api/src/main/java/org/apache/tez/dag/api/DAG.java | head

# Which EdgeManager for which movement type
grep -n "case ONE_TO_ONE\|case BROADCAST\|case SCATTER_GATHER\|case CUSTOM" \
  tez-dag/src/main/java/org/apache/tez/dag/app/dag/impl/Edge.java

Then answer, with file + class citations:

What exception does a cycle produce, and how does the message ("A <- B <- C <- A") get assembled? Why does strongConnect delay throwing until the component is popped?
Which three conditions allow a vertex to keep parallelism -1 past verify()? Which fourth condition (data source with known numShards) also passes, and why does it not need an initializer?
dag.addVertex(v) twice with the same name — what happens, and is it caught at add-time or verify-time? Compare with adding the same Edge twice.
Why does GroupInputEdge need its own mergedInput descriptor when each member edge already has an InputDescriptor?
What is in TezEntityDescriptorProto, and what is history_text for? (Cross-reference counters-diagnostics.md.)
DataSourceType.PERSISTED vs PERSISTED_RELIABLE: what does verify() say about the latter, and what would reliable storage change about task re-execution? (Read the enum Javadoc, then grep tez-dag for consumers of getDataSourceType.)

Common bugs and symptoms

Symptom	Root cause	Where to look
`IllegalStateException: DAG contains a cycle: B <- A <- B` at submit	Back-edge added, often by generated planners (Hive/Pig)	`DAG.strongConnect`; print the DAG with `dag.toString()` first
`IllegalStateException: ... has -1 tasks but does not have input initializers...`	Parallelism left at `-1` with no initializer/1-1 source/vertex manager	`DAG.checkAndInferOneToOneParallelism`
`TezUncheckedException: 1-1 Edge. Destination vertex parallelism must match source vertex`	Both ends of a `ONE_TO_ONE` edge set explicitly but unequal	Same method; fix the plan, don't fight the check
`IllegalStateException: Vertex: X contains an Input with the same name as vertex: Y`	Root input named after a vertex (common with copy-pasted `addDataSource` names)	`DAG.verify` name-collision loops
Wrong data at the consumer of a `CUSTOM` edge	`EdgeManagerPlugin` returns routing inconsistent with the output's partitioning	`dag.impl.Edge.createEdgeManager`; unit-test the plugin against the `EdgeManagerPluginOnDemand` contract
`SubmitDAGRequestProto` enormous / submission slow	Fat `UserPayload` (serialized job-sized objects) embedded in every descriptor	`TezEntityDescriptorProto`; move data to a `LocalResource`, keep payload = config
Union output missing rows from one branch	`VertexGroup` member not wired via `GroupInputEdge`, or merged input class wrong	`PlanVertexGroupInfo` in the plan dump; ipo-abstractions.md

Validation: prove you understand this

On a whiteboard, draw a 4-vertex DAG with two SCATTER_GATHER edges and one BROADCAST edge; annotate every edge with its full (DataMovementType, DataSourceType, SchedulingType) triple and the EdgeManager class the AM will instantiate for it.
From memory, list the five arguments of the primary EdgeProperty.create(...) overload and the seven arguments of the full DAG.createDag(...) overload. Check yourself against the source.
Write six one-line DAG mutations, each failing a different verify() check, and predict each exception message before running.
Explain why topological order falls out of Tarjan's algorithm for free, and where the AM consumes topologicalVertexStack.
Dump a real plan: run any tez-examples job with AM log level DEBUG so the client stages the text plan (TezConstants.TEZ_PB_PLAN_TEXT_NAME), and map every section of the dump back to a proto message from this chapter.

Open-Source Engineer & Contributor