YARN Integration

The Tez AM is, from YARN's perspective, an ordinary YARN application: an ApplicationMaster running in a container, talking to the ResourceManager to request more containers, talking to NodeManagers to launch them, and (when configured) publishing events to a Timeline Server. Nothing about Tez is special to YARN — which is exactly why so many "Tez won't start" and "tasks can't fetch" problems are really YARN-deployment problems wearing a Tez stack trace.

This chapter walks every YARN-facing interface Tez touches, quoting the real classes from your Tez checkout at ~/tez-src. All commands run from the Tez repo root and are verified against current master. Where a class name matters for grep, it is given by module path plus class, never by line number.

After this chapter you can:

• Trace the AM boot path from a YARN-launched container to AMRMClientAsync.registerApplicationMaster.
• Name the client (TezYarnClient) that submits the app and the classes that build the AM's ContainerLaunchContext.
• Read the YarnTaskSchedulerService allocate/heartbeat/callback loop.
• Explain how Tez reacts to preemption, disk failure, node loss, and RM errors — by pointing at the real handler in each case.
• Deploy the tez_shuffle aux-service and reason about token flow.

This chapter pairs with three siblings: Scheduler (how the AM decides what to ask YARN for), Container Reuse (why the AM holds containers instead of releasing them), and Failure Handling (the state-machine reactions to the YARN events described here).

Two YARN roles: the client and the AM

Tez talks to YARN from two processes, and they use different clients:

Confirm the client wrapper — TezYarnClient extends FrameworkClient and holds a YarnClient:

grep -n "class TezYarnClient\|private final YarnClient\|submitApplication\|createApplication\|killApplication" \
  tez-api/src/main/java/org/apache/tez/client/TezYarnClient.java

FrameworkClient is the abstraction that lets Tez also run against a local or alternate runtime; TezYarnClient is the YARN implementation and simply delegates to YarnClient.submitApplication, getApplicationReport, killApplication.

Building the AM: TezClientUtils

Before the app is submitted, the client assembles everything YARN needs to launch the AM container. This is TezClientUtils in tez-api (org.apache.tez.client.TezClientUtils).

grep -n "createApplicationSubmissionContext\|ensureStagingDirExists\|setupTezJarsLocalResources\|setupDAGCredentials\|createSessionToken" \
  tez-api/src/main/java/org/apache/tez/client/TezClientUtils.java

The staging directory comes first (ensureStagingDirExists), keyed on tez.staging-dir (TEZ_AM_STAGING_DIR). Tez localizes the framework tarball from tez.lib.uris (TEZ_LIB_URIS) into per-app local resources via setupTezJarsLocalResources, and stages the serialized DAG plan (TEZ_PB_PLAN_BINARY_NAME) as an application-visibility LocalResource.

The environment (CLASSPATH, etc.) is assembled by TezYARNUtils.setupDefaultEnv (tez-api, org.apache.tez.common.TezYARNUtils), whose classpath comes from getFrameworkClasspath:

grep -n "setupDefaultEnv\|getFrameworkClasspath" \
  tez-api/src/main/java/org/apache/tez/common/TezYARNUtils.java

The AM command line is built in TezClientUtils and ends in the AM main class constant TezConstants.TEZ_APPLICATION_MASTER_CLASS (org.apache.tez.dag.app.DAGAppMaster); for session mode it adds the --session CLI option, and it appends the standard YARN stdout/stderr log redirections:

grep -n "TEZ_APPLICATION_MASTER_CLASS\|TEZ_SESSION_MODE_CLI_OPTION\|LOG_DIR_EXPANSION_VAR" \
  tez-api/src/main/java/org/apache/tez/client/TezClientUtils.java \
  tez-api/src/main/java/org/apache/tez/dag/api/TezConstants.java

The AM ContainerLaunchContext

The CLC is what the NM uses to fork the AM JVM. It is built at the end of createApplicationSubmissionContext with ContainerLaunchContext.newInstance(amLocalResources, environment, vargsFinal, serviceData, securityTokens, acls). Its fields:

The serviceData entry is the subtle one. From TezClientUtils (org.apache.tez.client.TezClientUtils, method createApplicationSubmissionContext):

Map<String, ByteBuffer> serviceData = new HashMap<String, ByteBuffer>();
String auxiliaryService = conf.get(TezConfiguration.TEZ_AM_SHUFFLE_AUXILIARY_SERVICE_ID,
    TezConfiguration.TEZ_AM_SHUFFLE_AUXILIARY_SERVICE_ID_DEFAULT);
