ApplicationMaster

Computed 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.

ApplicationMaster takes the following when created:

ApplicationMaster initializes the internal registries and counters.

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).

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).

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

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

ApplicationMaster is started as a standalone 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 new YarnRMClient) and then runs it.

run reads the application attempt id.

(only in cluster deploy mode) run sets cluster deploy mode-specific settings and sets the application attempt id (from YARN).

run sets a CallerContext for APPMASTER.

You should see the following INFO message in the logs:

run creates a Hadoop FileSystem (using the internal YarnConfiguration).

run registers the cleanup shutdown hook.

run creates a SecurityManager.

(only when spark.yarn.credentials.file is defined) run creates a ConfigurableCredentialManager to get a AMCredentialRenewer and schedules login from keytab.

In the end, run registers ApplicationMaster (with YARN ResourceManager) for the Spark application — either calling runDriver (in cluster deploy mode) or runExecutorLauncher (for client deploy mode).

run exits with 0 exit code.

In case of an exception, you should see the following ERROR message in the logs and run finishes with FAILED final application status.

runExecutorLauncher creates sparkYarnAM RPC environment (on spark.yarn.am.port port, the internal SparkConf and clientMode enabled).

runExecutorLauncher then waits until the driver accepts connections and creates RpcEndpointRef to communicate.

runExecutorLauncher registers web UI security filters.

In the end, runExecutorLauncher registers ApplicationMaster with YARN ResourceManager and requests resources and then pauses until reporterThread finishes.

runDriver starts a Spark application on a separate thread, registers YarnAM endpoint in the application’s RpcEnv followed by registering ApplicationMaster with YARN ResourceManager. In the end, runDriver waits for the Spark application to finish.

Internally, runDriver registers web UI security filters and starts a Spark application (on a separate Thread).

You should see the following INFO message in the logs:

runDriver waits spark.yarn.am.waitTime time till the Spark application’s SparkContext is available and accesses the current RpcEnv (and saves it as the internal rpcEnv).

runDriver creates RpcEndpointRef to the driver’s YarnScheduler endpoint and registers YarnAM endpoint (using spark.driver.host and spark.driver.port properties for the driver’s host and port and isClusterMode enabled).

runDriver registers ApplicationMaster with YARN ResourceManager and requests cluster resources (using the Spark application’s RpcEnv, the driver’s RPC endpoint reference, webUrl if web UI is enabled and the input securityMgr).

runDriver pauses until the Spark application finishes.

If the Spark application has not started in spark.yarn.am.waitTime time, runDriver reports a IllegalStateException:

If TimeoutException is reported while waiting for the Spark application to start, you should see the following ERROR message in the logs and runDriver finishes with FAILED final application status and the error code 13.

startUserApplication starts a Spark application as a separate Driver thread.

Internally, when startUserApplication is executed, you should see the following INFO message in the logs:

startUserApplication takes the user-specified jars and maps them to use the file: protocol.

startUserApplication then creates a class loader to load the main class of the Spark application given the precedence of the Spark system jars and the user-specified jars.

startUserApplication works on custom configurations for Python and R applications (which I don’t bother including here).

startUserApplication loads the main class (using the custom class loader created above with the user-specified jars) and creates a reference to the main method.

startUserApplication starts a Java Thread (with the name Driver) that invokes the main method (with the application arguments from userArgs from ApplicationMasterArguments). The Driver thread uses the internal sparkContextPromise to notify ApplicationMaster about the execution status of the main method (success or failure).

When the main method (of the Spark application) finishes successfully, the Driver thread will finish with SUCCEEDED final application status and code status 0 and you should see the following DEBUG message in the logs:

Any exceptions in the Driver thread are reported with corresponding ERROR message in the logs, FAILED final application status, appropriate code status.

Internally, registerAM first takes the application and attempt ids, and creates the URL of Spark History Server for the Spark application, i.e. [address]/history/[appId]/[attemptId], by substituting Hadoop variables (using the internal YarnConfiguration) in the optional spark.yarn.historyServer.address setting.

registerAM then creates a RpcEndpointAddress for the driver’s CoarseGrainedScheduler RPC endpoint available at spark.driver.host and spark.driver.port.

registerAM prints YARN launch context diagnostic information (with command, environment and resources) for executors (with spark.executor.memory, spark.executor.cores and dummy <executorId> and <hostname>)

registerAM requests YarnRMClient to register ApplicationMaster (with YARN ResourceManager) and the internal YarnAllocator to allocate required cluster resources (given placement hints about where to allocate resource containers for executors to be as close to the data as possible).

In the end, registerAM launches reporter thread.

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 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 in cluster deploy 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 (to launch the ApplicationMaster for a Spark application).

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

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

isClusterMode 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.

waitForSparkDriver waits until the driver is network-accessible, i.e. accepts connections on a given host and port, and returns a RpcEndpointRef to the driver.

When executed, you should see the following INFO message in the logs:

waitForSparkDriver takes the driver’s host and port (using ApplicationMasterArguments passed in when ApplicationMaster was created).

waitForSparkDriver tries to connect to the driver’s host and port until the driver accepts the connection but no longer than spark.yarn.am.waitTime setting or finished internal flag is enabled.

You should see the following INFO message in the logs:

While waitForSparkDriver tries to connect (while the socket is down), you can see the following ERROR message and waitForSparkDriver pauses for 100 ms and tries to connect again (until the waitTime elapses).

Once waitForSparkDriver could connect to the driver, waitForSparkDriver sets spark.driver.host and spark.driver.port properties to driverHost and driverPort, respectively (using the internal SparkConf).

In the end, waitForSparkDriver runAMEndpoint.

If waitForSparkDriver did not manage to connect (before waitTime elapses or finished internal flag was enabled), waitForSparkDriver reports a SparkException:

runAMEndpoint sets up a RpcEndpointRef to the driver’s YarnScheduler endpoint and registers YarnAM endpoint.

Internally, runAMEndpoint gets a RpcEndpointRef to the driver’s YarnScheduler endpoint (available on the host and port).

runAMEndpoint then registers the RPC endpoint as YarnAM (and AMEndpoint implementation with ApplicationMaster‘s RpcEnv, YarnScheduler endpoint reference, and isClusterMode flag).