org.apache.flink.yarn

Class YarnFlinkResourceManager

java.lang.Object

akka.actor.UntypedActor

org.apache.flink.runtime.akka.FlinkUntypedActor

org.apache.flink.runtime.clusterframework.FlinkResourceManager<RegisteredYarnWorkerNode>

org.apache.flink.yarn.YarnFlinkResourceManager

All Implemented Interfaces:

akka.actor.Actor

public class YarnFlinkResourceManager
extends org.apache.flink.runtime.clusterframework.FlinkResourceManager<RegisteredYarnWorkerNode>

Specialized Flink Resource Manager implementation for YARN clusters. It is started as the YARN ApplicationMaster and implements the YARN-specific logic for container requests and failure monitoring.

Nested Class Summary

Nested classes/interfaces inherited from interface akka.actor.Actor

akka.actor.Actor.emptyBehavior$, akka.actor.Actor.ignoringBehavior$

Field Summary

Fields inherited from class org.apache.flink.runtime.clusterframework.FlinkResourceManager

config, EXIT_CODE_FATAL_ERROR, RESOURCE_MANAGER_NAME

Fields inherited from class org.apache.flink.runtime.akka.FlinkUntypedActor

LOG

Constructor Summary

Constructors

Constructor and Description
YarnFlinkResourceManager(org.apache.flink.configuration.Configuration flinkConfig, org.apache.hadoop.yarn.conf.YarnConfiguration yarnConfig, org.apache.flink.runtime.leaderretrieval.LeaderRetrievalService leaderRetrievalService, String applicationMasterHostName, String webInterfaceURL, org.apache.flink.runtime.clusterframework.ContaineredTaskManagerParameters taskManagerParameters, org.apache.hadoop.yarn.api.records.ContainerLaunchContext taskManagerLaunchContext, int yarnHeartbeatIntervalMillis, int maxFailedContainers, int numInitialTaskManagers)
YarnFlinkResourceManager(org.apache.flink.configuration.Configuration flinkConfig, org.apache.hadoop.yarn.conf.YarnConfiguration yarnConfig, org.apache.flink.runtime.leaderretrieval.LeaderRetrievalService leaderRetrievalService, String applicationMasterHostName, String webInterfaceURL, org.apache.flink.runtime.clusterframework.ContaineredTaskManagerParameters taskManagerParameters, org.apache.hadoop.yarn.api.records.ContainerLaunchContext taskManagerLaunchContext, int yarnHeartbeatIntervalMillis, int maxFailedContainers, int numInitialTaskManagers, YarnResourceManagerCallbackHandler callbackHandler)
YarnFlinkResourceManager(org.apache.flink.configuration.Configuration flinkConfig, org.apache.hadoop.yarn.conf.YarnConfiguration yarnConfig, org.apache.flink.runtime.leaderretrieval.LeaderRetrievalService leaderRetrievalService, String applicationMasterHostName, String webInterfaceURL, org.apache.flink.runtime.clusterframework.ContaineredTaskManagerParameters taskManagerParameters, org.apache.hadoop.yarn.api.records.ContainerLaunchContext taskManagerLaunchContext, int yarnHeartbeatIntervalMillis, int maxFailedContainers, int numInitialTaskManagers, YarnResourceManagerCallbackHandler callbackHandler, org.apache.hadoop.yarn.client.api.async.AMRMClientAsync<org.apache.hadoop.yarn.client.api.AMRMClient.ContainerRequest> resourceManagerClient, org.apache.hadoop.yarn.client.api.NMClient nodeManagerClient)

Method Summary

Modifier and Type	Method and Description
static akka.actor.Props	createActorProps(Class<? extends YarnFlinkResourceManager> actorClass, org.apache.flink.configuration.Configuration flinkConfig, org.apache.hadoop.yarn.conf.YarnConfiguration yarnConfig, org.apache.flink.runtime.leaderretrieval.LeaderRetrievalService leaderRetrievalService, String applicationMasterHostName, String webFrontendURL, org.apache.flink.runtime.clusterframework.ContaineredTaskManagerParameters taskManagerParameters, org.apache.hadoop.yarn.api.records.ContainerLaunchContext taskManagerLaunchContext, int numInitialTaskManagers, org.slf4j.Logger log) Creates the props needed to instantiate this actor.
protected void	fatalError(String message, Throwable error)
protected int	getNumWorkerRequestsPending()
protected int	getNumWorkersPendingRegistration()
protected void	handleMessage(Object message)
protected void	initialize()
protected Collection<RegisteredYarnWorkerNode>	reacceptRegisteredWorkers(Collection<org.apache.flink.runtime.clusterframework.types.ResourceID> toConsolidate)
protected void	releasePendingWorker(org.apache.flink.runtime.clusterframework.types.ResourceID id)
protected void	releaseStartedWorker(RegisteredYarnWorkerNode worker)
protected void	requestNewWorkers(int numWorkers)
protected void	shutdownApplication(org.apache.flink.runtime.clusterframework.ApplicationStatus finalStatus, String optionalDiagnostics)
protected RegisteredYarnWorkerNode	workerStarted(org.apache.flink.runtime.clusterframework.types.ResourceID resourceID)