serviceData.put(auxiliaryService,
    TezCommonUtils.serializeServiceData(TokenCache.getSessionToken(amLaunchCredentials)));

The comment right above it in the source explains why the AM needs a shuffle token: tasks can run inside the AM container, and the NM's shuffle aux-service on the AM's node must be able to authorize fetches against it. The same shuffle-token-in-serviceData pattern is used for every task container — built in AMContainerHelpers.createCommonContainerLaunchContext (tez-dag, org.apache.tez.dag.app.rm.container.AMContainerHelpers):

grep -n "serviceData\|auxiliaryService\|createCommonContainerLaunchContext" \
  tez-dag/src/main/java/org/apache/tez/dag/app/rm/container/AMContainerHelpers.java

AM boot: DAGAppMaster as a YARN AM

find tez-dag/src/main/java -name "DAGAppMaster.java"
grep -n "public static void main\|serviceInit\|serviceStart" \
  tez-dag/src/main/java/org/apache/tez/dag/app/DAGAppMaster.java | head

Boot sequence when YARN launches the AM container:

1. NodeManager runs the command line built by TezClientUtils, i.e. java ... org.apache.tez.dag.app.DAGAppMaster.
2. DAGAppMaster.main reads ApplicationAttemptId, container ID, AM Resource, and the NM host/port from the container environment YARN injects.
3. It constructs the service tree — dispatchers, state machines, the TaskSchedulerManager, the ContainerLauncherManager, and the history (ATS) service — via serviceInit.
4. serviceStart brings up the client RPC server and, through TaskSchedulerManager, registers with the RM.
5. In session mode it waits for DAG submissions over RPC; in non-session mode it picks up the single pre-submitted DAG.

The scheduler is not created directly by DAGAppMaster; it is instantiated by TaskSchedulerManager.instantiateSchedulers, which is what actually registers the AM with the RM. Follow the chain:

grep -n "instantiateSchedulers\|registerApplicationMaster" \
  tez-dag/src/main/java/org/apache/tez/dag/app/rm/TaskSchedulerManager.java

Which scheduler class is instantiated is configurable. The default is DagAwareYarnTaskScheduler (tez.am.yarn.scheduler.class / TEZ_AM_YARN_SCHEDULER_CLASS), with YarnTaskSchedulerService as the classic alternative — both implement AMRMClientAsync.CallbackHandler:

grep -n "TEZ_AM_YARN_SCHEDULER_CLASS\b" tez-api/src/main/java/org/apache/tez/dag/api/TezConfiguration.java
grep -n "class DagAwareYarnTaskScheduler\|class YarnTaskSchedulerService" \
  tez-dag/src/main/java/org/apache/tez/dag/app/rm/DagAwareYarnTaskScheduler.java \
  tez-dag/src/main/java/org/apache/tez/dag/app/rm/YarnTaskSchedulerService.java

AMRMClientAsync: register, allocate, heartbeat

The task scheduler is the AMRM callback handler. In YarnTaskSchedulerService (tez-dag, org.apache.tez.dag.app.rm.YarnTaskSchedulerService):

