You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by wa...@apache.org on 2022/07/23 02:30:35 UTC
[flink] branch master updated: [FLINK-26047][yarn] Support remote usrlib in HDFS for YARN deployment
This is an automated email from the ASF dual-hosted git repository.
wangyang0918 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
The following commit(s) were added to refs/heads/master by this push:
new fc533b9d9c1 [FLINK-26047][yarn] Support remote usrlib in HDFS for YARN deployment
fc533b9d9c1 is described below
commit fc533b9d9c1252db124b4c1cb8365b3906b009cf
Author: Biao Geng <gb...@gmail.com>
AuthorDate: Fri Mar 11 21:22:52 2022 +0800
[FLINK-26047][yarn] Support remote usrlib in HDFS for YARN deployment
---
.../generated/yarn_config_configuration.html | 6 +++
.../src/main/java/org/apache/flink/yarn/Utils.java | 47 ++++++++++++++---
.../flink/yarn/YarnApplicationFileUploader.java | 6 +--
.../apache/flink/yarn/YarnClusterDescriptor.java | 18 +++++--
.../yarn/configuration/YarnConfigOptions.java | 15 ++++++
.../test/java/org/apache/flink/yarn/UtilsTest.java | 60 +++++++++++++++++++---
6 files changed, 130 insertions(+), 22 deletions(-)
diff --git a/docs/layouts/shortcodes/generated/yarn_config_configuration.html b/docs/layouts/shortcodes/generated/yarn_config_configuration.html
index f59cd63844b..af0156f7bf3 100644
--- a/docs/layouts/shortcodes/generated/yarn_config_configuration.html
+++ b/docs/layouts/shortcodes/generated/yarn_config_configuration.html
@@ -134,6 +134,12 @@
<td>List<String></td>
<td>A semicolon-separated list of provided lib directories. They should be pre-uploaded and world-readable. Flink will use them to exclude the local Flink jars(e.g. flink-dist, lib/, plugins/)uploading to accelerate the job submission process. Also YARN will cache them on the nodes so that they doesn't need to be downloaded every time for each application. An example could be hdfs://$namenode_address/path/of/flink/lib</td>
</tr>
+ <tr>
+ <td><h5>yarn.provided.usrlib.dir</h5></td>
+ <td style="word-wrap: break-word;">(none)</td>
+ <td>String</td>
+ <td>The provided usrlib directory in remote. It should be pre-uploaded and world-readable. Flink will use it to exclude the local usrlib directory(i.e. usrlib/ under the parent directory of FLINK_LIB_DIR). Unlike yarn.provided.lib.dirs, YARN will not cache it on the nodes as it is for each application. An example could be hdfs://$namenode_address/path/of/flink/usrlib</td>
+ </tr>
<tr>
<td><h5>yarn.security.kerberos.localized-keytab-path</h5></td>
<td style="word-wrap: break-word;">"krb5.keytab"</td>
diff --git a/flink-yarn/src/main/java/org/apache/flink/yarn/Utils.java b/flink-yarn/src/main/java/org/apache/flink/yarn/Utils.java
index 7550715d13e..5d0da8efb9a 100644
--- a/flink-yarn/src/main/java/org/apache/flink/yarn/Utils.java
+++ b/flink-yarn/src/main/java/org/apache/flink/yarn/Utils.java
@@ -19,12 +19,12 @@
package org.apache.flink.yarn;
import org.apache.flink.annotation.VisibleForTesting;
+import org.apache.flink.configuration.ConfigConstants;
import org.apache.flink.configuration.ConfigUtils;
import org.apache.flink.runtime.clusterframework.BootstrapTools;
import org.apache.flink.runtime.clusterframework.ContaineredTaskManagerParameters;
import org.apache.flink.runtime.security.token.DelegationTokenConverter;
import org.apache.flink.runtime.util.HadoopUtils;
-import org.apache.flink.util.FlinkException;
import org.apache.flink.util.StringUtils;
import org.apache.flink.util.function.FunctionWithException;
import org.apache.flink.yarn.configuration.YarnConfigOptions;
@@ -67,7 +67,9 @@ import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Objects;
+import java.util.Optional;
+import static org.apache.flink.util.Preconditions.checkArgument;
import static org.apache.flink.util.Preconditions.checkNotNull;
import static org.apache.flink.yarn.YarnConfigKeys.ENV_FLINK_CLASSPATH;
import static org.apache.flink.yarn.YarnConfigKeys.LOCAL_RESOURCE_DESCRIPTOR_SEPARATOR;
@@ -633,12 +635,12 @@ public final class Utils {
return Resource.newInstance(unitMemMB, unitVcore);
}
- public static List<Path> getQualifiedRemoteSharedPaths(
+ public static List<Path> getQualifiedRemoteProvidedLibDirs(
org.apache.flink.configuration.Configuration configuration,
YarnConfiguration yarnConfiguration)
- throws IOException, FlinkException {
+ throws IOException {
- return getRemoteSharedPaths(
+ return getRemoteSharedLibPaths(
configuration,
pathStr -> {
final Path path = new Path(pathStr);
@@ -646,10 +648,10 @@ public final class Utils {
});
}
- private static List<Path> getRemoteSharedPaths(
+ private static List<Path> getRemoteSharedLibPaths(
org.apache.flink.configuration.Configuration configuration,
FunctionWithException<String, Path, IOException> strToPathMapper)
- throws IOException, FlinkException {
+ throws IOException {
final List<Path> providedLibDirs =
ConfigUtils.decodeListFromConfig(
@@ -657,7 +659,7 @@ public final class Utils {
for (Path path : providedLibDirs) {
if (!Utils.isRemotePath(path.toString())) {
- throw new FlinkException(
+ throw new IllegalArgumentException(
"The \""
+ YarnConfigOptions.PROVIDED_LIB_DIRS.key()
+ "\" should only contain"
@@ -669,6 +671,37 @@ public final class Utils {
return providedLibDirs;
}
+ public static boolean isUsrLibDirectory(final FileSystem fileSystem, final Path path)
+ throws IOException {
+ final FileStatus fileStatus = fileSystem.getFileStatus(path);
+ // Use the Path obj from fileStatus to get rid of trailing slash
+ return fileStatus.isDirectory()
+ && ConfigConstants.DEFAULT_FLINK_USR_LIB_DIR.equals(fileStatus.getPath().getName());
+ }
+
+ public static Optional<Path> getQualifiedRemoteProvidedUsrLib(
+ org.apache.flink.configuration.Configuration configuration,
+ YarnConfiguration yarnConfiguration)
+ throws IOException, IllegalArgumentException {
+ String usrlib = configuration.getString(YarnConfigOptions.PROVIDED_USRLIB_DIR);
+ if (usrlib == null) {
+ return Optional.empty();
+ }
+ final Path qualifiedUsrLibPath =
+ FileSystem.get(yarnConfiguration).makeQualified(new Path(usrlib));
+ checkArgument(
+ isRemotePath(qualifiedUsrLibPath.toString()),
+ "The \"%s\" must point to a remote dir "
+ + "which is accessible from all worker nodes.",
+ YarnConfigOptions.PROVIDED_USRLIB_DIR.key());
+ checkArgument(
+ isUsrLibDirectory(FileSystem.get(yarnConfiguration), qualifiedUsrLibPath),
+ "The \"%s\" should be named with \"%s\".",
+ YarnConfigOptions.PROVIDED_USRLIB_DIR.key(),
+ ConfigConstants.DEFAULT_FLINK_USR_LIB_DIR);
+ return Optional.of(qualifiedUsrLibPath);
+ }
+
public static YarnConfiguration getYarnAndHadoopConfiguration(
org.apache.flink.configuration.Configuration flinkConfig) {
final YarnConfiguration yarnConfig = getYarnConfiguration(flinkConfig);
diff --git a/flink-yarn/src/main/java/org/apache/flink/yarn/YarnApplicationFileUploader.java b/flink-yarn/src/main/java/org/apache/flink/yarn/YarnApplicationFileUploader.java
index 0e385b1ba51..c03d15e3552 100644
--- a/flink-yarn/src/main/java/org/apache/flink/yarn/YarnApplicationFileUploader.java
+++ b/flink-yarn/src/main/java/org/apache/flink/yarn/YarnApplicationFileUploader.java
@@ -514,11 +514,7 @@ class YarnApplicationFileUploader implements AutoCloseable {
private boolean isUsrLibDirIncludedInProvidedLib(final List<Path> providedLibDirs)
throws IOException {
for (Path path : providedLibDirs) {
- final FileStatus fileStatus = fileSystem.getFileStatus(path);
- // Use the Path obj from fileStatus to get rid of trailing slash
- if (fileStatus.isDirectory()
- && ConfigConstants.DEFAULT_FLINK_USR_LIB_DIR.equals(
- fileStatus.getPath().getName())) {
+ if (Utils.isUsrLibDirectory(fileSystem, path)) {
return true;
}
}
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 04b35098e8b..7f599490aa9 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
@@ -811,7 +811,10 @@ public class YarnClusterDescriptor implements ClusterDescriptor<ApplicationId> {
ApplicationSubmissionContext appContext = yarnApplication.getApplicationSubmissionContext();
final List<Path> providedLibDirs =
- Utils.getQualifiedRemoteSharedPaths(configuration, yarnConfiguration);
+ Utils.getQualifiedRemoteProvidedLibDirs(configuration, yarnConfiguration);
+
+ final Optional<Path> providedUsrLibDir =
+ Utils.getQualifiedRemoteProvidedUsrLib(configuration, yarnConfiguration);
Path stagingDirPath = getStagingDir(fs);
FileSystem stagingDirFs = stagingDirPath.getFileSystem(yarnConfiguration);
@@ -945,8 +948,17 @@ public class YarnClusterDescriptor implements ClusterDescriptor<ApplicationId> {
: Path.CUR_DIR,
LocalResourceType.FILE);
- // usrlib will be automatically shipped if it exists.
- if (ClusterEntrypointUtils.tryFindUserLibDirectory().isPresent()) {
+ // usrlib in remote will be used first.
+ if (providedUsrLibDir.isPresent()) {
+ final List<String> usrLibClassPaths =
+ fileUploader.registerMultipleLocalResources(
+ Collections.singletonList(providedUsrLibDir.get()),
+ Path.CUR_DIR,
+ LocalResourceType.FILE);
+ userClassPaths.addAll(usrLibClassPaths);
+ } else if (ClusterEntrypointUtils.tryFindUserLibDirectory().isPresent()) {
+ // local usrlib will be automatically shipped if it exists and there is no remote
+ // usrlib.
final Set<File> usrLibShipFiles = new HashSet<>();
addUsrLibFolderToShipFiles(usrLibShipFiles);
final List<String> usrLibClassPaths =
diff --git a/flink-yarn/src/main/java/org/apache/flink/yarn/configuration/YarnConfigOptions.java b/flink-yarn/src/main/java/org/apache/flink/yarn/configuration/YarnConfigOptions.java
index 78a04149ef8..c7ec43bfd48 100644
--- a/flink-yarn/src/main/java/org/apache/flink/yarn/configuration/YarnConfigOptions.java
+++ b/flink-yarn/src/main/java/org/apache/flink/yarn/configuration/YarnConfigOptions.java
@@ -340,6 +340,21 @@ public class YarnConfigOptions {
+ "they doesn't need to be downloaded every time for each application. An example could be "
+ "hdfs://$namenode_address/path/of/flink/lib");
+ /**
+ * Allows users to directly utilize usrlib directory in HDFS for YARN application mode. The
+ * classloader for loading jars under the usrlib will be controlled by {@link
+ * YarnConfigOptions#CLASSPATH_INCLUDE_USER_JAR}.
+ */
+ public static final ConfigOption<String> PROVIDED_USRLIB_DIR =
+ key("yarn.provided.usrlib.dir")
+ .stringType()
+ .noDefaultValue()
+ .withDescription(
+ "The provided usrlib directory in remote. It should be pre-uploaded and "
+ + "world-readable. Flink will use it to exclude the local usrlib directory(i.e. usrlib/ under the parent directory of FLINK_LIB_DIR)."
+ + " Unlike yarn.provided.lib.dirs, YARN will not cache it on the nodes as it is for each application. An example could be "
+ + "hdfs://$namenode_address/path/of/flink/usrlib");
+
@SuppressWarnings("unused")
public static final ConfigOption<String> HADOOP_CONFIG_KEY =
key("flink.hadoop.<key>")
diff --git a/flink-yarn/src/test/java/org/apache/flink/yarn/UtilsTest.java b/flink-yarn/src/test/java/org/apache/flink/yarn/UtilsTest.java
index e7956aca991..fd8aadd7b98 100644
--- a/flink-yarn/src/test/java/org/apache/flink/yarn/UtilsTest.java
+++ b/flink-yarn/src/test/java/org/apache/flink/yarn/UtilsTest.java
@@ -18,16 +18,19 @@
package org.apache.flink.yarn;
+import org.apache.flink.configuration.ConfigConstants;
import org.apache.flink.configuration.Configuration;
-import org.apache.flink.util.FlinkException;
import org.apache.flink.yarn.configuration.YarnConfigOptions;
import org.apache.hadoop.fs.CommonConfigurationKeysPublic;
+import org.apache.hadoop.hdfs.MiniDFSCluster;
import org.apache.hadoop.yarn.api.records.Resource;
import org.apache.hadoop.yarn.conf.YarnConfiguration;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.io.TempDir;
+import java.io.File;
+import java.io.IOException;
import java.nio.file.Files;
import java.nio.file.Path;
import java.util.Collections;
@@ -98,7 +101,7 @@ class UtilsTest {
yarnConfig.set(CommonConfigurationKeysPublic.FS_DEFAULT_NAME_KEY, defaultFs);
final List<org.apache.hadoop.fs.Path> sharedLibs =
- Utils.getQualifiedRemoteSharedPaths(flinkConfig, yarnConfig);
+ Utils.getQualifiedRemoteProvidedLibDirs(flinkConfig, yarnConfig);
assertThat(sharedLibs).hasSize(1);
assertThat(sharedLibs.get(0).toUri()).hasToString(qualifiedPath);
}
@@ -115,14 +118,57 @@ class UtilsTest {
+ "\" should only "
+ "contain dirs accessible from all worker nodes";
assertThatThrownBy(
- () -> {
- Utils.getQualifiedRemoteSharedPaths(
- flinkConfig, new YarnConfiguration());
- })
- .isInstanceOf(FlinkException.class)
+ () ->
+ Utils.getQualifiedRemoteProvidedLibDirs(
+ flinkConfig, new YarnConfiguration()))
+ .isInstanceOf(IllegalArgumentException.class)
.hasMessageContaining(msg);
}
+ @Test
+ void testInvalidRemoteUsrLib(@TempDir Path tempDir) throws IOException {
+ final String sharedLibPath = "hdfs:///flink/badlib";
+
+ final org.apache.hadoop.conf.Configuration hdConf =
+ new org.apache.hadoop.conf.Configuration();
+ hdConf.set(MiniDFSCluster.HDFS_MINIDFS_BASEDIR, tempDir.toAbsolutePath().toString());
+ try (final MiniDFSCluster hdfsCluster = new MiniDFSCluster.Builder(hdConf).build()) {
+ final org.apache.hadoop.fs.Path hdfsRootPath =
+ new org.apache.hadoop.fs.Path(hdfsCluster.getURI());
+ hdfsCluster.getFileSystem().mkdirs(new org.apache.hadoop.fs.Path(sharedLibPath));
+
+ final Configuration flinkConfig = new Configuration();
+ flinkConfig.set(YarnConfigOptions.PROVIDED_USRLIB_DIR, sharedLibPath);
+ final YarnConfiguration yarnConfig = new YarnConfiguration();
+ yarnConfig.set(
+ CommonConfigurationKeysPublic.FS_DEFAULT_NAME_KEY, hdfsRootPath.toString());
+ assertThatThrownBy(
+ () -> Utils.getQualifiedRemoteProvidedUsrLib(flinkConfig, yarnConfig))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessage(
+ "The \"%s\" should be named with \"%s\".",
+ YarnConfigOptions.PROVIDED_USRLIB_DIR.key(),
+ ConfigConstants.DEFAULT_FLINK_USR_LIB_DIR);
+ }
+ }
+
+ @Test
+ void testSharedUsrLibIsNotRemotePathShouldThrowException(@TempDir Path tempDir) {
+ final File localLib = new File(tempDir.toAbsolutePath().toString(), "usrlib");
+ assertThat(localLib.mkdirs()).isTrue();
+ final Configuration flinkConfig = new Configuration();
+ flinkConfig.set(YarnConfigOptions.PROVIDED_USRLIB_DIR, localLib.getAbsolutePath());
+ assertThatThrownBy(
+ () ->
+ Utils.getQualifiedRemoteProvidedUsrLib(
+ flinkConfig, new YarnConfiguration()))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessage(
+ "The \"%s\" must point to a remote dir "
+ + "which is accessible from all worker nodes.",
+ YarnConfigOptions.PROVIDED_USRLIB_DIR.key());
+ }
+
@Test
void testGetYarnConfiguration() {
final String flinkPrefix = "flink.yarn.";