Methods inherited from class org.apache.flink.runtime.clusterframework.FlinkResourceManager

allStartedWorkers, getDesignatedWorkerPoolSize, getLeaderSessionID, getNumberOfStartedTaskManagers, getResourceManagerProps, getStartedTaskManagers, isStarted, notifyWorkerFailed, postStop, preStart, sendInfoMessage, startResourceManagerActors, startResourceManagerActors, triggerCheckWorkers, triggerConnectingToJobManager

Methods inherited from class org.apache.flink.runtime.akka.FlinkUntypedActor

decorateMessage, onReceive

Methods inherited from class akka.actor.UntypedActor

akka$actor$Actor$_setter_$context_$eq, akka$actor$Actor$_setter_$self_$eq, aroundPostRestart, aroundPostStop, aroundPreRestart, aroundPreStart, aroundReceive, context, getContext, getSelf, getSender, postRestart, preRestart, receive, self, sender, supervisorStrategy, unhandled

Methods inherited from class java.lang.Object

clone, equals, finalize, getClass, hashCode, notify, notifyAll, toString, wait, wait, wait

Constructor Detail

YarnFlinkResourceManager

public YarnFlinkResourceManager(org.apache.flink.configuration.Configuration flinkConfig,
                                org.apache.hadoop.yarn.conf.YarnConfiguration yarnConfig,
                                org.apache.flink.runtime.leaderretrieval.LeaderRetrievalService leaderRetrievalService,
                                String applicationMasterHostName,
                                String webInterfaceURL,
                                org.apache.flink.runtime.clusterframework.ContaineredTaskManagerParameters taskManagerParameters,
                                org.apache.hadoop.yarn.api.records.ContainerLaunchContext taskManagerLaunchContext,
                                int yarnHeartbeatIntervalMillis,
                                int maxFailedContainers,
                                int numInitialTaskManagers)

YarnFlinkResourceManager

public YarnFlinkResourceManager(org.apache.flink.configuration.Configuration flinkConfig,
                                org.apache.hadoop.yarn.conf.YarnConfiguration yarnConfig,
                                org.apache.flink.runtime.leaderretrieval.LeaderRetrievalService leaderRetrievalService,
                                String applicationMasterHostName,
                                String webInterfaceURL,
                                org.apache.flink.runtime.clusterframework.ContaineredTaskManagerParameters taskManagerParameters,
                                org.apache.hadoop.yarn.api.records.ContainerLaunchContext taskManagerLaunchContext,
                                int yarnHeartbeatIntervalMillis,
                                int maxFailedContainers,
                                int numInitialTaskManagers,
                                YarnResourceManagerCallbackHandler callbackHandler)

YarnFlinkResourceManager

public YarnFlinkResourceManager(org.apache.flink.configuration.Configuration flinkConfig,
                                org.apache.hadoop.yarn.conf.YarnConfiguration yarnConfig,
                                org.apache.flink.runtime.leaderretrieval.LeaderRetrievalService leaderRetrievalService,
                                String applicationMasterHostName,
                                String webInterfaceURL,
                                org.apache.flink.runtime.clusterframework.ContaineredTaskManagerParameters taskManagerParameters,
                                org.apache.hadoop.yarn.api.records.ContainerLaunchContext taskManagerLaunchContext,
                                int yarnHeartbeatIntervalMillis,
                                int maxFailedContainers,
                                int numInitialTaskManagers,
                                YarnResourceManagerCallbackHandler callbackHandler,
                                org.apache.hadoop.yarn.client.api.async.AMRMClientAsync<org.apache.hadoop.yarn.client.api.AMRMClient.ContainerRequest> resourceManagerClient,
                                org.apache.hadoop.yarn.client.api.NMClient nodeManagerClient)

