org.apache.flink.yarn

Class AbstractYarnClusterDescriptor

java.lang.Object

org.apache.flink.yarn.AbstractYarnClusterDescriptor

All Implemented Interfaces:

AutoCloseable, org.apache.flink.client.deployment.ClusterDescriptor<org.apache.hadoop.yarn.api.records.ApplicationId>

Direct Known Subclasses:

LegacyYarnClusterDescriptor, YarnClusterDescriptor

public abstract class AbstractYarnClusterDescriptor
extends Object
implements org.apache.flink.client.deployment.ClusterDescriptor<org.apache.hadoop.yarn.api.records.ApplicationId>

The descriptor with deployment information for spawning or resuming a YarnClusterClient.

Field Summary

Fields

Modifier and Type	Field and Description
protected List<File>	shipFiles Lazily initialized list of files to ship.

Constructor Summary

Constructors

Constructor and Description
AbstractYarnClusterDescriptor(org.apache.flink.configuration.Configuration flinkConfiguration, org.apache.hadoop.yarn.conf.YarnConfiguration yarnConfiguration, String configurationDirectory, org.apache.hadoop.yarn.client.api.YarnClient yarnClient, boolean sharedYarnClient)

Method Summary

Modifier and Type	Method and Description
protected void	addLibFolderToShipFiles(Collection<File> effectiveShipFiles)
void	addShipFiles(List<File> shipFiles) Adds the given files to the list of files to ship.
void	close()
protected abstract org.apache.flink.client.program.ClusterClient<org.apache.hadoop.yarn.api.records.ApplicationId>	createYarnClusterClient(AbstractYarnClusterDescriptor descriptor, int numberTaskManagers, int slotsPerTaskManager, org.apache.hadoop.yarn.api.records.ApplicationReport report, org.apache.flink.configuration.Configuration flinkConfiguration, boolean perJobCluster) Creates a YarnClusterClient; may be overriden in tests.
protected org.apache.flink.client.program.ClusterClient<org.apache.hadoop.yarn.api.records.ApplicationId>	deployInternal(org.apache.flink.client.deployment.ClusterSpecification clusterSpecification, String applicationName, String yarnClusterEntrypoint, org.apache.flink.runtime.jobgraph.JobGraph jobGraph, boolean detached) This method will block until the ApplicationMaster/JobManager have been deployed on YARN.
org.apache.flink.client.program.ClusterClient<org.apache.hadoop.yarn.api.records.ApplicationId>	deploySessionCluster(org.apache.flink.client.deployment.ClusterSpecification clusterSpecification)
String	getClusterDescription()
String	getDynamicPropertiesEncoded()
org.apache.flink.configuration.Configuration	getFlinkConfiguration()
String	getNodeLabel()
org.apache.hadoop.yarn.client.api.YarnClient	getYarnClient()
protected abstract String	getYarnJobClusterEntrypoint() The class to start the application master with.
protected abstract String	getYarnSessionClusterEntrypoint() The class to start the application master with.
String	getZookeeperNamespace()
boolean	hasUserJarFiles(List<URL> requiredJarFiles) Returns true if the descriptor has the job jars to include in the classpath.
boolean	isDetachedMode() Deprecated. The cluster descriptor should not know about this option.
void	killCluster(org.apache.hadoop.yarn.api.records.ApplicationId applicationId)
org.apache.flink.client.program.ClusterClient<org.apache.hadoop.yarn.api.records.ApplicationId>	retrieve(org.apache.hadoop.yarn.api.records.ApplicationId applicationId)
void	setDetachedMode(boolean detachedMode) Deprecated. The cluster descriptor should not know about this option.
void	setDynamicPropertiesEncoded(String dynamicPropertiesEncoded)
void	setLocalJarPath(org.apache.hadoop.fs.Path localJarPath)
void	setName(String name)
void	setNodeLabel(String nodeLabel)
void	setProvidedUserJarFiles(List<URL> userJarFiles) Sets the user jar which is included in the system classloader of all nodes.
void	setQueue(String queue)
protected org.apache.hadoop.yarn.api.records.ContainerLaunchContext	setupApplicationMasterContainer(String yarnClusterEntrypoint, boolean hasLogback, boolean hasLog4j, boolean hasKrb5, int jobManagerMemoryMb)
void	setZookeeperNamespace(String zookeeperNamespace)
org.apache.hadoop.yarn.api.records.ApplicationReport	startAppMaster(org.apache.flink.configuration.Configuration configuration, String applicationName, String yarnClusterEntrypoint, org.apache.flink.runtime.jobgraph.JobGraph jobGraph, org.apache.hadoop.yarn.client.api.YarnClient yarnClient, org.apache.hadoop.yarn.client.api.YarnClientApplication yarnApplication, org.apache.flink.client.deployment.ClusterSpecification clusterSpecification)
protected org.apache.flink.client.deployment.ClusterSpecification	validateClusterResources(org.apache.flink.client.deployment.ClusterSpecification clusterSpecification, int yarnMinAllocationMB, org.apache.hadoop.yarn.api.records.Resource maximumResourceCapability, org.apache.flink.yarn.AbstractYarnClusterDescriptor.ClusterResourceDescription freeClusterResources)

