You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by di...@apache.org on 2022/03/29 01:46:16 UTC
[flink] branch release-1.15 updated: [FLINK-26814][python][yarn] Ensure PyFlink jobs works in YARN application mode
This is an automated email from the ASF dual-hosted git repository.
dianfu pushed a commit to branch release-1.15
in repository https://gitbox.apache.org/repos/asf/flink.git
The following commit(s) were added to refs/heads/release-1.15 by this push:
new 52cc77d [FLINK-26814][python][yarn] Ensure PyFlink jobs works in YARN application mode
52cc77d is described below
commit 52cc77df9f11b8ba21de80c445b1261959cf315a
Author: zhangjingcun <cu...@163.com>
AuthorDate: Wed Mar 23 12:24:27 2022 +0800
[FLINK-26814][python][yarn] Ensure PyFlink jobs works in YARN application mode
This closes #19208.
---
docs/content.zh/docs/deployment/cli.md | 16 +++++++++++
docs/content/docs/deployment/cli.md | 16 +++++++++++
.../flink/configuration/ConfigConstants.java | 5 ++++
.../apache/flink/yarn/YarnClusterDescriptor.java | 32 ++++++++++++++++++----
.../YarnApplicationClusterEntryPoint.java | 13 +++++++--
5 files changed, 75 insertions(+), 7 deletions(-)
diff --git a/docs/content.zh/docs/deployment/cli.md b/docs/content.zh/docs/deployment/cli.md
index 68ec237..c0af1a4 100644
--- a/docs/content.zh/docs/deployment/cli.md
+++ b/docs/content.zh/docs/deployment/cli.md
@@ -418,6 +418,22 @@ $ ./bin/flink run \
--python examples/python/table/word_count.py
```
+- Run a PyFlink job using a [YARN cluster in Application Mode]({{< ref "docs/deployment/resource-providers/yarn" >}}#application-mode):
+```bash
+$ ./bin/flink run-application -t yarn-application \
+ -Djobmanager.memory.process.size=1024m \
+ -Dtaskmanager.memory.process.size=1024m \
+ -Dyarn.application.name=<ApplicationName> \
+ -Dyarn.ship-files=/path/to/shipfiles \
+ -pyarch shipfiles/venv.zip \
+ -pyclientexec venv.zip/venv/bin/python3 \
+ -pyexec venv.zip/venv/bin/python3 \
+ -py shipfiles/word_count.py
+```
+<span class="label label-info">Note</span> It assumes that the Python dependencies needed to execute the job are already placed in the directory `/path/to/shipfiles`. For example, it should contain venv.zip and word_count.py for the above example.
+
+<span class="label label-info">Note</span> As it executes the job on the JobManager in YARN application mode, the paths specified in `-pyarch` and `-py` are paths relative to `shipfiles` which is the directory name of the shipped files.
+
- Run a PyFlink application on a native Kubernetes cluster having the cluster ID `<ClusterId>`, it requires a docker image with PyFlink installed, please refer to [Enabling PyFlink in docker]({{< ref "docs/deployment/resource-providers/standalone/docker" >}}#enabling-python):
```bash
$ ./bin/flink run-application \
diff --git a/docs/content/docs/deployment/cli.md b/docs/content/docs/deployment/cli.md
index 983a941..7a8e0ad 100644
--- a/docs/content/docs/deployment/cli.md
+++ b/docs/content/docs/deployment/cli.md
@@ -416,6 +416,22 @@ $ ./bin/flink run \
--python examples/python/table/word_count.py
```
+- Run a PyFlink job using a [YARN cluster in Application Mode]({{< ref "docs/deployment/resource-providers/yarn" >}}#application-mode):
+```bash
+$ ./bin/flink run-application -t yarn-application \
+ -Djobmanager.memory.process.size=1024m \
+ -Dtaskmanager.memory.process.size=1024m \
+ -Dyarn.application.name=<ApplicationName> \
+ -Dyarn.ship-files=/path/to/shipfiles \
+ -pyarch shipfiles/venv.zip \
+ -pyclientexec venv.zip/venv/bin/python3 \
+ -pyexec venv.zip/venv/bin/python3 \
+ -py shipfiles/word_count.py
+```
+<span class="label label-info">Note</span> It assumes that the Python dependencies needed to execute the job are already placed in the directory `/path/to/shipfiles`. For example, it should contain venv.zip and word_count.py for the above example.
+
+<span class="label label-info">Note</span> As it executes the job on the JobManager in YARN application mode, the paths specified in `-pyarch` and `-py` are paths relative to `shipfiles` which is the directory name of the shipped files.
+
- Run a PyFlink application on a native Kubernetes cluster having the cluster ID `<ClusterId>`, it requires a docker image with PyFlink installed, please refer to [Enabling PyFlink in docker]({{< ref "docs/deployment/resource-providers/standalone/docker" >}}#enabling-python):
```bash
$ ./bin/flink run-application \
diff --git a/flink-core/src/main/java/org/apache/flink/configuration/ConfigConstants.java b/flink-core/src/main/java/org/apache/flink/configuration/ConfigConstants.java
index c14399e..e7c441c 100644
--- a/flink-core/src/main/java/org/apache/flink/configuration/ConfigConstants.java
+++ b/flink-core/src/main/java/org/apache/flink/configuration/ConfigConstants.java
@@ -1732,6 +1732,11 @@ public final class ConfigConstants {
/** The environment variable name which contains the location of the opt directory. */
public static final String ENV_FLINK_OPT_DIR = "FLINK_OPT_DIR";
+ /**
+ * The default Flink opt directory if none has been specified via {@link #ENV_FLINK_OPT_DIR}.
+ */
+ public static final String DEFAULT_FLINK_OPT_DIR = "opt";
+
/** The environment variable name which contains the location of the plugins folder. */
public static final String ENV_FLINK_PLUGINS_DIR = "FLINK_PLUGINS_DIR";
diff --git a/flink-yarn/src/main/java/org/apache/flink/yarn/YarnClusterDescriptor.java b/flink-yarn/src/main/java/org/apache/flink/yarn/YarnClusterDescriptor.java
index c99fd9c..e90bf65 100644
--- a/flink-yarn/src/main/java/org/apache/flink/yarn/YarnClusterDescriptor.java
+++ b/flink-yarn/src/main/java/org/apache/flink/yarn/YarnClusterDescriptor.java
@@ -27,6 +27,7 @@ import org.apache.flink.client.deployment.ClusterRetrieveException;
import org.apache.flink.client.deployment.ClusterSpecification;
import org.apache.flink.client.deployment.application.ApplicationConfiguration;
import org.apache.flink.client.program.ClusterClientProvider;
+import org.apache.flink.client.program.PackagedProgramUtils;
import org.apache.flink.client.program.rest.RestClusterClient;
import org.apache.flink.configuration.ConfigConstants;
import org.apache.flink.configuration.ConfigOption;
@@ -118,8 +119,10 @@ import java.util.Optional;
import java.util.Set;
import java.util.stream.Collectors;
+import static org.apache.flink.client.deployment.application.ApplicationConfiguration.APPLICATION_MAIN_CLASS;
import static org.apache.flink.configuration.ConfigConstants.DEFAULT_FLINK_USR_LIB_DIR;
import static org.apache.flink.configuration.ConfigConstants.ENV_FLINK_LIB_DIR;
+import static org.apache.flink.configuration.ConfigConstants.ENV_FLINK_OPT_DIR;
import static org.apache.flink.runtime.entrypoint.component.FileJobGraphRetriever.JOB_GRAPH_FILE_PATH;
import static org.apache.flink.util.Preconditions.checkArgument;
import static org.apache.flink.util.Preconditions.checkNotNull;
@@ -447,11 +450,15 @@ public class YarnClusterDescriptor implements ClusterDescriptor<ApplicationId> {
applicationConfiguration.applyToConfiguration(flinkConfiguration);
- final List<String> pipelineJars =
- flinkConfiguration
- .getOptional(PipelineOptions.JARS)
- .orElse(Collections.emptyList());
- Preconditions.checkArgument(pipelineJars.size() == 1, "Should only have one jar");
+ // No need to do pipelineJars validation if it is a PyFlink job.
+ if (!(PackagedProgramUtils.isPython(applicationConfiguration.getApplicationClassName())
+ || PackagedProgramUtils.isPython(applicationConfiguration.getProgramArguments()))) {
+ final List<String> pipelineJars =
+ flinkConfiguration
+ .getOptional(PipelineOptions.JARS)
+ .orElse(Collections.emptyList());
+ Preconditions.checkArgument(pipelineJars.size() == 1, "Should only have one jar");
+ }
try {
return deployInternal(
@@ -910,6 +917,17 @@ public class YarnClusterDescriptor implements ClusterDescriptor<ApplicationId> {
LocalResourceType.ARCHIVE);
}
+ // only for application mode
+ // Python jar file only needs to be shipped and should not be added to classpath.
+ if (YarnApplicationClusterEntryPoint.class.getName().equals(yarnClusterEntrypoint)
+ && PackagedProgramUtils.isPython(configuration.get(APPLICATION_MAIN_CLASS))) {
+ fileUploader.registerMultipleLocalResources(
+ Collections.singletonList(
+ new Path(PackagedProgramUtils.getPythonJar().toURI())),
+ ConfigConstants.DEFAULT_FLINK_OPT_DIR,
+ LocalResourceType.FILE);
+ }
+
// Upload and register user jars
final List<String> userClassPaths =
fileUploader.registerMultipleLocalResources(
@@ -1134,6 +1152,10 @@ public class YarnClusterDescriptor implements ClusterDescriptor<ApplicationId> {
// set Flink app class path
appMasterEnv.put(YarnConfigKeys.ENV_FLINK_CLASSPATH, classPathBuilder.toString());
+ // Set FLINK_OPT_DIR to `opt` folder under working dir in container
+ appMasterEnv.put(
+ ENV_FLINK_OPT_DIR, Path.CUR_DIR + "/" + ConfigConstants.DEFAULT_FLINK_OPT_DIR);
+
// set Flink on YARN internal configuration values
appMasterEnv.put(YarnConfigKeys.FLINK_DIST_JAR, localResourceDescFlinkJar.toString());
appMasterEnv.put(YarnConfigKeys.ENV_APP_ID, appId.toString());
diff --git a/flink-yarn/src/main/java/org/apache/flink/yarn/entrypoint/YarnApplicationClusterEntryPoint.java b/flink-yarn/src/main/java/org/apache/flink/yarn/entrypoint/YarnApplicationClusterEntryPoint.java
index d38ebae..d619d37 100644
--- a/flink-yarn/src/main/java/org/apache/flink/yarn/entrypoint/YarnApplicationClusterEntryPoint.java
+++ b/flink-yarn/src/main/java/org/apache/flink/yarn/entrypoint/YarnApplicationClusterEntryPoint.java
@@ -24,6 +24,7 @@ import org.apache.flink.client.deployment.application.ApplicationConfiguration;
import org.apache.flink.client.program.DefaultPackagedProgramRetriever;
import org.apache.flink.client.program.PackagedProgram;
import org.apache.flink.client.program.PackagedProgramRetriever;
+import org.apache.flink.client.program.PackagedProgramUtils;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.configuration.PipelineOptions;
import org.apache.flink.runtime.entrypoint.ClusterEntrypoint;
@@ -132,9 +133,17 @@ public final class YarnApplicationClusterEntryPoint extends ApplicationClusterEn
throws FlinkException {
final File userLibDir = YarnEntrypointUtils.getUsrLibDir(configuration).orElse(null);
- final File userApplicationJar = getUserApplicationJar(userLibDir, configuration);
+
+ // No need to do pipelineJars validation if it is a PyFlink job.
+ if (!(PackagedProgramUtils.isPython(jobClassName)
+ || PackagedProgramUtils.isPython(programArguments))) {
+ final File userApplicationJar = getUserApplicationJar(userLibDir, configuration);
+ return DefaultPackagedProgramRetriever.create(
+ userLibDir, userApplicationJar, jobClassName, programArguments, configuration);
+ }
+
return DefaultPackagedProgramRetriever.create(
- userLibDir, userApplicationJar, jobClassName, programArguments, configuration);
+ userLibDir, jobClassName, programArguments, configuration);
}
private static File getUserApplicationJar(