[BEAM-14038] Auto-startup for Python expansion service. #17035

robertwb · 2022-03-07T23:28:38Z

This adds a script and some java code for automatically starting a Beam Python service.

It may be useful for Go, Typescript, etc. as well.

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

Choose reviewer(s) and mention them in a comment (R: @username).
Format the pull request title like [BEAM-XXX] Fixes bug in ApproximateQuantiles, where you replace BEAM-XXX with the appropriate JIRA issue, if applicable. This will automatically link the pull request to the issue.
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.

robertwb · 2022-03-07T23:28:49Z

R: @ihji

codecov · 2022-03-07T23:47:39Z

Codecov Report

Merging #17035 (04b843d) into master (2aa4da0) will increase coverage by 0.07%.
The diff coverage is n/a.

@@            Coverage Diff             @@
##           master   #17035      +/-   ##
==========================================
+ Coverage   73.81%   73.89%   +0.07%     
==========================================
  Files         667      667              
  Lines       87340    88017     +677     
==========================================
+ Hits        64474    65039     +565     
- Misses      21763    21875     +112     
  Partials     1103     1103

Flag	Coverage Δ
python	`83.63% <ø> (-0.01%)`	⬇️

Flags with carried forward coverage won't be shown. Click here to find out more.

Impacted Files	Coverage Δ
...nternal/clients/cloudbuild/cloudbuild_v1_client.py	`52.21% <0.00%> (-2.97%)`	⬇️
..._beam/runners/portability/sdk_container_builder.py	`37.93% <0.00%> (-1.02%)`	⬇️
...che_beam/runners/interactive/interactive_runner.py	`92.02% <0.00%> (-0.73%)`	⬇️
...dks/python/apache_beam/options/pipeline_options.py	`94.85% <0.00%> (-0.43%)`	⬇️
sdks/python/apache_beam/coders/coder_impl.py	`93.97% <0.00%> (-0.31%)`	⬇️
...hon/apache_beam/runners/worker/bundle_processor.py	`93.39% <0.00%> (-0.25%)`	⬇️
...eam/runners/interactive/interactive_environment.py	`90.51% <0.00%> (-0.23%)`	⬇️
...ython/apache_beam/io/gcp/experimental/spannerio.py	`82.15% <0.00%> (-0.17%)`	⬇️
sdks/python/apache_beam/transforms/util.py	`95.96% <0.00%> (-0.17%)`	⬇️
...ks/python/apache_beam/runners/worker/sdk_worker.py	`88.90% <0.00%> (-0.16%)`	⬇️
... and 13 more

Continue to review full report at Codecov.

Legend - Click here to learn more
Δ = absolute <relative> (impact), ø = not affected, ? = missing data
Powered by Codecov. Last update 2aa4da0...04b843d. Read the comment docs.

robertwb · 2022-03-09T02:17:17Z

Failure in org.apache.beam.sdk.transforms.ParDoLifecycleTest.testTeardownCalledAfterExceptionInFinishBundleStateful

robertwb · 2022-03-09T02:17:23Z

Run Java PreCommit

robertwb · 2022-03-09T02:17:32Z

Run CommunityMetrics PreCommit

robertwb · 2022-03-10T00:01:25Z

Flake due to BEAM-8101

robertwb · 2022-03-10T00:01:30Z

Run Java PreCommit

ihji

Thanks, looks good.

ihji · 2022-03-11T00:09:58Z

...ions/python/src/main/java/org/apache/beam/sdk/extensions/python/ExternalPythonTransform.java