Method Detail

handleMessage

protected void handleMessage(Object message)

Overrides:

handleMessage in class org.apache.flink.runtime.clusterframework.FlinkResourceManager<RegisteredYarnWorkerNode>

initialize

protected void initialize()
                   throws Exception

Specified by:

initialize in class org.apache.flink.runtime.clusterframework.FlinkResourceManager<RegisteredYarnWorkerNode>

Throws:

Exception

shutdownApplication

protected void shutdownApplication(org.apache.flink.runtime.clusterframework.ApplicationStatus finalStatus,
                                   String optionalDiagnostics)

Specified by:

shutdownApplication in class org.apache.flink.runtime.clusterframework.FlinkResourceManager<RegisteredYarnWorkerNode>

fatalError

protected void fatalError(String message,
                          Throwable error)

Specified by:

fatalError in class org.apache.flink.runtime.clusterframework.FlinkResourceManager<RegisteredYarnWorkerNode>

requestNewWorkers

protected void requestNewWorkers(int numWorkers)

Specified by:

requestNewWorkers in class org.apache.flink.runtime.clusterframework.FlinkResourceManager<RegisteredYarnWorkerNode>

releasePendingWorker

protected void releasePendingWorker(org.apache.flink.runtime.clusterframework.types.ResourceID id)

Specified by:

releasePendingWorker in class org.apache.flink.runtime.clusterframework.FlinkResourceManager<RegisteredYarnWorkerNode>

releaseStartedWorker

protected void releaseStartedWorker(RegisteredYarnWorkerNode worker)

Specified by:

releaseStartedWorker in class org.apache.flink.runtime.clusterframework.FlinkResourceManager<RegisteredYarnWorkerNode>

workerStarted

protected RegisteredYarnWorkerNode workerStarted(org.apache.flink.runtime.clusterframework.types.ResourceID resourceID)

Specified by:

workerStarted in class org.apache.flink.runtime.clusterframework.FlinkResourceManager<RegisteredYarnWorkerNode>

reacceptRegisteredWorkers

protected Collection<RegisteredYarnWorkerNode> reacceptRegisteredWorkers(Collection<org.apache.flink.runtime.clusterframework.types.ResourceID> toConsolidate)

Specified by:

reacceptRegisteredWorkers in class org.apache.flink.runtime.clusterframework.FlinkResourceManager<RegisteredYarnWorkerNode>

getNumWorkerRequestsPending

protected int getNumWorkerRequestsPending()

Specified by:

getNumWorkerRequestsPending in class org.apache.flink.runtime.clusterframework.FlinkResourceManager<RegisteredYarnWorkerNode>

getNumWorkersPendingRegistration

protected int getNumWorkersPendingRegistration()

Specified by:

getNumWorkersPendingRegistration in class org.apache.flink.runtime.clusterframework.FlinkResourceManager<RegisteredYarnWorkerNode>

createActorProps

public static akka.actor.Props createActorProps(Class<? extends YarnFlinkResourceManager> actorClass,
                                                org.apache.flink.configuration.Configuration flinkConfig,
                                                org.apache.hadoop.yarn.conf.YarnConfiguration yarnConfig,
                                                org.apache.flink.runtime.leaderretrieval.LeaderRetrievalService leaderRetrievalService,
                                                String applicationMasterHostName,
                                                String webFrontendURL,
                                                org.apache.flink.runtime.clusterframework.ContaineredTaskManagerParameters taskManagerParameters,
                                                org.apache.hadoop.yarn.api.records.ContainerLaunchContext taskManagerLaunchContext,
                                                int numInitialTaskManagers,
                                                org.slf4j.Logger log)

Creates the props needed to instantiate this actor.

Rather than extracting and validating parameters in the constructor, this factory method takes care of that. That way, errors occur synchronously, and are not swallowed simply in a failed asynchronous attempt to start the actor.

Parameters:

actorClass - The actor class, to allow overriding this actor with subclasses for testing.

flinkConfig - The Flink configuration object.

yarnConfig - The YARN configuration object.

applicationMasterHostName - The hostname where this application master actor runs.

webFrontendURL - The URL of the tracking web frontend.

taskManagerParameters - The parameters for launching TaskManager containers.

taskManagerLaunchContext - The parameters for launching the TaskManager processes in the TaskManager containers.

numInitialTaskManagers - The initial number of TaskManagers to allocate.

log - The logger to log to.

Returns:

The Props object to instantiate the YarnFlinkResourceManager actor.