Allow local runners to execute arbitrary cross language pipelines without Docker. #29283

robertwb · 2023-11-03T00:36:12Z

This is accomplished by re-using the expansion service subprocesses as the workers themselves.

Thank you for your contribution! Follow this checklist to help us incorporate your contribution quickly and easily:

Mention the appropriate issue in your description (for example: addresses #123), if applicable. This will automatically add a link to the pull request in the issue. If you would like the issue to automatically close on merging the pull request, comment fixes #<ISSUE NUMBER> instead.
Update CHANGES.md with noteworthy changes.
If this contribution is large, please file an Apache Individual Contributor License Agreement.

See the Contributor Guide for more tips on how to make review process smoother.

To check the build health, please visit https://github.com/apache/beam/blob/master/.test-infra/BUILD_STATUS.md

GitHub Actions Tests Status (on master branch)

See CI.md for more information about GitHub Actions CI or the workflows README to see a list of phrases to trigger workflows.

model/pipeline/src/main/proto/org/apache/beam/model/pipeline/v1/beam_runner_api.proto

...-construction-java/src/main/java/org/apache/beam/runners/core/construction/Environments.java

github-actions · 2023-11-03T02:12:48Z

Checks are failing. Will not request review until checks are succeeding. If you'd like to override that behavior, comment assign set of reviewers

github-actions · 2023-11-08T21:34:47Z

Assigning reviewers. If you would like to opt out of this review, comment assign to next reviewer:

R: @tvalentyn for label python.
R: @m-trieu for label java.

Available commands:

stop reviewer notifications - opt out of the automated review tooling
remind me after tests pass - tag the comment author after tests pass
waiting on author - shift the attention set back to the author (any comment or push by the author will return the attention set to the reviewers)

The PR bot will only process comments in the main thread (not review comments).

m-trieu · 2023-11-08T22:00:19Z

...-construction-java/src/main/java/org/apache/beam/runners/core/construction/Environments.java

+        .build();
+  }
+
+  public static List<Environment> expandAnyOfEnvironments(Environment environment) {


Can this be an ImmutableList?

I generally try to avoid exposing (our vendered) guava in our public API. Given that this is created on the fly, there's little benefit to making it immutable.

m-trieu · 2023-11-08T22:01:57Z

...java-job-service/src/main/java/org/apache/beam/runners/jobsubmission/InMemoryJobService.java

@@ -287,26 +283,50 @@ public void run(RunJobRequest request, StreamObserver<RunJobResponse> responseOb
    }
  }

