[incubator-gobblin] branch master updated: [GOBBLIN-739] Add a way to propagate the Azkaban job config to Gobblin on YARN

[hu...@apache.org]

This is an automated email from the ASF dual-hosted git repository.

hutran pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/incubator-gobblin.git


The following commit(s) were added to refs/heads/master by this push:
     new e40852f  [GOBBLIN-739] Add a way to propagate the Azkaban job config to Gobblin on YARN
e40852f is described below

commit e40852f2ff97466fd60ade1ec68ea60321ce8c6c
Author: Hung Tran <hu...@linkedin.com>
AuthorDate: Tue Apr 16 09:18:56 2019 -0700

    [GOBBLIN-739] Add a way to propagate the Azkaban job config to Gobblin on YARN
    
    Closes #2606 from htran1/azkaban_config_to_yarn
---
 .../azkaban/AzkabanGobblinYarnAppLauncher.java     | 42 +++++++++++++++
 .../azkaban/AzkabanGobblinYarnAppLauncherTest.java | 61 ++++++++++++++++++++++
 2 files changed, 103 insertions(+)

diff --git a/gobblin-modules/gobblin-azkaban/src/main/java/org/apache/gobblin/azkaban/AzkabanGobblinYarnAppLauncher.java b/gobblin-modules/gobblin-azkaban/src/main/java/org/apache/gobblin/azkaban/AzkabanGobblinYarnAppLauncher.java
index 52c193b..6fa7eab 100644
--- a/gobblin-modules/gobblin-azkaban/src/main/java/org/apache/gobblin/azkaban/AzkabanGobblinYarnAppLauncher.java
+++ b/gobblin-modules/gobblin-azkaban/src/main/java/org/apache/gobblin/azkaban/AzkabanGobblinYarnAppLauncher.java
@@ -17,14 +17,19 @@
 
 package org.apache.gobblin.azkaban;
 
+import java.io.File;
 import java.io.IOException;
 import java.util.Properties;
 import java.util.concurrent.TimeoutException;
 
+import org.apache.commons.io.FileUtils;
 import org.apache.hadoop.yarn.conf.YarnConfiguration;
 import org.apache.log4j.Logger;
 
+import com.google.common.annotations.VisibleForTesting;
+import com.google.common.base.Charsets;
 import com.typesafe.config.Config;
+import com.typesafe.config.ConfigRenderOptions;
 
 import org.apache.gobblin.util.ConfigUtils;
 import org.apache.gobblin.yarn.GobblinYarnAppLauncher;
@@ -48,6 +53,8 @@ import azkaban.jobExecutor.AbstractJob;
  * @author Yinan Li
  */
 public class AzkabanGobblinYarnAppLauncher extends AbstractJob {
+  // if this is set then the Azkaban config will be written to the specified file path
+  public static final String AZKABAN_CONFIG_OUTPUT_PATH = "gobblin.yarn.akabanConfigOutputPath";
 
   private static final Logger LOGGER = Logger.getLogger(AzkabanJobLauncher.class);
 
@@ -56,6 +63,9 @@ public class AzkabanGobblinYarnAppLauncher extends AbstractJob {
   public AzkabanGobblinYarnAppLauncher(String jobId, Properties props) throws IOException {
     super(jobId, LOGGER);
     Config gobblinConfig = ConfigUtils.propertiesToConfig(props);
+
+    outputConfigToFile(gobblinConfig);
+
     this.gobblinYarnAppLauncher = new GobblinYarnAppLauncher(gobblinConfig, new YarnConfiguration());
   }
 
@@ -87,4 +97,36 @@ public class AzkabanGobblinYarnAppLauncher extends AbstractJob {
       super.cancel();
     }
   }
+
+  /**
+   * Write the config to the file specified with the config key {@value AZKABAN_CONFIG_OUTPUT_PATH} if it
+   * is configured.
+   * @param config the config to output
+   * @throws IOException
+   */
+  @VisibleForTesting
+  static void outputConfigToFile(Config config) throws IOException {
+    // If a file path is specified then write the Azkaban config to that path in HOCON format.
+    // This can be used to generate an application.conf file to pass to the yarn app master and containers.
+    if (config.hasPath(AZKABAN_CONFIG_OUTPUT_PATH)) {
+      File configFile = new File(config.getString(AZKABAN_CONFIG_OUTPUT_PATH));
+      File parentDir = configFile.getParentFile();
+
+      if (parentDir != null && !parentDir.exists()) {
+        if (!parentDir.mkdirs()) {
+          throw new IOException("Error creating directories for " + parentDir);
+        }
+      }
+
+      ConfigRenderOptions configRenderOptions = ConfigRenderOptions.defaults();
+      configRenderOptions = configRenderOptions.setComments(false);
+      configRenderOptions = configRenderOptions.setOriginComments(false);
+      configRenderOptions = configRenderOptions.setFormatted(true);
+      configRenderOptions = configRenderOptions.setJson(false);
+
+      String renderedConfig = config.root().render(configRenderOptions);
+
+      FileUtils.writeStringToFile(configFile, renderedConfig, Charsets.UTF_8);
+    }
+  }
 }
diff --git a/gobblin-modules/gobblin-azkaban/src/test/java/org/apache/gobblin/azkaban/AzkabanGobblinYarnAppLauncherTest.java b/gobblin-modules/gobblin-azkaban/src/test/java/org/apache/gobblin/azkaban/AzkabanGobblinYarnAppLauncherTest.java
new file mode 100644
index 0000000..ccb49c0
--- /dev/null
+++ b/gobblin-modules/gobblin-azkaban/src/test/java/org/apache/gobblin/azkaban/AzkabanGobblinYarnAppLauncherTest.java
@@ -0,0 +1,61 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.gobblin.azkaban;
+
+import java.io.File;
+import java.io.IOException;
+import java.nio.file.Path;
+import java.nio.file.Paths;
+
+import org.apache.commons.io.FileUtils;
+import org.testng.Assert;
+import org.testng.annotations.Test;
+
+import com.google.common.base.Charsets;
+import com.google.common.io.Files;
+import com.typesafe.config.Config;
+import com.typesafe.config.ConfigFactory;
+import com.typesafe.config.ConfigValueFactory;
+
+import lombok.extern.slf4j.Slf4j;
+
+import org.apache.gobblin.configuration.ConfigurationKeys;
+
+@Slf4j
+public class AzkabanGobblinYarnAppLauncherTest {
+
+  @Test
+  public void testOutputConfig() throws IOException {
+    File tmpTestDir = Files.createTempDir();
+
+    try {
+      Path outputPath = Paths.get(tmpTestDir.toString(), "application.conf");
+      Config config = ConfigFactory.empty()
+          .withValue(ConfigurationKeys.FS_URI_KEY, ConfigValueFactory.fromAnyRef("file:///"))
+          .withValue(AzkabanGobblinYarnAppLauncher.AZKABAN_CONFIG_OUTPUT_PATH,
+              ConfigValueFactory.fromAnyRef(outputPath.toString()));
+
+      AzkabanGobblinYarnAppLauncher.outputConfigToFile(config);
+
+      String configString = Files.toString(outputPath.toFile(), Charsets.UTF_8);
+      Assert.assertTrue(configString.contains("fs"));
+    } finally {
+      FileUtils.deleteDirectory(tmpTestDir);
+    }
+  }
+}