+                .withMultiOutputs();
+        PCollectionTuple outputs;
+        if (input instanceof PCollection) {
+          outputs = (PCollectionTuple) ((PCollection) input).apply(transform);


Do we need this explicit casting to (PCollectionTuple) ?

Yes, unfortunately. It seems java struggles with "paritally-raw" types.

OK, I was able to work around this by using PCollection<?>.

ihji · 2022-03-11T00:30:58Z

...ava/extensions/python/src/main/java/org/apache/beam/sdk/extensions/python/PythonService.java

+    List<String> bootstrapCommand = ImmutableList.of("python", bootstrapScript.getAbsolutePath());
+    LOG.info("Running bootstrap command " + bootstrapCommand);
+    Process bootstrap =
+        new ProcessBuilder("python", bootstrapScript.getAbsolutePath())


Isn't it simpler to run bootstrap_beam_venv.py not by creating a temporary copy but just passing the script text resource to python executable via stdin pipe?

Good question, I actually tried to do it this way originally. The problem is that the os.execlp bootstrap from Python 2 to Python 3 doesn't work if it isn't run as a file (as the stdin is already partially consumed). There are still too many systems out there that, despite having Python 3, map python to Python 2.x. (An alternative would be to make the full script Python 2 compatible, but that would cause ongoing pain, especially if one can't count on any dependencies like six...)

ihji · 2022-03-11T00:35:06Z

...sions/python/src/main/resources/org/apache/beam/sdk/extensions/python/bootstrap_beam_venv.py

+    parser.add_argument('--python_version', help="Python major version.")
+    parser.add_argument('--beam_version',
+                        help="Beam version.",
+                        default="2.36.0")


Is there any way to avoid hard-coded Beam version? Do we need to update this version string every time we release a new version of Beam?

We could consider making this a required argument. (Note that it's not an error to mix and match Beam versions, as long as what you're counting on is available, though it's generally best to have the latest.

ihji · 2022-03-11T00:36:58Z

...ions/python/src/main/java/org/apache/beam/sdk/extensions/python/ExternalPythonTransform.java

+              Schema.Field.of("constructor", Schema.FieldType.STRING),
+              Schema.Field.of("args", Schema.FieldType.row(args.getSchema())),
+              Schema.Field.of("kwargs", Schema.FieldType.row(kwargs.getSchema())));
+      payloadSchema.setUUID(UUID.randomUUID());


Just out of curiosity, why do we need to set UUID for the schema?

Python has a schema cache (e.g. so it's not constantly re-creating types for a dynamic schema). Results in some very surprising behavior if this is not set.

chamikaramj

Thanks.

chamikaramj · 2022-03-15T19:11:21Z

...ions/python/src/main/java/org/apache/beam/sdk/extensions/python/ExternalPythonTransform.java

+import org.apache.beam.vendor.guava.v26_0_jre.com.google.common.collect.Iterables;
+
+/** Wrapper for invoking external Python transforms. */
+public class ExternalPythonTransform<InputT extends PInput, OutputT extends POutput>


We want this to be the main API for Java cross-language transforms, right ? So probably we should advertise in the PR description.

chamikaramj · 2022-03-15T19:13:21Z

...ions/python/src/main/java/org/apache/beam/sdk/extensions/python/ExternalPythonTransform.java

+  private final Row args;
+  private final Row kwargs;
+
+  public ExternalPythonTransform(String fullyQualifiedName, Row args, Row kwargs) {


How about adding following that wraps this transform:
External.of (String transformName, Map parameters, String endpoint);

And may be deprecate existing:
External.of(String urn, byte[] payload, String endpoint)

(this can be a separate PR)

I agree that we need a nicer API, which should probably have a 1-pager and be a separate PR. (To seed this, we should have a RowBuilder that starts with the empty row and can be built with withField(name, value) with several overloads for different value types that builds up the schema dynamically. I don't know that a Map<String, Object> parameters is the most java-natural, but it could be allowed as an option as well.) I'm not convinced it makes sense to add a back-reference from External to ExternalPython.

Adding a Row builder (on top of this) seems relatively straightforward. See #17101

I think it will be confusing to add another main External transform API (in addition to External.of and ExternalPythonTransform) so probably ExternalPythonTransform should be the main API to be used along with the Row builder.

ExternalPythonTransform only makes sense for Python transforms, as it starts up a Python expansion service (rather than require one be provided) and expects a fqn + positional arg + kwargs rather than a urn + row (though one could argue row is basically kwargs).

Yeah, sorry, I meant "main API for using Python transforms from Java".

Here's a short doc that gives an updated API: https://docs.google.com/document/d/164rn4otqtQA-8QoLlaoGunUDjLSBBvS2DuR0bn344MQ/edit?usp=sharing

chamikaramj · 2022-03-15T19:29:44Z

.../python/src/test/java/org/apache/beam/sdk/extensions/python/ExternalPythonTransformTest.java

+            .apply(
+                new ExternalPythonTransform<
+                    PCollection<KV<String, String>>, PCollection<KV<String, Iterable<String>>>>(
+                    "apache_beam.GroupByKey", Row.nullRow(Schema.of()), Row.nullRow(Schema.of())))


Can we add a more comprehensive set of unit tests here that uses DirectRunner ? Also "apache_beam.GroupByKey" is not a good example since it doesn't involve the SDK during execution.

The Java DirectRunner can't run cross-language pipelines. I used GBK because I wanted something simple and obvious--eventually we'll hopefully have a richer set of transforms to invoke.

robertwb · 2022-03-15T20:10:22Z

Timeout waiting to lock Build cache (/home/jenkins/.gradle/caches/build-cache-1). It is currently in use by another Gradle instance.

robertwb · 2022-03-15T20:10:30Z

Run Java PreCommit

robertwb · 2022-03-16T00:42:08Z

Run Java PreCommit

robertwb · 2022-03-16T06:32:35Z

Run Java PreCommit

robertwb · 2022-03-16T17:36:35Z

Run Java PreCommit

robertwb · 2022-03-17T03:45:38Z

Failure in org.apache.beam.sdk.io.hadoop.format.HadoopFormatIOElasticTest.classMethod

lukecwik · 2022-03-22T20:17:05Z

ExternalPythonTransformTest is extremely flaky, for example: https://ci-beam.apache.org/job/beam_PreCommit_Java_Phrase/4806/testReport/junit/org.apache.beam.sdk.extensions.python/ExternalPythonTransformTest/trivialPythonTransform/

Filed https://issues.apache.org/jira/browse/BEAM-14148

Started rollback in #17154

…)" This reverts commit 0510cec.

…)" (#17154) This reverts commit 0510cec.

…e. (apache#17035)"" This reverts commit f5d62dd.

Generic python script for bootstrapping a Python environment.

9d32546

github-actions bot added build extensions java labels Mar 7, 2022

robertwb changed the title ~~Auto-startup for Python expansion service.~~ [BEAM-14038] Auto-startup for Python expansion service. Mar 7, 2022

robertwb force-pushed the py-expansion-service branch from b2e8bff to b6f214e Compare March 8, 2022 23:28

Add new external python extension for java.

bc7db3e

robertwb force-pushed the py-expansion-service branch from b6f214e to bc7db3e Compare March 8, 2022 23:37

ihji reviewed Mar 11, 2022

View reviewed changes

Use latest Beam rather than hard-coding it.

e01e939

robertwb force-pushed the py-expansion-service branch from 1326808 to 731b9a3 Compare March 11, 2022 06:17

Java type cleanup.

049408c

robertwb force-pushed the py-expansion-service branch from 731b9a3 to 049408c Compare March 14, 2022 16:22

Increase timeout.

c93d6e1

chamikaramj reviewed Mar 15, 2022

View reviewed changes

chamikaramj mentioned this pull request Mar 16, 2022

[BEAM-14143] Simplifies the ExternalPythonTransform API #17101

Merged

4 tasks

Build improvements.

d04e0be

debugging

850be6c

robertwb added 2 commits March 16, 2022 00:19

debugging

440b91b

debug shorter wait

96f441a

robertwb added 2 commits March 16, 2022 10:51

try to trigger jenkins

6bae873

another commit to trigger presubmits

c7c975c

remove debugging

04b843d

robertwb merged commit 0510cec into apache:master Mar 17, 2022

lukecwik added a commit that referenced this pull request Mar 22, 2022

Revert "[BEAM-14038] Auto-startup for Python expansion service. (#17035…

f5d62dd

…)" This reverts commit 0510cec.

lukecwik mentioned this pull request Mar 22, 2022

[BEAM-14148] Revert "[BEAM-14038] Auto-startup for Python expansion service." #17154

Merged

lukecwik added a commit that referenced this pull request Mar 22, 2022

Revert "[BEAM-14038] Auto-startup for Python expansion service. (#17035…

c8013d9

…)" (#17154) This reverts commit 0510cec.

robertwb added a commit to robertwb/incubator-beam that referenced this pull request Mar 31, 2022

Revert "Revert "[BEAM-14038] Auto-startup for Python expansion servic…

30ebe10

…e. (apache#17035)"" This reverts commit f5d62dd.

[BEAM-14038] Auto-startup for Python expansion service. #17035

[BEAM-14038] Auto-startup for Python expansion service. #17035

Conversation

robertwb commented Mar 7, 2022

GitHub Actions Tests Status (on master branch)

robertwb commented Mar 7, 2022

codecov bot commented Mar 7, 2022 • edited Loading

Codecov Report

robertwb commented Mar 9, 2022

robertwb commented Mar 9, 2022

robertwb commented Mar 9, 2022

robertwb commented Mar 10, 2022

robertwb commented Mar 10, 2022

ihji left a 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

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

chamikaramj left a 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

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

robertwb commented Mar 15, 2022

robertwb commented Mar 15, 2022

robertwb commented Mar 16, 2022

robertwb commented Mar 16, 2022

robertwb commented Mar 16, 2022

robertwb commented Mar 17, 2022

lukecwik commented Mar 22, 2022 • edited Loading

codecov bot commented Mar 7, 2022 •

edited

Loading

lukecwik commented Mar 22, 2022 •

edited

Loading