client is the internal reference to YarnRMClient that ApplicationMaster is given when created.

client is primarily used to register the ApplicationMaster and request containers for executors from YARN and later unregister ApplicationMaster from YARN ResourceManager.

Besides, it helps obtaining an application attempt id and the allowed number of attempts to register ApplicationMaster. It also gets filter parameters to secure ApplicationMaster’s UI.

allocator is the internal reference to YarnAllocator that ApplicationMaster uses to request new or release outstanding containers for executors.

It is created when ApplicationMaster is registered (using the internal YarnRMClient reference).

ApplicationMaster is started as a standalone command-line application inside a YARN container on a node.

When executed, main first parses command-line parameters and then uses SparkHadoopUtil.runAsSparkUser to run the main code with a Hadoop UserGroupInformation as a thread local variable (distributed to child threads) for authenticating HDFS and YARN calls.

You should see the following message in the logs:

SparkHadoopUtil.runAsSparkUser function executes a block that creates a ApplicationMaster (passing the ApplicationMasterArguments instance and a brand new YarnRMClient) and then runs it.

ApplicationMaster uses ApplicationMasterArguments class to handle command-line parameters.

ApplicationMasterArguments is created right after main method has been executed for args command-line parameters.

It accepts the following command-line parameters:

When an unsupported parameter is found the following message is printed out to standard error output and ApplicationMaster exits with the exit code 1.

When runDriver or runExecutorLauncher are executed, they use the private helper procedure registerAM to register the ApplicationMaster (with the YARN ResourceManager) and request resources (given hints about where to allocate containers to be as close to the data as possible).

Internally, it first reads spark.yarn.historyServer.address setting and substitute Hadoop variables to create a complete address of the History Server, i.e. [address]/history/[appId]/[attemptId].

Then, registerAM creates a RpcEndpointAddress for CoarseGrainedScheduler RPC Endpoint on the driver available on spark.driver.host and spark.driver.port.

It registers the ApplicationMaster with the YARN ResourceManager and request resources (given hints about where to allocate containers to be as close to the data as possible).

Ultimately, registerAM launches reporter thread.

runDriver is a private procedure to…​???

It starts by registering Web UI security filters.

It then starts the user class (with the driver) in a separate thread. You should see the following INFO message in the logs:

You should see the following INFO message in the logs:

runExecutorLauncher reads spark.yarn.am.port (or assume 0) and starts the sparkYarnAM RPC Environment (in client mode).

It then waits for the driver to be available.

It registers Web UI security filters.

Ultimately, runExecutorLauncher registers the ApplicationMaster and requests resources and waits until the reporterThread dies.

sparkContextInitialized passes the call on to the ApplicationMaster.sparkContextInitialized that sets the internal sparkContextRef reference (to be sc).

sparkContextStopped passes the call on to the ApplicationMaster.sparkContextStopped that clears the internal sparkContextRef reference (i.e. sets it to null).

When creating an instance of ApplicationMaster it requires ApplicationMasterArguments and YarnRMClient.

It instantiates SparkConf and Hadoop’s YarnConfiguration (using SparkHadoopUtil.newConfiguration).

It assumes cluster deploy mode when --class was specified.

It computes the internal maxNumExecutorFailures using the optional spark.yarn.max.executor.failures if set. Otherwise, it is twice spark.executor.instances or spark.dynamicAllocation.maxExecutors (with dynamic allocation enabled) with the minimum of 3.

It reads yarn.am.liveness-monitor.expiry-interval-ms (default: 120000) from YARN to set the heartbeat interval. It is set to the minimum of the half of the YARN setting or spark.yarn.scheduler.heartbeat.interval-ms with the minimum of 0.

initialAllocationInterval is set to the minimum of the heartbeat interval or spark.yarn.scheduler.initial-allocation.interval.

It then loads the localized files (as set by the client).

When ApplicationMaster is instantiated, it computes internal localResources collection of YARN’s LocalResource by name based on the internal spark.yarn.cache.* configuration settings.

You should see the following INFO message in the logs:

It starts by reading the internal Spark configuration settings (that were earlier set when Client prepared local resources to distribute):

For each file name in spark.yarn.cache.filenames it maps spark.yarn.cache.types to an appropriate YARN’s LocalResourceType and creates a new YARN LocalResource.

If spark.yarn.cache.confArchive is set, it is added to localResources as ARCHIVE resource type and PRIVATE visibility.

Ultimately, it removes the cache-related settings from the Spark configuration and system properties.