+  private Map<String, List<RunnerApi.ArtifactInformation>> extractDependencies(


Can this return/be an ImmutableMap

Same as above. (I suppose here the method is private, but it's stell IMHO preferable to not declare implementation details in the signature.)

m-trieu · 2023-11-08T22:04:44Z

...java-job-service/src/main/java/org/apache/beam/runners/jobsubmission/InMemoryJobService.java

+    }
+    return dependencies;
+  }
+
  private RunnerApi.Pipeline resolveDependencies(RunnerApi.Pipeline pipeline, String stagingToken) {
    Map<String, List<RunnerApi.ArtifactInformation>> resolvedDependencies =


Instead of allowing null here can we default to an empty map if stagingService.getService().getStagedArtifacts(stagingToken) returns null?

or wrap in an optional

can mark as @Nullable if the other 2 are not feasible

null seems to indicate that no staging was attempted here (which is technically fine for something that has no artifacts, but we can't assume that if there's no entry). This is different from "the dependencies resolved to the empty set" which would be an error to just accept.

Marking it as @nullable is painful as the null checker isn't smart enough to figure out that it'll never be null in the second clause.

Ack thanks!

tvalentyn · 2023-11-15T02:03:55Z

@robertwb is this still a WIP or ready to review?

robertwb · 2023-11-17T20:17:34Z

Yes, this is ready for review. (I was trying to get all the presubmits to pass, but I think this is the CI go issues that have been plaguing us elsewhere.)

lostluck · 2023-11-17T20:30:11Z

I can confirm I'm chatting down flakes in prism.

github-actions · 2023-11-25T12:13:36Z

Reminder, please take a look at this pr: @tvalentyn @m-trieu

This allows an SDK (including, importantly, an expansion service) to provide several alternatives environments suitable for running a pipeline.

robertwb · 2023-11-29T19:57:01Z

I rebased and resolved the conflicts with master. Please take a look.

sdks/python/apache_beam/runners/portability/fn_api_runner/fn_runner.py

tvalentyn · 2023-11-29T19:54:07Z

sdks/python/apache_beam/runners/portability/fn_api_runner/fn_runner.py

-      pipeline_proto.components.environments[docker_env_id].CopyFrom(
-          embedded_env)
+      for docker_env_id in python_docker_environments:
+        pipeline_proto.components.environments[docker_env_id].CopyFrom(


for my education, what is the meaning of this? We initializing missing attributes (env capabilities, etc) on the docker environment using default embedded environment as template?

Also, could you please add a docstring to embed_default_docker_image ? Thanks!

I've changed this as I realize it wasn't very clear. You can't assign to a proto map in Python, you have to use CopyFrom.

Also, docstring added.

sdks/python/apache_beam/runners/portability/local_job_service.py

sdks/python/apache_beam/utils/subprocess_server.py

tvalentyn · 2023-11-29T20:32:50Z

sdks/python/apache_beam/utils/subprocess_server.py

+      if key not in self._cache:
+        self._cache[key] = (self._constructor(*key), set())
+      for owner in self._live_owners:
+        self._cache[key][-1].add(owner)


should we store NamedTuples instead of tuples so that self._cache[key][-1] becomes self._cache[key].owners ?

sdks/python/apache_beam/utils/subprocess_server.py

tvalentyn · 2023-11-29T20:53:10Z

sdks/python/apache_beam/utils/subprocess_server.py

+    for _ in range(5):
+      if process.poll() is not None:
+        break
+      logging.debug("Sending SIGINT to job_server")


is the process always a job_server?

No. Updated.

sdks/python/apache_beam/transforms/external.py

m-trieu

LGTM for java

tvalentyn

Thank you! Also, thanks for structuring commits in a way that made the change easier to review.

tvalentyn · 2023-11-30T23:10:37Z

PTAL at lint. ERROR: /runner/_work/beam/beam/sdks/python/test-suites/tox/py38/build/srcs/sdks/python/apache_beam/utils/subprocess_server.py Imports are incorrectly sorted.

robertwb · 2023-11-30T23:37:45Z

Thanks! Yeah, I'll fix that lint error before merging.

…peline. This can greatly reduce startup time when many cross-langauge transforms are used, but more importantly by keeping these processes alive we open up the potential for using them as workers as well. These can be cached across longer durations as well, but this is the default.

Due to the AnyOf environment, remote runners can choose more expensive but remote-friendly options such as docker.

Abacn · 2024-03-06T19:57:25Z

It appears Python XVR Direct is failing after this change. Specifically #28972 (comment) any ideas?

Last successful run (before it was flaky passing 1-2 jobs): https://github.com/apache/beam/actions/runs/7059453030
First failed run (after that both jobs always failing): https://github.com/apache/beam/actions/runs/7063238797

and this is the only relavant change:

github-actions bot added python java model runners dataflow core fn-execution labels Nov 3, 2023

robertwb force-pushed the expansion-loopback branch from 615a874 to 292cf84 Compare November 3, 2023 00:44

lostluck reviewed Nov 3, 2023

View reviewed changes

model/pipeline/src/main/proto/org/apache/beam/model/pipeline/v1/beam_runner_api.proto Show resolved Hide resolved

...-construction-java/src/main/java/org/apache/beam/runners/core/construction/Environments.java Outdated Show resolved Hide resolved

robertwb force-pushed the expansion-loopback branch 8 times, most recently from f8477a7 to 93af8ac Compare November 8, 2023 20:25

github-actions bot added the Next Action: Reviewers label Nov 8, 2023

m-trieu suggested changes Nov 8, 2023

View reviewed changes

robertwb mentioned this pull request Nov 9, 2023

Expansion loopback testing #29374

Closed

3 tasks

github-actions bot added the slow-review label Nov 25, 2023

robertwb added 2 commits November 29, 2023 11:53

Add new AnyOf environment.

90b1a6b

This allows an SDK (including, importantly, an expansion service) to provide several alternatives environments suitable for running a pipeline.

Add AnyOf environment handling to the various runners.

e5a9ab6

robertwb force-pushed the expansion-loopback branch from 93af8ac to 825e494 Compare November 29, 2023 19:56

tvalentyn reviewed Nov 29, 2023

View reviewed changes

m-trieu approved these changes Nov 29, 2023

View reviewed changes

robertwb force-pushed the expansion-loopback branch from 825e494 to bd9fe38 Compare November 30, 2023 19:33

Add AnyOf environment handling to the artifact staging services.

b136f1d

robertwb force-pushed the expansion-loopback branch from bd9fe38 to d7fcd2f Compare November 30, 2023 20:35

tvalentyn approved these changes Nov 30, 2023

View reviewed changes

github-actions bot removed the slow-review label Nov 30, 2023

robertwb added 4 commits November 30, 2023 16:22

Add an option to start loopback workers in expansion services.

e8cebdf

Use Java expansion services as workers by default in Python.

e88fb5d

Due to the AnyOf environment, remote runners can choose more expensive but remote-friendly options such as docker.

More complete documentation of environment proto definitions.

d85ce3f

robertwb force-pushed the expansion-loopback branch from d7fcd2f to d85ce3f Compare December 1, 2023 00:22

robertwb merged commit f337c74 into apache:master Dec 1, 2023
97 checks passed

robertwb mentioned this pull request Apr 4, 2024

Proposed edits for Beam YAML overview #30842

Merged

3 tasks

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Allow local runners to execute arbitrary cross language pipelines without Docker. #29283

Allow local runners to execute arbitrary cross language pipelines without Docker. #29283

robertwb commented Nov 3, 2023

github-actions bot commented Nov 3, 2023

github-actions bot commented Nov 8, 2023

m-trieu Nov 8, 2023

robertwb Nov 17, 2023

m-trieu Nov 8, 2023

robertwb Nov 17, 2023

m-trieu Nov 8, 2023

robertwb Nov 17, 2023

m-trieu Nov 28, 2023

tvalentyn commented Nov 15, 2023

robertwb commented Nov 17, 2023

lostluck commented Nov 17, 2023

github-actions bot commented Nov 25, 2023

robertwb commented Nov 29, 2023

tvalentyn Nov 29, 2023

robertwb Nov 30, 2023

tvalentyn Nov 29, 2023

robertwb Nov 30, 2023

tvalentyn Nov 29, 2023

robertwb Nov 30, 2023

m-trieu left a comment

tvalentyn left a comment •

edited

Loading

tvalentyn commented Nov 30, 2023

robertwb commented Nov 30, 2023

Abacn commented Mar 6, 2024 •

edited

Loading

Allow local runners to execute arbitrary cross language pipelines without Docker. #29283

Allow local runners to execute arbitrary cross language pipelines without Docker. #29283

Conversation

robertwb commented Nov 3, 2023

GitHub Actions Tests Status (on master branch)

github-actions bot commented Nov 3, 2023

github-actions bot commented Nov 8, 2023

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

tvalentyn commented Nov 15, 2023

robertwb commented Nov 17, 2023

lostluck commented Nov 17, 2023

github-actions bot commented Nov 25, 2023

robertwb commented Nov 29, 2023

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

m-trieu left a comment

Choose a reason for hiding this comment

tvalentyn left a comment • edited Loading

Choose a reason for hiding this comment

tvalentyn commented Nov 30, 2023

robertwb commented Nov 30, 2023

Abacn commented Mar 6, 2024 • edited Loading

tvalentyn left a comment •

edited

Loading

Abacn commented Mar 6, 2024 •

edited

Loading