Methods inherited from class java.lang.Object

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

Methods inherited from interface org.apache.flink.client.deployment.ClusterDescriptor

deployJobCluster

Field Detail

shipFiles

protected List<File> shipFiles

Lazily initialized list of files to ship.

Constructor Detail

AbstractYarnClusterDescriptor

public AbstractYarnClusterDescriptor(org.apache.flink.configuration.Configuration flinkConfiguration,
                                     org.apache.hadoop.yarn.conf.YarnConfiguration yarnConfiguration,
                                     String configurationDirectory,
                                     org.apache.hadoop.yarn.client.api.YarnClient yarnClient,
                                     boolean sharedYarnClient)

Method Detail

getYarnClient

public org.apache.hadoop.yarn.client.api.YarnClient getYarnClient()

getYarnSessionClusterEntrypoint

protected abstract String getYarnSessionClusterEntrypoint()

The class to start the application master with. This class runs the main method in case of session cluster.

getYarnJobClusterEntrypoint

protected abstract String getYarnJobClusterEntrypoint()

The class to start the application master with. This class runs the main method in case of the job cluster.

getFlinkConfiguration

public org.apache.flink.configuration.Configuration getFlinkConfiguration()

setQueue

public void setQueue(String queue)

setLocalJarPath

public void setLocalJarPath(org.apache.hadoop.fs.Path localJarPath)

addShipFiles

public void addShipFiles(List<File> shipFiles)

Adds the given files to the list of files to ship.

Note that any file matching "flink-dist*.jar" will be excluded from the upload by uploadAndRegisterFiles(Collection, FileSystem, Path, ApplicationId, List, Map, StringBuilder) since we upload the Flink uber jar ourselves and do not need to deploy it multiple times.

Parameters:

shipFiles - files to ship

setDynamicPropertiesEncoded

public void setDynamicPropertiesEncoded(String dynamicPropertiesEncoded)

hasUserJarFiles

public boolean hasUserJarFiles(List<URL> requiredJarFiles)

Returns true if the descriptor has the job jars to include in the classpath.

setProvidedUserJarFiles

public void setProvidedUserJarFiles(List<URL> userJarFiles)

Sets the user jar which is included in the system classloader of all nodes.

getDynamicPropertiesEncoded

public String getDynamicPropertiesEncoded()

setDetachedMode

@Deprecated
public void setDetachedMode(boolean detachedMode)

Deprecated. The cluster descriptor should not know about this option.

isDetachedMode

@Deprecated
public boolean isDetachedMode()

Deprecated. The cluster descriptor should not know about this option.

getZookeeperNamespace

public String getZookeeperNamespace()

setZookeeperNamespace

public void setZookeeperNamespace(String zookeeperNamespace)

getNodeLabel

public String getNodeLabel()

setNodeLabel

public void setNodeLabel(String nodeLabel)

close

public void close()

Specified by:

close in interface AutoCloseable

retrieve

public org.apache.flink.client.program.ClusterClient<org.apache.hadoop.yarn.api.records.ApplicationId> retrieve(org.apache.hadoop.yarn.api.records.ApplicationId applicationId)
                                                                                                         throws org.apache.flink.client.deployment.ClusterRetrieveException

Specified by:

retrieve in interface org.apache.flink.client.deployment.ClusterDescriptor<org.apache.hadoop.yarn.api.records.ApplicationId>

Throws:

org.apache.flink.client.deployment.ClusterRetrieveException

deploySessionCluster

public org.apache.flink.client.program.ClusterClient<org.apache.hadoop.yarn.api.records.ApplicationId> deploySessionCluster(org.apache.flink.client.deployment.ClusterSpecification clusterSpecification)
                                                                                                                     throws org.apache.flink.client.deployment.ClusterDeploymentException