You should see the following INFO message in the logs:

When ApplicationMaster is started as a standalone command-line application (in a YARN container on a node in a YARN cluster), ultimately run is executed.

The result of calling run is the final result of the ApplicationMaster command-line application.

run sets cluster mode settings, registers the cleanup shutdown hook, schedules AMDelegationTokenRenewer and finally registers ApplicationMaster for the Spark application (either calling runDriver for cluster mode or runExecutorLauncher for client mode).

After the cluster mode settings are set, run prints the following INFO message out to the logs:

The appAttemptId is the current application attempt id (using the constructor’s YarnRMClient as client).

The cleanup shutdown hook is registered with shutdown priority lower than that of SparkContext (so it is executed after SparkContext).

SecurityManager is instantiated with the internal Spark configuration. If the credentials file config (as spark.yarn.credentials.file) is present, a AMDelegationTokenRenewer is started.

It finally runs ApplicationMaster for the Spark application (either calling runDriver when in cluster mode or runExecutorLauncher otherwise).

It exits with 0 exit code.

In case of an exception, run prints the following ERROR message out to the logs:

And the application run attempt is finished with FAILED status and EXIT_UNCAUGHT_EXCEPTION (10) exit code.

When in cluster mode, ApplicationMaster sets the following system properties (in run):

isClusterMode is an internal flag that is enabled (i.e. true) for cluster mode.

Specifically, it says whether the main class of the Spark application (through --class command-line argument) was specified or not. That is how the developers decided to inform ApplicationMaster about being run in cluster mode when Client creates YARN’s ContainerLaunchContext (for launching ApplicationMaster).

It is used to set additional system properties in run and runDriver (the flag is enabled) or runExecutorLauncher (when disabled).

Besides, it controls the default final status of a Spark application being FinalApplicationStatus.FAILED (when the flag is enabled) or FinalApplicationStatus.UNDEFINED.

The flag also controls whether to set system properties in addAmIpFilter (when the flag is enabled) or send a AddWebUIFilter instead.

unregister unregisters the ApplicationMaster for the Spark application from the YARN ResourceManager.

It first checks that the ApplicationMaster has not already been unregistered (using the internal unregistered flag). If so, you should see the following INFO message in the logs:

There can also be an optional diagnostic message in the logs:

The internal unregistered flag is set to be enabled, i.e. true.

It then requests YarnRMClient to unregister.

When ApplicationMaster starts running, it registers a shutdown hook that unregisters the Spark application from the YARN ResourceManager and cleans up the staging directory.

Internally, it checks the internal finished flag, and if it is disabled, it marks the Spark application as failed with EXIT_EARLY.

If the internal unregistered flag is disabled, it unregisters the Spark application and cleans up the staging directory afterwards only when the final status of the ApplicationMaster’s registration is FinalApplicationStatus.SUCCEEDED or the number of application attempts is more than allowed.

The shutdown hook runs after the SparkContext is shut down, i.e. the shutdown priority is one less than SparkContext’s.

The shutdown hook is registered using Spark’s own ShutdownHookManager.addShutdownHook.

ExecutorLauncher comes with no extra functionality when compared to ApplicationMaster. It serves as a helper class to run ApplicationMaster under another class name in client deploy mode.

With the two different class names (pointing at the same class ApplicationMaster) you should be more successful to distinguish between ExecutorLauncher (which is really a ApplicationMaster) in client deploy mode and the ApplicationMaster in cluster deploy mode using tools like ps or jps.

getAttemptId returns YARN’s ApplicationAttemptId (of the Spark application to which the container was assigned).

Internally, it queries YARN by means of YarnRMClient.

addAmIpFilter is a helper method that …​???

It starts by reading Hadoop’s environmental variable ApplicationConstants.APPLICATION_WEB_PROXY_BASE_ENV that it passes to YarnRMClient to compute the configuration for the AmIpFilter for web UI.

In cluster deploy mode (when ApplicationMaster runs with web UI), it sets spark.ui.filters system property as org.apache.hadoop.yarn.server.webproxy.amfilter.AmIpFilter. It also sets system properties from the key-value configuration of AmIpFilter (computed earlier) as spark.org.apache.hadoop.yarn.server.webproxy.amfilter.AmIpFilter.param.[key] being [value].

In client deploy mode (when ApplicationMaster runs on another JVM or even host than web UI), it simply sends a AddWebUIFilter to ApplicationMaster (namely to AMEndpoint RPC Endpoint).