public class YarnTaskSchedulerService extends TaskScheduler
implements AMRMClientAsync.CallbackHandler {

It builds its async client through Tez's own subclass, TezAMRMClientAsync (tez-dag, org.apache.tez.dag.app.rm.TezAMRMClientAsync extends AMRMClientAsyncImpl):

grep -n "createAMRMClientAsync\|amRmClient\|setHeartbeatInterval" \
  tez-dag/src/main/java/org/apache/tez/dag/app/rm/YarnTaskSchedulerService.java | head

Register

From YarnTaskSchedulerService.start() (method start, same class):

amRmClient.start();
response = amRmClient.registerApplicationMaster(appHostName,
                                                appHostPort,
                                                appTrackingUrl);

The RegisterApplicationMasterResponse carries the maximum container Resource, the application ACLs, the client-to-AM token master key, and the queue — all pushed back into the AM via setApplicationRegistrationData. The heartbeat interval is capped by tez.am.am-rm.heartbeat.interval-ms.max (TEZ_AM_RM_HEARTBEAT_INTERVAL_MS_MAX):

grep -n "am-rm.heartbeat.interval-ms.max" tez-api/src/main/java/org/apache/tez/dag/api/TezConfiguration.java

The callbacks

AMRMClientAsync's internal heartbeat thread fires these on the scheduler. Tez keeps them short — mostly forwarding to its own dispatcher and doing allocation math under a lock:

grep -n "onContainersAllocated\|onContainersCompleted\|onNodesUpdated\|onShutdownRequest\|onError\|getProgress" \
  tez-dag/src/main/java/org/apache/tez/dag/app/rm/YarnTaskSchedulerService.java

• onContainersAllocated(List<Container>) — new containers to match against pending task requests (or release if the app is shutting down).
• onContainersCompleted(List<ContainerStatus>) — a container exited; its exit status decides how Tez classifies the loss (see below).
• onNodesUpdated(List<NodeReport>) — node health/decommission reports.
• onShutdownRequest() — the RM wants the AM to stop; forwarded as appShutdownRequested().
• onError(Throwable) — an RM-side error; reported as YarnTaskSchedulerServiceError.RESOURCEMANAGER_ERROR.
• getProgress() — returns app progress 0..1 and is the hook where Tez runs preemptIfNeeded() on each heartbeat.

Unregister

On clean shutdown the scheduler calls amRmClient.unregisterApplicationMaster(status, msg, trackingUrl) and then amRmClient.stop(). Failing to unregister (AM killed, GC pause past the liveness expiry) is what leaves an app in a FAILED/KILLED state with no final diagnostics.

sequenceDiagram
  participant AM as YarnTaskSchedulerService
  participant Async as TezAMRMClientAsync (heartbeat thread)
  participant RM as ResourceManager
  AM->>Async: registerApplicationMaster(host, port, url)
  Async->>RM: register
  RM-->>Async: maxResource, ACLs, clientToAMKey, queue
  loop every heartbeat
    Async->>RM: allocate(pending requests, releases, progress)
    RM-->>Async: allocated + completed containers, node reports
    Async->>AM: onContainersAllocated / onContainersCompleted / onNodesUpdated
  end
  AM->>Async: unregisterApplicationMaster(state, msg, url)
  Async->>RM: unregister

Launching containers: TezContainerLauncherImpl

Once the RM allocates a container, the AM must tell that container's NM to fork the JVM. That is TezContainerLauncherImpl (tez-dag, org.apache.tez.dag.app.launcher.TezContainerLauncherImpl), managed by ContainerLauncherManager. Note it uses ContainerManagementProtocolProxy directly (not NMClientAsync):

grep -n "ContainerManagementProtocolProxy\|startContainers\|stopContainers" \
  tez-dag/src/main/java/org/apache/tez/dag/app/launcher/TezContainerLauncherImpl.java

The launch call, from TezContainerLauncherImpl (inner launch path):

StartContainersResponse response =
    proxy.getContainerManagementProtocol().startContainers(
        StartContainersRequest.newInstance(
            Collections.singletonList(startRequest)));

A failed launch (NM rejects the request, or the container was killed before launch) is turned into a sendContainerLaunchFailedMsg, which the AM's container state machine consumes — see Failure Handling. The local/uber counterpart is LocalContainerLauncher (same package), used in local mode.

How Tez reacts when YARN misbehaves

This is the payoff section: match the YARN misbehavior to the exact Tez handler.

Preemption and disk failure

onContainersCompleted forwards each ContainerStatus to TaskSchedulerManager.containerCompleted, which classifies by ContainerExitStatus. From TaskSchedulerManager (tez-dag, org.apache.tez.dag.app.rm.TaskSchedulerManager, method containerCompleted):

if (exitStatus == ContainerExitStatus.PREEMPTED) {
  message = "Container preempted externally. ";
  errCause = TaskAttemptTerminationCause.EXTERNAL_PREEMPTION;
} else if (exitStatus == ContainerExitStatus.DISKS_FAILED) {
  message = "Container disk failed. ";
  errCause = TaskAttemptTerminationCause.NODE_DISK_ERROR;
}

The TaskAttemptTerminationCause propagates into the task-attempt state machine so a preempted attempt is retried without counting as a "real" failure — preemption is expected, not a bug.

Node loss and blacklisting

Two distinct mechanisms:

• RM-driven node updates arrive through onNodesUpdated and flow to the AM's node tracker.
• AM-driven blacklisting: repeated task-attempt failures on the same node trip AMNodeImpl (tez-dag, org.apache.tez.dag.app.rm.node.AMNodeImpl). Its decision is blacklistingEnabled && (numFailedTAs >= maxTaskFailuresPerNode):

grep -n "maxTaskFailuresPerNode\|blacklistingEnabled\|numFailedTAs" \
  tez-dag/src/main/java/org/apache/tez/dag/app/rm/node/AMNodeImpl.java

The knobs (all tez-api, TezConfiguration):

grep -n "TEZ_AM_NODE_BLACKLISTING_ENABLED\|TEZ_AM_MAX_TASK_FAILURES_PER_NODE\|TEZ_AM_NODE_BLACKLISTING_IGNORE_THRESHOLD\|TEZ_AM_NODE_UNHEALTHY_RESCHEDULE_TASKS" \
  tez-api/src/main/java/org/apache/tez/dag/api/TezConfiguration.java

The ignore-threshold exists to prevent a bad job from blacklisting the whole cluster: if a code bug fails every attempt, Tez must not conclude that every node is broken.

RM failover and AM re-attempt

If the RM restarts or the AM loses the RM, onError reports a RESOURCEMANAGER_ERROR. If the AM itself dies, YARN starts a new attempt up to tez.am.max.app.attempts (TEZ_AM_MAX_APP_ATTEMPTS, default 2), and the new AM rebuilds DAG state from what RecoveryService wrote — the recovery path covered in Failure Handling.

grep -n "TEZ_AM_MAX_APP_ATTEMPTS\b" tez-api/src/main/java/org/apache/tez/dag/api/TezConfiguration.java

AM GC pause → "RM expired"

getProgress() and the heartbeat run on the AMRM thread. A full-GC pause longer than YARN's AM liveness expiry means the RM stops seeing heartbeats and declares the AM dead, killing the container mid-DAG. The fix is AM heap tuning, not a Tez code change — but the symptom looks like a Tez crash.

The shuffle aux-service

Tez ships its own ShuffleHandler aux-service (tez-plugins, org.apache.tez.auxservices.ShuffleHandler extends AuxiliaryService) so the NM can serve map outputs to fetchers. Two identifiers matter and they are different:

grep -n "class ShuffleHandler\|TEZ_SHUFFLE_SERVICEID\|SHUFFLE_PORT_CONFIG_KEY\|DEFAULT_SHUFFLE_PORT" \
  tez-plugins/tez-aux-services/src/main/java/org/apache/tez/auxservices/ShuffleHandler.java
grep -n "TEZ_SHUFFLE_HANDLER_SERVICE_ID\|TEZ_AM_SHUFFLE_AUXILIARY_SERVICE_ID\b" \
  tez-api/src/main/java/org/apache/tez/dag/api/TezConstants.java \
  tez-api/src/main/java/org/apache/tez/dag/api/TezConfiguration.java

• ShuffleHandler.TEZ_SHUFFLE_SERVICEID = "tez_shuffle" — the aux-service id you register in yarn-site.xml if you deploy the Tez shuffle handler.
• TezConstants.TEZ_SHUFFLE_HANDLER_SERVICE_ID = "mapreduce_shuffle" — the default value of tez.am.shuffle-auxiliary-service-id (shuffle.auxiliary-service.id), i.e. the aux-service Tez asks the NM for when putting the shuffle token in serviceData.

The default port is 13563 (SHUFFLE_PORT_CONFIG_KEY = "tez.shuffle.port"). The mismatch between the two ids is the single most common Tez deployment error: a Fetcher: ConnectException almost always means the aux-service name/port that Tez requests does not match what the NM actually runs.

Deploy either the MapReduce mapreduce_shuffle handler (default, no extra config) or the Tez tez_shuffle handler (then set tez.am.shuffle-auxiliary-service-id=tez_shuffle and register tez_shuffle in yarn.nodemanager.aux-services). Do not mix them.

Tokens and security

Token collection happens client-side in TezClientUtils, mediated by TokenCache (tez-api, org.apache.tez.common.security.TokenCache):

grep -n "obtainTokensForFileSystems\|getSessionToken\|setSessionToken\|mergeBinaryTokens" \
  tez-api/src/main/java/org/apache/tez/common/security/TokenCache.java

On a secure cluster the RM renews the file-system delegation tokens it was handed at submit; note the comment in TokenCache that a token with an empty renewer is skipped by the RM. A ClientToAMToken auth failure almost always means client and AM disagree on hadoop.security.authentication.

The staging-dir lifecycle

The per-app staging dir (tez.staging-dir) is created client-side by ensureStagingDirExists, carries the framework tarball and serialized DAG, and is torn down by the AM on shutdown — but only if unregistration succeeded, guarded by tez.am.staging.scratch-data.auto-delete. From DAGAppMaster (tez-dag, org.apache.tez.dag.app.DAGAppMaster):

// Given pre-emption, we should delete tez scratch dir only if unregister is
// successful
boolean deleteTezScratchData = this.amConf.getBoolean(
    TezConfiguration.TEZ_AM_STAGING_SCRATCH_DATA_AUTO_DELETE,
    TezConfiguration.TEZ_AM_STAGING_SCRATCH_DATA_AUTO_DELETE_DEFAULT);

grep -n "TEZ_AM_STAGING_SCRATCH_DATA_AUTO_DELETE\b" \
  tez-api/src/main/java/org/apache/tez/dag/api/TezConfiguration.java

Consequence: a hard-killed AM (yarn application -kill, RM expiry, OOM) never runs the cleanup, so orphaned staging dirs accumulate under the parent path. In a busy HiveServer2 cluster this is a real disk-leak source — a periodic sweep of the staging parent is standard operational hygiene.

Timeline Server (ATS) and log aggregation

Two more YARN-facing surfaces matter for post-mortem debugging, and both are about where the evidence lives after the app is gone.

ATS. Tez publishes a rich event stream — DAG/vertex/task/attempt started and finished, container launched/stopped — to the YARN Timeline Server via a pluggable history service. The plugin is selected by tez.history.logging.service.class (TEZ_HISTORY_LOGGING_SERVICE_CLASS in tez-api), and the implementations live in tez-plugins:

ls tez-plugins/
grep -n "TEZ_HISTORY_LOGGING_SERVICE_CLASS\b" tez-api/src/main/java/org/apache/tez/dag/api/TezConfiguration.java
grep -rln "class ATSHistoryLoggingService\|class ATSV15HistoryLoggingService" \
  tez-plugins/tez-yarn-timeline-history*/src/main/java

The Tez UI reads these events to render the DAG view, task swimlanes, and counter trees; the UI location is surfaced in DAGStatus via tez.am.tez-ui.history-url.template (TEZ_AM_TEZ_UI_HISTORY_URL_TEMPLATE). An empty ATS for a completed app almost always means tez.history.logging.service.class is mis-set or the Timeline Server is not running — not that Tez failed to emit events.

Log aggregation. With yarn.log-aggregation-enable=true, every container's stdout, stderr, and syslog are uploaded to HDFS under yarn.nodemanager.remote-app-log-dir when the container exits, retrievable with yarn logs -applicationId <appId>. With it disabled, the logs sit in ${yarn.nodemanager.log-dirs}/<applicationId>/<containerId>/ on each NM until yarn.nodemanager.log.retain-seconds expires — which is where you go for a TezChild that crashed before it could aggregate. This is pure YARN configuration; Tez neither aggregates nor cleans these logs, it only writes to the redirections the CLC set up.

yarn CLI behaviors for Tez apps

grep -n "TEZ_APPLICATION_TYPE\b" tez-api/src/main/java/org/apache/tez/dag/api/TezConstants.java

Reading exercise

Run all from ~/tez-src:

1. grep -n "registerApplicationMaster\|unregisterApplicationMaster" \ tez-dag/src/main/java/org/apache/tez/dag/app/rm/YarnTaskSchedulerService.java — read the surrounding start() and stop() methods; what does the register response give back to the AM?
2. Read createApplicationSubmissionContext in TezClientUtils. List every argument passed to ContainerLaunchContext.newInstance.
3. grep -n "ContainerExitStatus" tez-dag/src/main/java/org/apache/tez/dag/app/rm/TaskSchedulerManager.java — enumerate every exit status Tez special-cases and the TaskAttemptTerminationCause it maps to.
4. Read AMNodeImpl around the blacklisting predicate. Under what two conditions does a node get blacklisted, and what disables it cluster-wide?
5. grep -n "TEZ_SHUFFLE_SERVICEID\|SHUFFLE_PORT_CONFIG_KEY" \ tez-plugins/tez-aux-services/src/main/java/org/apache/tez/auxservices/ShuffleHandler.java — reconcile tez_shuffle vs the requested mapreduce_shuffle default.
6. Read the staging-delete block in DAGAppMaster. Why is deletion gated on successful unregistration?

Common bugs and symptoms

Validation: prove you understand this

1. Trace the call path from DAGAppMaster.serviceStart to AMRMClientAsync.registerApplicationMaster, naming the intermediate class that actually creates the scheduler.
2. List the six arguments to ContainerLaunchContext.newInstance in the AM CLC and say who consumes serviceData.
3. A container completes with ContainerExitStatus.PREEMPTED. Name the Tez TaskAttemptTerminationCause it maps to and why the attempt is retried without penalty. Cite the class and method.
4. Explain the difference between tez_shuffle and the default mapreduce_shuffle id, and give the exact config change to run Tez's own shuffle handler.
5. Why does an AM full-GC pause manifest as "RM expired", and which side (Tez config or YARN config) owns the threshold?
6. Under what condition does the staging dir survive AM shutdown, and what operational problem does that create on a busy HS2 cluster?