[beam] branch master updated: Add `requirementsFile` parameter to LaunchConfig. (#28732)

[yhu]

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

yhu pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/beam.git


The following commit(s) were added to refs/heads/master by this push:
     new 5fbf65afd97 Add `requirementsFile` parameter to LaunchConfig. (#28732)
5fbf65afd97 is described below

commit 5fbf65afd97f23568fe3c78c20197ad3b0354ece
Author: Pranav Bhandari <bhandari.prana...@gmail.com>
AuthorDate: Wed Oct 18 11:07:51 2023 -0400

    Add `requirementsFile` parameter to LaunchConfig. (#28732)
    
    * Add requirementsFile parameter to LaunchConfig.
    
    * Install requirements in virtualenv for python jobs.
---
 .../apache/beam/it/common/PipelineLauncher.java    | 16 ++++++++++++++++
 .../it/gcp/dataflow/DefaultPipelineLauncher.java   | 22 ++++++++++++++++++----
 2 files changed, 34 insertions(+), 4 deletions(-)

diff --git 
a/it/common/src/main/java/org/apache/beam/it/common/PipelineLauncher.java 
b/it/common/src/main/java/org/apache/beam/it/common/PipelineLauncher.java
index 8777bbec6c4..6d1aeae21dd 100644
--- a/it/common/src/main/java/org/apache/beam/it/common/PipelineLauncher.java
+++ b/it/common/src/main/java/org/apache/beam/it/common/PipelineLauncher.java
@@ -121,6 +121,7 @@ public interface PipelineLauncher {
     private final @Nullable String specPath;
     private final @Nullable Sdk sdk;
     private final @Nullable String executable;
+    private final @Nullable String requirementsFile;
     private final @Nullable Pipeline pipeline;
 
     private LaunchConfig(Builder builder) {
@@ -130,6 +131,7 @@ public interface PipelineLauncher {
       this.specPath = builder.specPath;
       this.sdk = builder.sdk;
       this.executable = builder.executable;
+      this.requirementsFile = builder.requirementsFile;
       this.pipeline = builder.pipeline;
     }
 
@@ -161,6 +163,10 @@ public interface PipelineLauncher {
       return executable;
     }
 
+    public @Nullable String requirementsFile() {
+      return requirementsFile;
+    }
+
     public @Nullable Pipeline pipeline() {
       return pipeline;
     }
@@ -185,6 +191,7 @@ public interface PipelineLauncher {
       private Map<String, String> parameters;
       private Sdk sdk;
       private String executable;
+      private String requirementsFile;
       private Pipeline pipeline;
 
       private Builder(String jobName, String specPath) {
@@ -243,6 +250,15 @@ public interface PipelineLauncher {
         return this;
       }
 
+      public @Nullable String getRequirementsFile() {
+        return requirementsFile;
+      }
+
+      public Builder setRequirementsFile(String requirementsFile) {
+        this.requirementsFile = requirementsFile;
+        return this;
+      }
+
       public @Nullable Pipeline getPipeline() {
         return pipeline;
       }
diff --git 
a/it/google-cloud-platform/src/main/java/org/apache/beam/it/gcp/dataflow/DefaultPipelineLauncher.java
 
b/it/google-cloud-platform/src/main/java/org/apache/beam/it/gcp/dataflow/DefaultPipelineLauncher.java
index ad2dcafc007..3d43618821a 100644
--- 
a/it/google-cloud-platform/src/main/java/org/apache/beam/it/gcp/dataflow/DefaultPipelineLauncher.java
+++ 
b/it/google-cloud-platform/src/main/java/org/apache/beam/it/gcp/dataflow/DefaultPipelineLauncher.java
@@ -360,11 +360,22 @@ public class DefaultPipelineLauncher extends 
AbstractPipelineLauncher {
             options.executable() != null,
             "Cannot launch a dataflow job "
                 + "without executable specified. Please specify executable and 
try again!");
+        if (options.requirementsFile() != null) {
+          // install requirements
+          cmd.add(
+              "virtualenv . && source ./bin/activate && pip3 install -r "
+                  + options.requirementsFile());
+          cmd.add("&&");
+        }
         LOG.info("Using the executable at {}", options.executable());
         cmd.add("python3");
         cmd.add(options.executable());
         cmd.addAll(extractOptions(project, region, options));
-        jobId = executeCommandAndParseResponse(cmd);
+        if (options.requirementsFile() != null) {
+          cmd.add("&&");
+          cmd.add("deactivate");
+        }
+        jobId = executeCommandAndParseResponse(String.join(" ", cmd));
         break;
       case GO:
         checkState(
@@ -376,7 +387,7 @@ public class DefaultPipelineLauncher extends 
AbstractPipelineLauncher {
         cmd.add("run");
         cmd.add(options.executable());
         cmd.addAll(extractOptions(project, region, options));
-        jobId = executeCommandAndParseResponse(cmd);
+        jobId = executeCommandAndParseResponse(String.join(" ", cmd));
         break;
       default:
         throw new RuntimeException(
@@ -441,10 +452,13 @@ public class DefaultPipelineLauncher extends 
AbstractPipelineLauncher {
   }
 
   /** Executes the specified command and parses the response to get the Job 
ID. */
-  private String executeCommandAndParseResponse(List<String> cmd) throws 
IOException {
-    Process process = new 
ProcessBuilder().command(cmd).redirectErrorStream(true).start();
+  private String executeCommandAndParseResponse(String cmd) throws IOException 
{
+    LOG.info("Running command: {}", cmd);
+    Process process =
+        new ProcessBuilder().command("/bin/bash", "-c", 
cmd).redirectErrorStream(true).start();
     String output =
         new String(ByteStreams.toByteArray(process.getInputStream()), 
StandardCharsets.UTF_8);
+    LOG.info(output);
     Matcher m = JOB_ID_PATTERN.matcher(output);
     if (!m.find()) {
       throw new RuntimeException(