Specified by:

deploySessionCluster in interface org.apache.flink.client.deployment.ClusterDescriptor<org.apache.hadoop.yarn.api.records.ApplicationId>

Throws:

org.apache.flink.client.deployment.ClusterDeploymentException

killCluster

public void killCluster(org.apache.hadoop.yarn.api.records.ApplicationId applicationId)
                 throws org.apache.flink.util.FlinkException

Specified by:

killCluster in interface org.apache.flink.client.deployment.ClusterDescriptor<org.apache.hadoop.yarn.api.records.ApplicationId>

Throws:

org.apache.flink.util.FlinkException

deployInternal

protected org.apache.flink.client.program.ClusterClient<org.apache.hadoop.yarn.api.records.ApplicationId> deployInternal(org.apache.flink.client.deployment.ClusterSpecification clusterSpecification,
                                                                                                                         String applicationName,
                                                                                                                         String yarnClusterEntrypoint,
                                                                                                                         @Nullable
                                                                                                                         org.apache.flink.runtime.jobgraph.JobGraph jobGraph,
                                                                                                                         boolean detached)
                                                                                                                  throws Exception

This method will block until the ApplicationMaster/JobManager have been deployed on YARN.

Parameters:

clusterSpecification - Initial cluster specification for the Flink cluster to be deployed

applicationName - name of the Yarn application to start

yarnClusterEntrypoint - Class name of the Yarn cluster entry point.

jobGraph - A job graph which is deployed with the Flink cluster, null if none

detached - True if the cluster should be started in detached mode

Throws:

Exception

validateClusterResources

protected org.apache.flink.client.deployment.ClusterSpecification validateClusterResources(org.apache.flink.client.deployment.ClusterSpecification clusterSpecification,
                                                                                           int yarnMinAllocationMB,
                                                                                           org.apache.hadoop.yarn.api.records.Resource maximumResourceCapability,
                                                                                           org.apache.flink.yarn.AbstractYarnClusterDescriptor.ClusterResourceDescription freeClusterResources)
                                                                                    throws org.apache.flink.yarn.AbstractYarnClusterDescriptor.YarnDeploymentException

Throws:

org.apache.flink.yarn.AbstractYarnClusterDescriptor.YarnDeploymentException

startAppMaster

public org.apache.hadoop.yarn.api.records.ApplicationReport startAppMaster(org.apache.flink.configuration.Configuration configuration,
                                                                           String applicationName,
                                                                           String yarnClusterEntrypoint,
                                                                           org.apache.flink.runtime.jobgraph.JobGraph jobGraph,
                                                                           org.apache.hadoop.yarn.client.api.YarnClient yarnClient,
                                                                           org.apache.hadoop.yarn.client.api.YarnClientApplication yarnApplication,
                                                                           org.apache.flink.client.deployment.ClusterSpecification clusterSpecification)
                                                                    throws Exception

Throws:

Exception

getClusterDescription

public String getClusterDescription()

Specified by:

getClusterDescription in interface org.apache.flink.client.deployment.ClusterDescriptor<org.apache.hadoop.yarn.api.records.ApplicationId>

setName

public void setName(String name)

addLibFolderToShipFiles

protected void addLibFolderToShipFiles(Collection<File> effectiveShipFiles)

setupApplicationMasterContainer

protected org.apache.hadoop.yarn.api.records.ContainerLaunchContext setupApplicationMasterContainer(String yarnClusterEntrypoint,
                                                                                                    boolean hasLogback,
                                                                                                    boolean hasLog4j,
                                                                                                    boolean hasKrb5,
                                                                                                    int jobManagerMemoryMb)

createYarnClusterClient

protected abstract org.apache.flink.client.program.ClusterClient<org.apache.hadoop.yarn.api.records.ApplicationId> createYarnClusterClient(AbstractYarnClusterDescriptor descriptor,
                                                                                                                                           int numberTaskManagers,
                                                                                                                                           int slotsPerTaskManager,
                                                                                                                                           org.apache.hadoop.yarn.api.records.ApplicationReport report,
                                                                                                                                           org.apache.flink.configuration.Configuration flinkConfiguration,
                                                                                                                                           boolean perJobCluster)
                                                                                                                                    throws Exception

Creates a YarnClusterClient; may be overriden in tests.

Throws:

Exception