You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@gobblin.apache.org by wl...@apache.org on 2024/10/21 15:17:04 UTC
(gobblin) branch master updated: [GOBBLIN-2162] Only load added jars in the cache (#4062)
This is an automated email from the ASF dual-hosted git repository.
wlo pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/gobblin.git
The following commit(s) were added to refs/heads/master by this push:
new 4f604235b0 [GOBBLIN-2162] Only load added jars in the cache (#4062)
4f604235b0 is described below
commit 4f604235b037eeb801b78950c897adeef2449d67
Author: William Lo <lo...@gmail.com>
AuthorDate: Mon Oct 21 11:16:58 2024 -0400
[GOBBLIN-2162] Only load added jars in the cache (#4062)
* Only load added jars in the cache
---
.../apache/gobblin/temporal/yarn/YarnService.java | 6 ++++
.../gobblin/yarn/GobblinYarnAppLauncher.java | 4 +++
.../gobblin/yarn/GobblinYarnConfigurationKeys.java | 2 ++
.../org/apache/gobblin/yarn/YarnHelixUtils.java | 13 +++++++++
.../apache/gobblin/yarn/YarnHelixUtilsTest.java | 34 ++++++++++++++++++++++
5 files changed, 59 insertions(+)
diff --git a/gobblin-temporal/src/main/java/org/apache/gobblin/temporal/yarn/YarnService.java b/gobblin-temporal/src/main/java/org/apache/gobblin/temporal/yarn/YarnService.java
index 426521b921..47d76b963e 100644
--- a/gobblin-temporal/src/main/java/org/apache/gobblin/temporal/yarn/YarnService.java
+++ b/gobblin-temporal/src/main/java/org/apache/gobblin/temporal/yarn/YarnService.java
@@ -546,7 +546,13 @@ class YarnService extends AbstractIdleService {
FileStatus[] statuses = this.fs.listStatus(destDir);
if (statuses != null) {
+ Set<String> appLibJars = YarnHelixUtils.getAppLibJarList(this.config);
for (FileStatus status : statuses) {
+ String fileName = status.getPath().getName();
+ // Ensure that we are only adding jars that were uploaded by the YarnAppLauncher for this application
+ if (fileName.contains(".jar") && !appLibJars.contains(fileName)) {
+ continue;
+ }
YarnHelixUtils.addFileAsLocalResource(this.fs, status.getPath(), LocalResourceType.FILE, resourceMap);
}
}
diff --git a/gobblin-yarn/src/main/java/org/apache/gobblin/yarn/GobblinYarnAppLauncher.java b/gobblin-yarn/src/main/java/org/apache/gobblin/yarn/GobblinYarnAppLauncher.java
index 4ce40f21df..cf88678425 100644
--- a/gobblin-yarn/src/main/java/org/apache/gobblin/yarn/GobblinYarnAppLauncher.java
+++ b/gobblin-yarn/src/main/java/org/apache/gobblin/yarn/GobblinYarnAppLauncher.java
@@ -243,6 +243,8 @@ public class GobblinYarnAppLauncher {
private final boolean jarCacheEnabled;
+ private final Set<String> libJarNames = new HashSet<>(); // List of jars that are shared between appMaster and containers
+
public GobblinYarnAppLauncher(Config config, YarnConfiguration yarnConfiguration) throws IOException {
this.config = config.withValue(GobblinYarnConfigurationKeys.YARN_APPLICATION_LAUNCHER_START_TIME_KEY,
ConfigValueFactory.fromAnyRef(System.currentTimeMillis()));
@@ -668,6 +670,7 @@ public class GobblinYarnAppLauncher {
Path unsharedJarsDestDir = new Path(appWorkDir, GobblinYarnConfigurationKeys.LIB_JARS_DIR_NAME);
addLibJars(new Path(this.config.getString(GobblinYarnConfigurationKeys.LIB_JARS_DIR_KEY)),
Optional.of(appMasterResources), libJarsDestDir, unsharedJarsDestDir, localFs);
+ this.libJarNames.addAll(appMasterResources.keySet());
LOGGER.info("Added lib jars to directory: {} and execution-private directory: {}", libJarsDestDir, unsharedJarsDestDir);
}
if (this.config.hasPath(GobblinYarnConfigurationKeys.APP_MASTER_JARS_KEY)) {
@@ -814,6 +817,7 @@ public class GobblinYarnAppLauncher {
.append(" -D").append(GobblinYarnConfigurationKeys.GOBBLIN_YARN_CONTAINER_LOG_DIR_NAME).append("=").append(ApplicationConstants.LOG_DIR_EXPANSION_VAR)
.append(" -D").append(GobblinYarnConfigurationKeys.GOBBLIN_YARN_CONTAINER_LOG_FILE_NAME).append("=").append(logFileName).append(".").append(ApplicationConstants.STDOUT)
.append(" -D").append(GobblinYarnConfigurationKeys.YARN_APPLICATION_LAUNCHER_START_TIME_KEY).append("=").append(config.getString(GobblinYarnConfigurationKeys.YARN_APPLICATION_LAUNCHER_START_TIME_KEY))
+ .append(" -D").append(GobblinYarnConfigurationKeys.YARN_APPLICATION_LIB_JAR_LIST).append("=").append(String.join(",", this.libJarNames))
.append(" ").append(JvmUtils.formatJvmArguments(this.appMasterJvmArgs))
.append(" ").append(appMasterClass.getName())
.append(" --").append(GobblinClusterConfigurationKeys.APPLICATION_NAME_OPTION_NAME)
diff --git a/gobblin-yarn/src/main/java/org/apache/gobblin/yarn/GobblinYarnConfigurationKeys.java b/gobblin-yarn/src/main/java/org/apache/gobblin/yarn/GobblinYarnConfigurationKeys.java
index 93cc23dba6..b8fc95b29d 100644
--- a/gobblin-yarn/src/main/java/org/apache/gobblin/yarn/GobblinYarnConfigurationKeys.java
+++ b/gobblin-yarn/src/main/java/org/apache/gobblin/yarn/GobblinYarnConfigurationKeys.java
@@ -55,6 +55,8 @@ public class GobblinYarnConfigurationKeys {
public static final String JAR_CACHE_DIR = GOBBLIN_YARN_PREFIX + "jar.cache.dir";
+ public static final String YARN_APPLICATION_LIB_JAR_LIST = GOBBLIN_YARN_PREFIX + "lib.jar.list";
+
// Used to store the start time of the app launcher to propagate to workers and appmaster
public static final String YARN_APPLICATION_LAUNCHER_START_TIME_KEY = GOBBLIN_YARN_PREFIX + "application.start.time";
diff --git a/gobblin-yarn/src/main/java/org/apache/gobblin/yarn/YarnHelixUtils.java b/gobblin-yarn/src/main/java/org/apache/gobblin/yarn/YarnHelixUtils.java
index 4a74da44a5..d38ebe52ee 100644
--- a/gobblin-yarn/src/main/java/org/apache/gobblin/yarn/YarnHelixUtils.java
+++ b/gobblin-yarn/src/main/java/org/apache/gobblin/yarn/YarnHelixUtils.java
@@ -23,9 +23,11 @@ import java.net.URL;
import java.text.SimpleDateFormat;
import java.util.Arrays;
import java.util.Collection;
+import java.util.HashSet;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
+import java.util.Set;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.stream.Collectors;
@@ -52,6 +54,7 @@ import org.slf4j.LoggerFactory;
import com.google.common.base.Splitter;
import com.google.common.collect.Maps;
+import com.google.common.collect.Sets;
import com.typesafe.config.Config;
import org.apache.gobblin.util.ConfigUtils;
@@ -237,6 +240,16 @@ public class YarnHelixUtils {
return deletesSuccessful;
}
+
+ public static Set<String> getAppLibJarList(Config config) {
+ Set<String> libAppJars = new HashSet<>(Arrays.asList(
+ ConfigUtils.getString(config, GobblinYarnConfigurationKeys.YARN_APPLICATION_LIB_JAR_LIST, "").split(",")));
+ Set<String> containerJars = new HashSet<>(Arrays.asList(
+ ConfigUtils.getString(config, GobblinYarnConfigurationKeys.CONTAINER_JARS_KEY, "").split(",")));
+ libAppJars.addAll(containerJars);
+ return Sets.filter(libAppJars, s -> !s.isEmpty());
+ }
+
public static void addRemoteFilesToLocalResources(String hdfsFileList, Map<String, LocalResource> resourceMap, Configuration yarnConfiguration) throws IOException {
for (String hdfsFilePath : SPLITTER.split(hdfsFileList)) {
Path srcFilePath = new Path(hdfsFilePath);
diff --git a/gobblin-yarn/src/test/java/org/apache/gobblin/yarn/YarnHelixUtilsTest.java b/gobblin-yarn/src/test/java/org/apache/gobblin/yarn/YarnHelixUtilsTest.java
index c271258c1d..1030e7b024 100644
--- a/gobblin-yarn/src/test/java/org/apache/gobblin/yarn/YarnHelixUtilsTest.java
+++ b/gobblin-yarn/src/test/java/org/apache/gobblin/yarn/YarnHelixUtilsTest.java
@@ -17,6 +17,7 @@
package org.apache.gobblin.yarn;
import java.io.IOException;
+import java.util.Set;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
@@ -97,6 +98,39 @@ public class YarnHelixUtilsTest {
// Should be cleaned up
Assert.assertFalse(fs.exists(new Path(this.tempDir, "tmp/2024-07")));
Assert.assertFalse(fs.exists(new Path(this.tempDir, "tmp/2024-06")));
+ }
+
+ @Test
+ public void testGetJarListFromConfigs() {
+ // Test when container jars is empty
+ Config emptyContainerJarsList = ConfigFactory.empty()
+ .withValue(GobblinYarnConfigurationKeys.YARN_APPLICATION_LIB_JAR_LIST, ConfigValueFactory.fromAnyRef("a.jar,b.jar"))
+ .withValue(GobblinYarnConfigurationKeys.CONTAINER_JARS_KEY, ConfigValueFactory.fromAnyRef(""));
+
+ Set<String> jars = YarnHelixUtils.getAppLibJarList(emptyContainerJarsList);
+ Assert.assertEquals(2, jars.size());
+ Assert.assertTrue(jars.contains("a.jar"));
+ Assert.assertTrue(jars.contains("b.jar"));
+ // Test when yarn application lib jars is empty
+ Config emptyYarnAppLibJarsConfig = ConfigFactory.empty()
+ .withValue(GobblinYarnConfigurationKeys.YARN_APPLICATION_LIB_JAR_LIST, ConfigValueFactory.fromAnyRef(""))
+ .withValue(GobblinYarnConfigurationKeys.CONTAINER_JARS_KEY, ConfigValueFactory.fromAnyRef("c.jar,d.jar"));
+
+ jars = YarnHelixUtils.getAppLibJarList(emptyYarnAppLibJarsConfig);
+ Assert.assertEquals(2, jars.size());
+ Assert.assertTrue(jars.contains("c.jar"));
+ Assert.assertTrue(jars.contains("d.jar"));
+
+ // Test when both yarn application lib jars and container jars are not empty
+ Config config = ConfigFactory.empty()
+ .withValue(GobblinYarnConfigurationKeys.YARN_APPLICATION_LIB_JAR_LIST, ConfigValueFactory.fromAnyRef("a.jar,b.jar"))
+ .withValue(GobblinYarnConfigurationKeys.CONTAINER_JARS_KEY, ConfigValueFactory.fromAnyRef("c.jar,d.jar"));
+ jars = YarnHelixUtils.getAppLibJarList(config);
+ Assert.assertEquals(4, jars.size());
+ Assert.assertTrue(jars.contains("a.jar"));
+ Assert.assertTrue(jars.contains("b.jar"));
+ Assert.assertTrue(jars.contains("c.jar"));
+ Assert.assertTrue(jars.contains("d.jar"));
}
}
\ No newline at end of file