You are viewing a plain text version of this content. The canonical link for it is here.
Posted to github@beam.apache.org by GitBox <gi...@apache.org> on 2022/06/10 21:06:21 UTC
[GitHub] [beam] ihji opened a new pull request, #21809: [WIP/BEAM-14506] Adding testcases and examples for xlang Python RunInference
ihji opened a new pull request, #21809:
URL: https://github.com/apache/beam/pull/21809
`./gradlew :sdks:python:test-suites:direct:xlang:validatesCrossLanguageRunnerJavaUsingPython --tests RunInferenceTransformTest`
------------------------
Thank you for your contribution! Follow this checklist to help us incorporate your contribution quickly and easily:
- [ ] [**Choose reviewer(s)**](https://beam.apache.org/contribute/#make-your-change) and mention them in a comment (`R: @username`).
- [ ] 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](https://www.apache.org/licenses/icla.pdf).
See the [Contributor Guide](https://beam.apache.org/contribute) for more tips on [how to make review process smoother](https://beam.apache.org/contribute/#make-reviewers-job-easier).
To check the build health, please visit [https://github.com/apache/beam/blob/master/.test-infra/BUILD_STATUS.md](https://github.com/apache/beam/blob/master/.test-infra/BUILD_STATUS.md)
GitHub Actions Tests Status (on master branch)
------------------------------------------------------------------------------------------------
[](https://github.com/apache/beam/actions?query=workflow%3A%22Build+python+source+distribution+and+wheels%22+branch%3Amaster+event%3Aschedule)
[](https://github.com/apache/beam/actions?query=workflow%3A%22Python+Tests%22+branch%3Amaster+event%3Aschedule)
[](https://github.com/apache/beam/actions?query=workflow%3A%22Java+Tests%22+branch%3Amaster+event%3Aschedule)
See [CI.md](https://github.com/apache/beam/blob/master/CI.md) for more information about GitHub Actions CI.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] ihji commented on pull request #21809: [BEAM-14506] Adding testcases and examples for xlang Python RunInference
Posted by GitBox <gi...@apache.org>.
ihji commented on PR #21809:
URL: https://github.com/apache/beam/pull/21809#issuecomment-1163932275
Run XVR_Direct PostCommit
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] ihji commented on pull request #21809: [BEAM-14506] Adding testcases and examples for xlang Python RunInference
Posted by GitBox <gi...@apache.org>.
ihji commented on PR #21809:
URL: https://github.com/apache/beam/pull/21809#issuecomment-1164871638
Run XVR_Direct PostCommit
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] robertwb commented on a diff in pull request #21809: [BEAM-14506] Adding testcases and examples for xlang Python RunInference
Posted by GitBox <gi...@apache.org>.
robertwb commented on code in PR #21809:
URL: https://github.com/apache/beam/pull/21809#discussion_r918459953
##########
sdks/java/extensions/python/src/test/java/org/apache/beam/sdk/extensions/python/transforms/RunInferenceTransformTest.java:
##########
@@ -0,0 +1,68 @@
+/*
+ * 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.beam.sdk.extensions.python.transforms;
+
+import java.util.Arrays;
+import java.util.Optional;
+import org.apache.beam.runners.core.construction.BaseExternalTest;
+import org.apache.beam.sdk.coders.IterableCoder;
+import org.apache.beam.sdk.coders.VarLongCoder;
+import org.apache.beam.sdk.schemas.Schema;
+import org.apache.beam.sdk.testing.PAssert;
+import org.apache.beam.sdk.testing.UsesPythonExpansionService;
+import org.apache.beam.sdk.testing.ValidatesRunner;
+import org.apache.beam.sdk.transforms.Create;
+import org.apache.beam.sdk.values.PCollection;
+import org.apache.beam.sdk.values.Row;
+import org.junit.Test;
+import org.junit.experimental.categories.Category;
+import org.junit.runner.RunWith;
+import org.junit.runners.JUnit4;
+
+@RunWith(JUnit4.class)
+public class RunInferenceTransformTest extends BaseExternalTest {
+ @Test
+ @Category({ValidatesRunner.class, UsesPythonExpansionService.class})
+ public void testRunInference() {
+ String stagingLocation =
+ Optional.ofNullable(System.getProperty("semiPersistDir")).orElse("/tmp");
+ Schema schema =
+ Schema.of(
+ Schema.Field.of("example", Schema.FieldType.array(Schema.FieldType.INT64)),
+ Schema.Field.of("inference", Schema.FieldType.INT32));
+ Row row0 = Row.withSchema(schema).addArray(0L, 0L).addValue(0).build();
+ Row row1 = Row.withSchema(schema).addArray(1L, 1L).addValue(1).build();
+ PCollection<Row> col =
+ testPipeline
+ .apply(Create.<Iterable<Long>>of(Arrays.asList(0L, 0L), Arrays.asList(1L, 1L)))
+ .setCoder(IterableCoder.of(VarLongCoder.of()))
+ .apply(
+ RunInference.of(
+ "apache_beam.ml.inference.sklearn_inference.SklearnModelHandlerNumpy",
+ schema)
+ .withKwarg(
+ // The test expansion service creates the test model and saves it to the
+ // returning external environment as a dependency.
+ // (sdks/python/apache_beam/runners/portability/expansion_service_test.py)
+ // The dependencies for Python SDK harness are supposed to be staged to
Review Comment:
Just to clarify, this is the contract? (For file dependencies with a stage_to attribute?) Probably not worth blocking this PR, but I think we should document this better.
##########
runners/google-cloud-dataflow-java/build.gradle:
##########
@@ -413,6 +413,7 @@ createCrossLanguageValidatesRunnerTask(
classpath: configurations.validatesRunner,
numParallelTests: Integer.MAX_VALUE,
needsSdkLocation: true,
+ semiPersistDir: "/var/opt/google",
Review Comment:
Why do we need to override this here?
##########
buildSrc/src/main/groovy/org/apache/beam/gradle/BeamModulePlugin.groovy:
##########
@@ -351,6 +351,8 @@ class BeamModulePlugin implements Plugin<Project> {
Integer numParallelTests = 1
// Whether the pipeline needs --sdk_location option
boolean needsSdkLocation = false
+ // semi_persist_dir for SDK containers
+ String semiPersistDir = "/tmp"
Review Comment:
Will this get cleaned up or accumulate junk (until a restart)?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] codecov[bot] commented on pull request #21809: [WIP/BEAM-14506] Adding testcases and examples for xlang Python RunInference
Posted by GitBox <gi...@apache.org>.
codecov[bot] commented on PR #21809:
URL: https://github.com/apache/beam/pull/21809#issuecomment-1154597760
# [Codecov](https://codecov.io/gh/apache/beam/pull/21809?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation) Report
> Merging [#21809](https://codecov.io/gh/apache/beam/pull/21809?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation) (bdf1d39) into [master](https://codecov.io/gh/apache/beam/commit/383c08d7d52a995176e97afecac71358d0422ac1?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation) (383c08d) will **decrease** coverage by `0.01%`.
> The diff coverage is `80.00%`.
```diff
@@ Coverage Diff @@
## master #21809 +/- ##
==========================================
- Coverage 74.15% 74.14% -0.02%
==========================================
Files 698 698
Lines 92417 92414 -3
==========================================
- Hits 68530 68518 -12
- Misses 22636 22645 +9
Partials 1251 1251
```
| Flag | Coverage Δ | |
|---|---|---|
| python | `83.75% <80.00%> (-0.02%)` | :arrow_down: |
Flags with carried forward coverage won't be shown. [Click here](https://docs.codecov.io/docs/carryforward-flags?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#carryforward-flags-in-the-pull-request-comment) to find out more.
| [Impacted Files](https://codecov.io/gh/apache/beam/pull/21809?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation) | Coverage Δ | |
|---|---|---|
| [sdks/python/apache\_beam/ml/inference/base.py](https://codecov.io/gh/apache/beam/pull/21809/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-c2Rrcy9weXRob24vYXBhY2hlX2JlYW0vbWwvaW5mZXJlbmNlL2Jhc2UucHk=) | `95.23% <80.00%> (+0.17%)` | :arrow_up: |
| [sdks/python/apache\_beam/utils/interactive\_utils.py](https://codecov.io/gh/apache/beam/pull/21809/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-c2Rrcy9weXRob24vYXBhY2hlX2JlYW0vdXRpbHMvaW50ZXJhY3RpdmVfdXRpbHMucHk=) | `95.12% <0.00%> (-2.44%)` | :arrow_down: |
| [...n/apache\_beam/ml/gcp/recommendations\_ai\_test\_it.py](https://codecov.io/gh/apache/beam/pull/21809/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-c2Rrcy9weXRob24vYXBhY2hlX2JlYW0vbWwvZ2NwL3JlY29tbWVuZGF0aW9uc19haV90ZXN0X2l0LnB5) | `73.46% <0.00%> (-2.05%)` | :arrow_down: |
| [.../python/apache\_beam/transforms/periodicsequence.py](https://codecov.io/gh/apache/beam/pull/21809/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-c2Rrcy9weXRob24vYXBhY2hlX2JlYW0vdHJhbnNmb3Jtcy9wZXJpb2RpY3NlcXVlbmNlLnB5) | `96.72% <0.00%> (-1.64%)` | :arrow_down: |
| [sdks/python/apache\_beam/io/source\_test\_utils.py](https://codecov.io/gh/apache/beam/pull/21809/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-c2Rrcy9weXRob24vYXBhY2hlX2JlYW0vaW8vc291cmNlX3Rlc3RfdXRpbHMucHk=) | `88.01% <0.00%> (-1.39%)` | :arrow_down: |
| [...eam/runners/portability/fn\_api\_runner/execution.py](https://codecov.io/gh/apache/beam/pull/21809/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-c2Rrcy9weXRob24vYXBhY2hlX2JlYW0vcnVubmVycy9wb3J0YWJpbGl0eS9mbl9hcGlfcnVubmVyL2V4ZWN1dGlvbi5weQ==) | `92.44% <0.00%> (-0.65%)` | :arrow_down: |
| [...hon/apache\_beam/runners/worker/bundle\_processor.py](https://codecov.io/gh/apache/beam/pull/21809/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-c2Rrcy9weXRob24vYXBhY2hlX2JlYW0vcnVubmVycy93b3JrZXIvYnVuZGxlX3Byb2Nlc3Nvci5weQ==) | `93.67% <0.00%> (ø)` | |
| [...examples/inference/pytorch\_image\_classification.py](https://codecov.io/gh/apache/beam/pull/21809/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-c2Rrcy9weXRob24vYXBhY2hlX2JlYW0vZXhhbXBsZXMvaW5mZXJlbmNlL3B5dG9yY2hfaW1hZ2VfY2xhc3NpZmljYXRpb24ucHk=) | `0.00% <0.00%> (ø)` | |
| [sdks/python/apache\_beam/ml/inference/api.py](https://codecov.io/gh/apache/beam/pull/21809/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-c2Rrcy9weXRob24vYXBhY2hlX2JlYW0vbWwvaW5mZXJlbmNlL2FwaS5weQ==) | | |
| [sdks/python/apache\_beam/ml/inference/\_\_init\_\_.py](https://codecov.io/gh/apache/beam/pull/21809/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-c2Rrcy9weXRob24vYXBhY2hlX2JlYW0vbWwvaW5mZXJlbmNlL19faW5pdF9fLnB5) | `100.00% <0.00%> (ø)` | |
| ... and [4 more](https://codecov.io/gh/apache/beam/pull/21809/diff?src=pr&el=tree-more&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation) | |
------
[Continue to review full report at Codecov](https://codecov.io/gh/apache/beam/pull/21809?src=pr&el=continue&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation).
> **Legend** - [Click here to learn more](https://docs.codecov.io/docs/codecov-delta?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation)
> `Δ = absolute <relative> (impact)`, `ø = not affected`, `? = missing data`
> Powered by [Codecov](https://codecov.io/gh/apache/beam/pull/21809?src=pr&el=footer&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation). Last update [383c08d...bdf1d39](https://codecov.io/gh/apache/beam/pull/21809?src=pr&el=lastupdated&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation). Read the [comment docs](https://docs.codecov.io/docs/pull-request-comments?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation).
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] ihji commented on pull request #21809: [BEAM-14506] Adding testcases and examples for xlang Python RunInference
Posted by GitBox <gi...@apache.org>.
ihji commented on PR #21809:
URL: https://github.com/apache/beam/pull/21809#issuecomment-1165766354
Run SQL PreCommit
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] TheNeuralBit commented on a diff in pull request #21809: [BEAM-14506] Adding testcases and examples for xlang Python RunInference
Posted by GitBox <gi...@apache.org>.
TheNeuralBit commented on code in PR #21809:
URL: https://github.com/apache/beam/pull/21809#discussion_r912079491
##########
sdks/java/extensions/python/src/main/java/org/apache/beam/sdk/extensions/python/transforms/RunInference.java:
##########
@@ -0,0 +1,102 @@
+/*
+ * 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.beam.sdk.extensions.python.transforms;
+
+import java.util.Map;
+import org.apache.beam.sdk.coders.RowCoder;
+import org.apache.beam.sdk.extensions.python.PythonExternalTransform;
+import org.apache.beam.sdk.schemas.Schema;
+import org.apache.beam.sdk.transforms.PTransform;
+import org.apache.beam.sdk.util.PythonCallableSource;
+import org.apache.beam.sdk.values.PCollection;
+import org.apache.beam.sdk.values.Row;
+import org.apache.beam.vendor.guava.v26_0_jre.com.google.common.collect.ImmutableMap;
+
+/** Wrapper for invoking external Python RunInference. */
+public class RunInference extends PTransform<PCollection<?>, PCollection<Row>> {
+ private final String modelLoader;
+ private final Schema schema;
+ private final Map<String, Object> kwargs;
+ private final String expansionService;
+
+ /**
+ * Instantiates a multi-language wrapper for a Python RunInference with a given model loader.
+ *
+ * @param modelLoader A Python lambda function for a model loader class object.
+ * @param exampleType A schema field type for the example column in output rows.
+ * @param inferenceType A schema field type for the inference column in output rows.
+ * @return A {@link RunInference} for the given model loader.
+ */
+ public static RunInference of(
+ String modelLoader, Schema.FieldType exampleType, Schema.FieldType inferenceType) {
+ Schema schema =
+ Schema.of(
+ Schema.Field.of("example", exampleType), Schema.Field.of("inference", inferenceType));
+ return new RunInference(modelLoader, schema, ImmutableMap.of(), "");
+ }
+
+ /**
+ * Instantiates a multi-language wrapper for a Python RunInference with a given model loader.
+ *
+ * @param modelLoader A Python lambda function for a model loader class object.
+ * @param schema A schema for output rows.
+ * @return A {@link RunInference} for the given model loader.
+ */
+ public static RunInference of(String modelLoader, Schema schema) {
+ return new RunInference(modelLoader, schema, ImmutableMap.of(), "");
+ }
+
+ /**
+ * Sets keyword arguments for RunInference constructor.
+ *
+ * @return A {@link RunInference} with keyword arguments.
+ */
+ public RunInference withKwarg(String key, Object arg) {
+ ImmutableMap.Builder<String, Object> builder =
+ ImmutableMap.<String, Object>builder().putAll(kwargs).put(key, arg);
+ return new RunInference(modelLoader, schema, builder.build(), expansionService);
+ }
+
+ /**
+ * Sets an expansion service endpoint for RunInference.
+ *
+ * @param expansionService A URL for a Python expansion service.
+ * @return A {@link RunInference} for the given expansion service endpoint.
+ */
+ public RunInference withExpansionService(String expansionService) {
+ return new RunInference(modelLoader, schema, kwargs, expansionService);
+ }
+
+ private RunInference(
+ String modelLoader, Schema schema, Map<String, Object> kwargs, String expansionService) {
+ this.modelLoader = modelLoader;
+ this.schema = schema;
+ this.kwargs = kwargs;
+ this.expansionService = expansionService;
+ }
+
+ @Override
+ public PCollection<Row> expand(PCollection<?> input) {
+ return input.apply(
+ PythonExternalTransform.<PCollection<?>, PCollection<Row>>from(
+ "apache_beam.ml.inference.base.RunInference.create", expansionService)
+ .withKwarg("model_handler_provider", PythonCallableSource.of(modelLoader))
+ .withKwargs(kwargs)
+ .withOutputCoder(RowCoder.of(schema)));
Review Comment:
I'm surprised that PythonExternalTransform has `withOutputCoder` and `withOutputCoders` methods. Shouldn't the Python SDK be responsible for inferring the coder/schema for the PTransform's output?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] ihji commented on pull request #21809: [BEAM-14506] Adding testcases and examples for xlang Python RunInference
Posted by GitBox <gi...@apache.org>.
ihji commented on PR #21809:
URL: https://github.com/apache/beam/pull/21809#issuecomment-1163931651
Run XVR_Flink PostCommit
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] robertwb commented on a diff in pull request #21809: [BEAM-14506] Adding testcases and examples for xlang Python RunInference
Posted by GitBox <gi...@apache.org>.
robertwb commented on code in PR #21809:
URL: https://github.com/apache/beam/pull/21809#discussion_r919224050
##########
sdks/java/extensions/python/src/test/java/org/apache/beam/sdk/extensions/python/transforms/RunInferenceTransformTest.java:
##########
@@ -0,0 +1,68 @@
+/*
+ * 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.beam.sdk.extensions.python.transforms;
+
+import java.util.Arrays;
+import java.util.Optional;
+import org.apache.beam.runners.core.construction.BaseExternalTest;
+import org.apache.beam.sdk.coders.IterableCoder;
+import org.apache.beam.sdk.coders.VarLongCoder;
+import org.apache.beam.sdk.schemas.Schema;
+import org.apache.beam.sdk.testing.PAssert;
+import org.apache.beam.sdk.testing.UsesPythonExpansionService;
+import org.apache.beam.sdk.testing.ValidatesRunner;
+import org.apache.beam.sdk.transforms.Create;
+import org.apache.beam.sdk.values.PCollection;
+import org.apache.beam.sdk.values.Row;
+import org.junit.Test;
+import org.junit.experimental.categories.Category;
+import org.junit.runner.RunWith;
+import org.junit.runners.JUnit4;
+
+@RunWith(JUnit4.class)
+public class RunInferenceTransformTest extends BaseExternalTest {
+ @Test
+ @Category({ValidatesRunner.class, UsesPythonExpansionService.class})
+ public void testRunInference() {
+ String stagingLocation =
+ Optional.ofNullable(System.getProperty("semiPersistDir")).orElse("/tmp");
+ Schema schema =
+ Schema.of(
+ Schema.Field.of("example", Schema.FieldType.array(Schema.FieldType.INT64)),
+ Schema.Field.of("inference", Schema.FieldType.INT32));
+ Row row0 = Row.withSchema(schema).addArray(0L, 0L).addValue(0).build();
+ Row row1 = Row.withSchema(schema).addArray(1L, 1L).addValue(1).build();
+ PCollection<Row> col =
+ testPipeline
+ .apply(Create.<Iterable<Long>>of(Arrays.asList(0L, 0L), Arrays.asList(1L, 1L)))
+ .setCoder(IterableCoder.of(VarLongCoder.of()))
+ .apply(
+ RunInference.of(
+ "apache_beam.ml.inference.sklearn_inference.SklearnModelHandlerNumpy",
+ schema)
+ .withKwarg(
+ // The test expansion service creates the test model and saves it to the
+ // returning external environment as a dependency.
+ // (sdks/python/apache_beam/runners/portability/expansion_service_test.py)
+ // The dependencies for Python SDK harness are supposed to be staged to
Review Comment:
(I guess I am a bit concerned that people will try to use this as an example to follow, but maybe that's more a question for the docs...)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] robertwb commented on a diff in pull request #21809: [BEAM-14506] Adding testcases and examples for xlang Python RunInference
Posted by GitBox <gi...@apache.org>.
robertwb commented on code in PR #21809:
URL: https://github.com/apache/beam/pull/21809#discussion_r919219341
##########
runners/google-cloud-dataflow-java/build.gradle:
##########
@@ -413,6 +413,7 @@ createCrossLanguageValidatesRunnerTask(
classpath: configurations.validatesRunner,
numParallelTests: Integer.MAX_VALUE,
needsSdkLocation: true,
+ semiPersistDir: "/var/opt/google",
Review Comment:
Ah, this internal container thing keeps biting us. @tvalentyn
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] ihji commented on a diff in pull request #21809: [BEAM-14506] Adding testcases and examples for xlang Python RunInference
Posted by GitBox <gi...@apache.org>.
ihji commented on code in PR #21809:
URL: https://github.com/apache/beam/pull/21809#discussion_r918453264
##########
sdks/java/extensions/python/src/test/java/org/apache/beam/sdk/extensions/python/transforms/RunInferenceTransformTest.java:
##########
@@ -0,0 +1,63 @@
+/*
+ * 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.beam.sdk.extensions.python.transforms;
+
+import java.util.Arrays;
+import java.util.Optional;
+import org.apache.beam.runners.core.construction.BaseExternalTest;
+import org.apache.beam.sdk.coders.IterableCoder;
+import org.apache.beam.sdk.coders.VarLongCoder;
+import org.apache.beam.sdk.schemas.Schema;
+import org.apache.beam.sdk.testing.PAssert;
+import org.apache.beam.sdk.testing.UsesPythonExpansionService;
+import org.apache.beam.sdk.testing.ValidatesRunner;
+import org.apache.beam.sdk.transforms.Create;
+import org.apache.beam.sdk.values.PCollection;
+import org.apache.beam.sdk.values.Row;
+import org.junit.Test;
+import org.junit.experimental.categories.Category;
+import org.junit.runner.RunWith;
+import org.junit.runners.JUnit4;
+
+@RunWith(JUnit4.class)
+public class RunInferenceTransformTest extends BaseExternalTest {
+ @Test
+ @Category({ValidatesRunner.class, UsesPythonExpansionService.class})
+ public void testRunInference() {
+ String stagingLocation =
+ Optional.ofNullable(System.getProperty("semiPersistDir")).orElse("/tmp");
+ Schema schema =
+ Schema.of(
+ Schema.Field.of("example", Schema.FieldType.array(Schema.FieldType.INT64)),
+ Schema.Field.of("inference", Schema.FieldType.INT32));
+ Row row0 = Row.withSchema(schema).addArray(0L, 0L).addValue(0).build();
+ Row row1 = Row.withSchema(schema).addArray(1L, 1L).addValue(1).build();
+ PCollection<Row> col =
+ testPipeline
+ .apply(Create.<Iterable<Long>>of(Arrays.asList(0L, 0L), Arrays.asList(1L, 1L)))
+ .setCoder(IterableCoder.of(VarLongCoder.of()))
+ .apply(
+ RunInference.of(
+ "apache_beam.ml.inference.sklearn_inference.SklearnModelHandlerNumpy",
+ schema)
+ .withKwarg(
+ "model_uri", String.format("%s/staged/sklearn_model", stagingLocation))
Review Comment:
Added comment.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] ihji commented on a diff in pull request #21809: [BEAM-14506] Adding testcases and examples for xlang Python RunInference
Posted by GitBox <gi...@apache.org>.
ihji commented on code in PR #21809:
URL: https://github.com/apache/beam/pull/21809#discussion_r918453064
##########
buildSrc/src/main/groovy/org/apache/beam/gradle/BeamModulePlugin.groovy:
##########
@@ -2309,8 +2311,10 @@ class BeamModulePlugin implements Plugin<Project> {
throw new GradleException("unsupported java version.")
}
def setupTask = project.tasks.register(config.name+"Setup", Exec) {
- dependsOn ':sdks:java:container:'+javaContainerSuffix+':docker'
- dependsOn ':sdks:python:container:py'+pythonContainerSuffix+':docker'
+ if (!project.hasProperty('skipContainerBuilds')) {
Review Comment:
Removed.
##########
sdks/java/extensions/python/src/main/java/org/apache/beam/sdk/extensions/python/transforms/RunInference.java:
##########
@@ -0,0 +1,102 @@
+/*
+ * 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.beam.sdk.extensions.python.transforms;
+
+import java.util.Map;
+import org.apache.beam.sdk.coders.RowCoder;
+import org.apache.beam.sdk.extensions.python.PythonExternalTransform;
+import org.apache.beam.sdk.schemas.Schema;
+import org.apache.beam.sdk.transforms.PTransform;
+import org.apache.beam.sdk.util.PythonCallableSource;
+import org.apache.beam.sdk.values.PCollection;
+import org.apache.beam.sdk.values.Row;
+import org.apache.beam.vendor.guava.v26_0_jre.com.google.common.collect.ImmutableMap;
+
+/** Wrapper for invoking external Python RunInference. */
+public class RunInference extends PTransform<PCollection<?>, PCollection<Row>> {
+ private final String modelLoader;
+ private final Schema schema;
+ private final Map<String, Object> kwargs;
+ private final String expansionService;
+
+ /**
+ * Instantiates a multi-language wrapper for a Python RunInference with a given model loader.
+ *
+ * @param modelLoader A Python lambda function for a model loader class object.
Review Comment:
Done.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] TheNeuralBit commented on a diff in pull request #21809: [BEAM-14506] Adding testcases and examples for xlang Python RunInference
Posted by GitBox <gi...@apache.org>.
TheNeuralBit commented on code in PR #21809:
URL: https://github.com/apache/beam/pull/21809#discussion_r917160671
##########
sdks/java/extensions/python/src/main/java/org/apache/beam/sdk/extensions/python/transforms/RunInference.java:
##########
@@ -0,0 +1,102 @@
+/*
+ * 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.beam.sdk.extensions.python.transforms;
+
+import java.util.Map;
+import org.apache.beam.sdk.coders.RowCoder;
+import org.apache.beam.sdk.extensions.python.PythonExternalTransform;
+import org.apache.beam.sdk.schemas.Schema;
+import org.apache.beam.sdk.transforms.PTransform;
+import org.apache.beam.sdk.util.PythonCallableSource;
+import org.apache.beam.sdk.values.PCollection;
+import org.apache.beam.sdk.values.Row;
+import org.apache.beam.vendor.guava.v26_0_jre.com.google.common.collect.ImmutableMap;
+
+/** Wrapper for invoking external Python RunInference. */
+public class RunInference extends PTransform<PCollection<?>, PCollection<Row>> {
+ private final String modelLoader;
+ private final Schema schema;
+ private final Map<String, Object> kwargs;
+ private final String expansionService;
+
+ /**
+ * Instantiates a multi-language wrapper for a Python RunInference with a given model loader.
+ *
+ * @param modelLoader A Python lambda function for a model loader class object.
+ * @param exampleType A schema field type for the example column in output rows.
+ * @param inferenceType A schema field type for the inference column in output rows.
+ * @return A {@link RunInference} for the given model loader.
+ */
+ public static RunInference of(
+ String modelLoader, Schema.FieldType exampleType, Schema.FieldType inferenceType) {
+ Schema schema =
+ Schema.of(
+ Schema.Field.of("example", exampleType), Schema.Field.of("inference", inferenceType));
+ return new RunInference(modelLoader, schema, ImmutableMap.of(), "");
+ }
+
+ /**
+ * Instantiates a multi-language wrapper for a Python RunInference with a given model loader.
+ *
+ * @param modelLoader A Python lambda function for a model loader class object.
+ * @param schema A schema for output rows.
+ * @return A {@link RunInference} for the given model loader.
+ */
+ public static RunInference of(String modelLoader, Schema schema) {
+ return new RunInference(modelLoader, schema, ImmutableMap.of(), "");
+ }
+
+ /**
+ * Sets keyword arguments for RunInference constructor.
+ *
+ * @return A {@link RunInference} with keyword arguments.
+ */
+ public RunInference withKwarg(String key, Object arg) {
+ ImmutableMap.Builder<String, Object> builder =
+ ImmutableMap.<String, Object>builder().putAll(kwargs).put(key, arg);
+ return new RunInference(modelLoader, schema, builder.build(), expansionService);
+ }
+
+ /**
+ * Sets an expansion service endpoint for RunInference.
+ *
+ * @param expansionService A URL for a Python expansion service.
+ * @return A {@link RunInference} for the given expansion service endpoint.
+ */
+ public RunInference withExpansionService(String expansionService) {
+ return new RunInference(modelLoader, schema, kwargs, expansionService);
+ }
+
+ private RunInference(
+ String modelLoader, Schema schema, Map<String, Object> kwargs, String expansionService) {
+ this.modelLoader = modelLoader;
+ this.schema = schema;
+ this.kwargs = kwargs;
+ this.expansionService = expansionService;
+ }
+
+ @Override
+ public PCollection<Row> expand(PCollection<?> input) {
+ return input.apply(
+ PythonExternalTransform.<PCollection<?>, PCollection<Row>>from(
+ "apache_beam.ml.inference.base.RunInference.create", expansionService)
+ .withKwarg("model_handler_provider", PythonCallableSource.of(modelLoader))
+ .withKwargs(kwargs)
+ .withOutputCoder(RowCoder.of(schema)));
Review Comment:
> for cross language this information must be provided explicitly
Sure, but shouldn't it be the responsibility of the expanding SDK to figure this out? Not the calling SDK (and potentially the user)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] robertwb commented on a diff in pull request #21809: [BEAM-14506] Adding testcases and examples for xlang Python RunInference
Posted by GitBox <gi...@apache.org>.
robertwb commented on code in PR #21809:
URL: https://github.com/apache/beam/pull/21809#discussion_r918458856
##########
sdks/java/extensions/python/src/main/java/org/apache/beam/sdk/extensions/python/transforms/RunInference.java:
##########
@@ -0,0 +1,102 @@
+/*
+ * 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.beam.sdk.extensions.python.transforms;
+
+import java.util.Map;
+import org.apache.beam.sdk.coders.RowCoder;
+import org.apache.beam.sdk.extensions.python.PythonExternalTransform;
+import org.apache.beam.sdk.schemas.Schema;
+import org.apache.beam.sdk.transforms.PTransform;
+import org.apache.beam.sdk.util.PythonCallableSource;
+import org.apache.beam.sdk.values.PCollection;
+import org.apache.beam.sdk.values.Row;
+import org.apache.beam.vendor.guava.v26_0_jre.com.google.common.collect.ImmutableMap;
+
+/** Wrapper for invoking external Python RunInference. */
+public class RunInference extends PTransform<PCollection<?>, PCollection<Row>> {
+ private final String modelLoader;
+ private final Schema schema;
+ private final Map<String, Object> kwargs;
+ private final String expansionService;
+
+ /**
+ * Instantiates a multi-language wrapper for a Python RunInference with a given model loader.
+ *
+ * @param modelLoader A Python lambda function for a model loader class object.
+ * @param exampleType A schema field type for the example column in output rows.
+ * @param inferenceType A schema field type for the inference column in output rows.
+ * @return A {@link RunInference} for the given model loader.
+ */
+ public static RunInference of(
+ String modelLoader, Schema.FieldType exampleType, Schema.FieldType inferenceType) {
+ Schema schema =
+ Schema.of(
+ Schema.Field.of("example", exampleType), Schema.Field.of("inference", inferenceType));
+ return new RunInference(modelLoader, schema, ImmutableMap.of(), "");
+ }
+
+ /**
+ * Instantiates a multi-language wrapper for a Python RunInference with a given model loader.
+ *
+ * @param modelLoader A Python lambda function for a model loader class object.
+ * @param schema A schema for output rows.
+ * @return A {@link RunInference} for the given model loader.
+ */
+ public static RunInference of(String modelLoader, Schema schema) {
+ return new RunInference(modelLoader, schema, ImmutableMap.of(), "");
+ }
+
+ /**
+ * Sets keyword arguments for RunInference constructor.
+ *
+ * @return A {@link RunInference} with keyword arguments.
+ */
+ public RunInference withKwarg(String key, Object arg) {
+ ImmutableMap.Builder<String, Object> builder =
+ ImmutableMap.<String, Object>builder().putAll(kwargs).put(key, arg);
+ return new RunInference(modelLoader, schema, builder.build(), expansionService);
+ }
+
+ /**
+ * Sets an expansion service endpoint for RunInference.
+ *
+ * @param expansionService A URL for a Python expansion service.
+ * @return A {@link RunInference} for the given expansion service endpoint.
+ */
+ public RunInference withExpansionService(String expansionService) {
+ return new RunInference(modelLoader, schema, kwargs, expansionService);
+ }
+
+ private RunInference(
+ String modelLoader, Schema schema, Map<String, Object> kwargs, String expansionService) {
+ this.modelLoader = modelLoader;
+ this.schema = schema;
+ this.kwargs = kwargs;
+ this.expansionService = expansionService;
+ }
+
+ @Override
+ public PCollection<Row> expand(PCollection<?> input) {
+ return input.apply(
+ PythonExternalTransform.<PCollection<?>, PCollection<Row>>from(
+ "apache_beam.ml.inference.base.RunInference.create", expansionService)
+ .withKwarg("model_handler_provider", PythonCallableSource.of(modelLoader))
+ .withKwargs(kwargs)
+ .withOutputCoder(RowCoder.of(schema)));
Review Comment:
The SDK (in this case) figures out it's "Python Object." This essentially says "give it back to me as a a string [or fail]."
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] ihji commented on a diff in pull request #21809: [BEAM-14506] Adding testcases and examples for xlang Python RunInference
Posted by GitBox <gi...@apache.org>.
ihji commented on code in PR #21809:
URL: https://github.com/apache/beam/pull/21809#discussion_r918464465
##########
buildSrc/src/main/groovy/org/apache/beam/gradle/BeamModulePlugin.groovy:
##########
@@ -351,6 +351,8 @@ class BeamModulePlugin implements Plugin<Project> {
Integer numParallelTests = 1
// Whether the pipeline needs --sdk_location option
boolean needsSdkLocation = false
+ // semi_persist_dir for SDK containers
+ String semiPersistDir = "/tmp"
Review Comment:
I assume it's safe during the lifetime of the container. We use the staging location in `/tmp` by default for publicly built containers.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] ihji merged pull request #21809: [BEAM-14506] Adding testcases and examples for xlang Python RunInference
Posted by GitBox <gi...@apache.org>.
ihji merged PR #21809:
URL: https://github.com/apache/beam/pull/21809
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] ihji commented on pull request #21809: [BEAM-14506] Adding testcases and examples for xlang Python RunInference
Posted by GitBox <gi...@apache.org>.
ihji commented on PR #21809:
URL: https://github.com/apache/beam/pull/21809#issuecomment-1182146940
Run Portable_Python PreCommit
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] ihji commented on pull request #21809: [BEAM-14506] Adding testcases and examples for xlang Python RunInference
Posted by GitBox <gi...@apache.org>.
ihji commented on PR #21809:
URL: https://github.com/apache/beam/pull/21809#issuecomment-1182508492
Run Python PreCommit
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] pabloem commented on pull request #21809: [WIP/BEAM-14506] Adding testcases and examples for xlang Python RunInference
Posted by GitBox <gi...@apache.org>.
pabloem commented on PR #21809:
URL: https://github.com/apache/beam/pull/21809#issuecomment-1156980456
Run SQL PreCommit
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] ihji commented on pull request #21809: [BEAM-14506] Adding testcases and examples for xlang Python RunInference
Posted by GitBox <gi...@apache.org>.
ihji commented on PR #21809:
URL: https://github.com/apache/beam/pull/21809#issuecomment-1182147968
Run PythonDocs PreCommit
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] ihji commented on pull request #21809: [BEAM-14506] Adding testcases and examples for xlang Python RunInference
Posted by GitBox <gi...@apache.org>.
ihji commented on PR #21809:
URL: https://github.com/apache/beam/pull/21809#issuecomment-1182147513
Run Python PreCommit
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] ihji commented on pull request #21809: [BEAM-14506] Adding testcases and examples for xlang Python RunInference
Posted by GitBox <gi...@apache.org>.
ihji commented on PR #21809:
URL: https://github.com/apache/beam/pull/21809#issuecomment-1182146480
Run Java PreCommit
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] ihji commented on pull request #21809: [BEAM-14506] Adding testcases and examples for xlang Python RunInference
Posted by GitBox <gi...@apache.org>.
ihji commented on PR #21809:
URL: https://github.com/apache/beam/pull/21809#issuecomment-1165889934
R: @robertwb
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] ihji commented on pull request #21809: [BEAM-14506] Adding testcases and examples for xlang Python RunInference
Posted by GitBox <gi...@apache.org>.
ihji commented on PR #21809:
URL: https://github.com/apache/beam/pull/21809#issuecomment-1166068275
Run Python PreCommit
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] ihji commented on pull request #21809: [BEAM-14506] Adding testcases and examples for xlang Python RunInference
Posted by GitBox <gi...@apache.org>.
ihji commented on PR #21809:
URL: https://github.com/apache/beam/pull/21809#issuecomment-1165766244
Run Java PreCommit
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] pabloem commented on pull request #21809: [WIP/BEAM-14506] Adding testcases and examples for xlang Python RunInference
Posted by GitBox <gi...@apache.org>.
pabloem commented on PR #21809:
URL: https://github.com/apache/beam/pull/21809#issuecomment-1156979697
Run Java PostCommit
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] ihji commented on pull request #21809: [BEAM-14506] Adding testcases and examples for xlang Python RunInference
Posted by GitBox <gi...@apache.org>.
ihji commented on PR #21809:
URL: https://github.com/apache/beam/pull/21809#issuecomment-1183567940
Run Portable_Python PreCommit
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] ihji commented on pull request #21809: [BEAM-14506] Adding testcases and examples for xlang Python RunInference
Posted by GitBox <gi...@apache.org>.
ihji commented on PR #21809:
URL: https://github.com/apache/beam/pull/21809#issuecomment-1183577654
Run PythonDocs PreCommit
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] ihji commented on a diff in pull request #21809: [BEAM-14506] Adding testcases and examples for xlang Python RunInference
Posted by GitBox <gi...@apache.org>.
ihji commented on code in PR #21809:
URL: https://github.com/apache/beam/pull/21809#discussion_r918464465
##########
buildSrc/src/main/groovy/org/apache/beam/gradle/BeamModulePlugin.groovy:
##########
@@ -351,6 +351,8 @@ class BeamModulePlugin implements Plugin<Project> {
Integer numParallelTests = 1
// Whether the pipeline needs --sdk_location option
boolean needsSdkLocation = false
+ // semi_persist_dir for SDK containers
+ String semiPersistDir = "/tmp"
Review Comment:
I assume it's safe during the lifetime of the container. We use the staging location in `/tmp` by default for open-source runners.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] ihji commented on pull request #21809: [BEAM-14506] Adding testcases and examples for xlang Python RunInference
Posted by GitBox <gi...@apache.org>.
ihji commented on PR #21809:
URL: https://github.com/apache/beam/pull/21809#issuecomment-1163930865
Run XVR_Direct PostCommit
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] github-actions[bot] commented on pull request #21809: [BEAM-14506] Adding testcases and examples for xlang Python RunInference
Posted by GitBox <gi...@apache.org>.
github-actions[bot] commented on PR #21809:
URL: https://github.com/apache/beam/pull/21809#issuecomment-1165890554
Stopping reviewer notifications for this pull request: review requested by someone other than the bot, ceding control
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] TheNeuralBit commented on a diff in pull request #21809: [BEAM-14506] Adding testcases and examples for xlang Python RunInference
Posted by GitBox <gi...@apache.org>.
TheNeuralBit commented on code in PR #21809:
URL: https://github.com/apache/beam/pull/21809#discussion_r912236727
##########
sdks/java/extensions/python/src/main/java/org/apache/beam/sdk/extensions/python/transforms/RunInference.java:
##########
@@ -0,0 +1,102 @@
+/*
+ * 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.beam.sdk.extensions.python.transforms;
+
+import java.util.Map;
+import org.apache.beam.sdk.coders.RowCoder;
+import org.apache.beam.sdk.extensions.python.PythonExternalTransform;
+import org.apache.beam.sdk.schemas.Schema;
+import org.apache.beam.sdk.transforms.PTransform;
+import org.apache.beam.sdk.util.PythonCallableSource;
+import org.apache.beam.sdk.values.PCollection;
+import org.apache.beam.sdk.values.Row;
+import org.apache.beam.vendor.guava.v26_0_jre.com.google.common.collect.ImmutableMap;
+
+/** Wrapper for invoking external Python RunInference. */
+public class RunInference extends PTransform<PCollection<?>, PCollection<Row>> {
+ private final String modelLoader;
+ private final Schema schema;
+ private final Map<String, Object> kwargs;
+ private final String expansionService;
+
+ /**
+ * Instantiates a multi-language wrapper for a Python RunInference with a given model loader.
+ *
+ * @param modelLoader A Python lambda function for a model loader class object.
+ * @param exampleType A schema field type for the example column in output rows.
+ * @param inferenceType A schema field type for the inference column in output rows.
+ * @return A {@link RunInference} for the given model loader.
+ */
+ public static RunInference of(
+ String modelLoader, Schema.FieldType exampleType, Schema.FieldType inferenceType) {
+ Schema schema =
+ Schema.of(
+ Schema.Field.of("example", exampleType), Schema.Field.of("inference", inferenceType));
+ return new RunInference(modelLoader, schema, ImmutableMap.of(), "");
+ }
+
+ /**
+ * Instantiates a multi-language wrapper for a Python RunInference with a given model loader.
+ *
+ * @param modelLoader A Python lambda function for a model loader class object.
+ * @param schema A schema for output rows.
+ * @return A {@link RunInference} for the given model loader.
+ */
+ public static RunInference of(String modelLoader, Schema schema) {
+ return new RunInference(modelLoader, schema, ImmutableMap.of(), "");
+ }
+
+ /**
+ * Sets keyword arguments for RunInference constructor.
+ *
+ * @return A {@link RunInference} with keyword arguments.
+ */
+ public RunInference withKwarg(String key, Object arg) {
+ ImmutableMap.Builder<String, Object> builder =
+ ImmutableMap.<String, Object>builder().putAll(kwargs).put(key, arg);
+ return new RunInference(modelLoader, schema, builder.build(), expansionService);
+ }
+
+ /**
+ * Sets an expansion service endpoint for RunInference.
+ *
+ * @param expansionService A URL for a Python expansion service.
+ * @return A {@link RunInference} for the given expansion service endpoint.
+ */
+ public RunInference withExpansionService(String expansionService) {
+ return new RunInference(modelLoader, schema, kwargs, expansionService);
+ }
+
+ private RunInference(
+ String modelLoader, Schema schema, Map<String, Object> kwargs, String expansionService) {
+ this.modelLoader = modelLoader;
+ this.schema = schema;
+ this.kwargs = kwargs;
+ this.expansionService = expansionService;
+ }
+
+ @Override
+ public PCollection<Row> expand(PCollection<?> input) {
+ return input.apply(
+ PythonExternalTransform.<PCollection<?>, PCollection<Row>>from(
+ "apache_beam.ml.inference.base.RunInference.create", expansionService)
+ .withKwarg("model_handler_provider", PythonCallableSource.of(modelLoader))
+ .withKwargs(kwargs)
+ .withOutputCoder(RowCoder.of(schema)));
Review Comment:
I see. Could we then record the instances where the Python SDK is falling short? It would be nice if there was a path toward deprecating and removing this workaround.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] ihji commented on pull request #21809: [BEAM-14506] Adding testcases and examples for xlang Python RunInference
Posted by GitBox <gi...@apache.org>.
ihji commented on PR #21809:
URL: https://github.com/apache/beam/pull/21809#issuecomment-1182508083
Run Portable_Python PreCommit
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] ihji commented on pull request #21809: [BEAM-14506] Adding testcases and examples for xlang Python RunInference
Posted by GitBox <gi...@apache.org>.
ihji commented on PR #21809:
URL: https://github.com/apache/beam/pull/21809#issuecomment-1183577417
Run Python PreCommit
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] ihji commented on a diff in pull request #21809: [BEAM-14506] Adding testcases and examples for xlang Python RunInference
Posted by GitBox <gi...@apache.org>.
ihji commented on code in PR #21809:
URL: https://github.com/apache/beam/pull/21809#discussion_r912223535
##########
sdks/java/extensions/python/src/main/java/org/apache/beam/sdk/extensions/python/transforms/RunInference.java:
##########
@@ -0,0 +1,102 @@
+/*
+ * 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.beam.sdk.extensions.python.transforms;
+
+import java.util.Map;
+import org.apache.beam.sdk.coders.RowCoder;
+import org.apache.beam.sdk.extensions.python.PythonExternalTransform;
+import org.apache.beam.sdk.schemas.Schema;
+import org.apache.beam.sdk.transforms.PTransform;
+import org.apache.beam.sdk.util.PythonCallableSource;
+import org.apache.beam.sdk.values.PCollection;
+import org.apache.beam.sdk.values.Row;
+import org.apache.beam.vendor.guava.v26_0_jre.com.google.common.collect.ImmutableMap;
+
+/** Wrapper for invoking external Python RunInference. */
+public class RunInference extends PTransform<PCollection<?>, PCollection<Row>> {
+ private final String modelLoader;
+ private final Schema schema;
+ private final Map<String, Object> kwargs;
+ private final String expansionService;
+
+ /**
+ * Instantiates a multi-language wrapper for a Python RunInference with a given model loader.
+ *
+ * @param modelLoader A Python lambda function for a model loader class object.
+ * @param exampleType A schema field type for the example column in output rows.
+ * @param inferenceType A schema field type for the inference column in output rows.
+ * @return A {@link RunInference} for the given model loader.
+ */
+ public static RunInference of(
+ String modelLoader, Schema.FieldType exampleType, Schema.FieldType inferenceType) {
+ Schema schema =
+ Schema.of(
+ Schema.Field.of("example", exampleType), Schema.Field.of("inference", inferenceType));
+ return new RunInference(modelLoader, schema, ImmutableMap.of(), "");
+ }
+
+ /**
+ * Instantiates a multi-language wrapper for a Python RunInference with a given model loader.
+ *
+ * @param modelLoader A Python lambda function for a model loader class object.
+ * @param schema A schema for output rows.
+ * @return A {@link RunInference} for the given model loader.
+ */
+ public static RunInference of(String modelLoader, Schema schema) {
+ return new RunInference(modelLoader, schema, ImmutableMap.of(), "");
+ }
+
+ /**
+ * Sets keyword arguments for RunInference constructor.
+ *
+ * @return A {@link RunInference} with keyword arguments.
+ */
+ public RunInference withKwarg(String key, Object arg) {
+ ImmutableMap.Builder<String, Object> builder =
+ ImmutableMap.<String, Object>builder().putAll(kwargs).put(key, arg);
+ return new RunInference(modelLoader, schema, builder.build(), expansionService);
+ }
+
+ /**
+ * Sets an expansion service endpoint for RunInference.
+ *
+ * @param expansionService A URL for a Python expansion service.
+ * @return A {@link RunInference} for the given expansion service endpoint.
+ */
+ public RunInference withExpansionService(String expansionService) {
+ return new RunInference(modelLoader, schema, kwargs, expansionService);
+ }
+
+ private RunInference(
+ String modelLoader, Schema schema, Map<String, Object> kwargs, String expansionService) {
+ this.modelLoader = modelLoader;
+ this.schema = schema;
+ this.kwargs = kwargs;
+ this.expansionService = expansionService;
+ }
+
+ @Override
+ public PCollection<Row> expand(PCollection<?> input) {
+ return input.apply(
+ PythonExternalTransform.<PCollection<?>, PCollection<Row>>from(
+ "apache_beam.ml.inference.base.RunInference.create", expansionService)
+ .withKwarg("model_handler_provider", PythonCallableSource.of(modelLoader))
+ .withKwargs(kwargs)
+ .withOutputCoder(RowCoder.of(schema)));
Review Comment:
Yes, ideally the Python SDK should be all responsible for inferring output coders but in reality there are many instances that the type inference doesn’t work as expected. Users may want to use `withOutputCoder` method for giving a hint to the Python SDK (it’s optional and not guaranteed to be satisfied).
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] ihji commented on pull request #21809: [BEAM-14506] Adding testcases and examples for xlang Python RunInference
Posted by GitBox <gi...@apache.org>.
ihji commented on PR #21809:
URL: https://github.com/apache/beam/pull/21809#issuecomment-1163931351
Run XVR_Direct PostCommit
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] ihji commented on pull request #21809: [BEAM-14506] Adding testcases and examples for xlang Python RunInference
Posted by GitBox <gi...@apache.org>.
ihji commented on PR #21809:
URL: https://github.com/apache/beam/pull/21809#issuecomment-1164910866
https://ci-beam.apache.org/job/beam_PostCommit_XVR_Direct_PR/126
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] ihji commented on pull request #21809: [BEAM-14506] Adding testcases and examples for xlang Python RunInference
Posted by GitBox <gi...@apache.org>.
ihji commented on PR #21809:
URL: https://github.com/apache/beam/pull/21809#issuecomment-1166068139
Run SQL PreCommit
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] ihji commented on a diff in pull request #21809: [BEAM-14506] Adding testcases and examples for xlang Python RunInference
Posted by GitBox <gi...@apache.org>.
ihji commented on code in PR #21809:
URL: https://github.com/apache/beam/pull/21809#discussion_r918463345
##########
sdks/java/extensions/python/src/test/java/org/apache/beam/sdk/extensions/python/transforms/RunInferenceTransformTest.java:
##########
@@ -0,0 +1,68 @@
+/*
+ * 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.beam.sdk.extensions.python.transforms;
+
+import java.util.Arrays;
+import java.util.Optional;
+import org.apache.beam.runners.core.construction.BaseExternalTest;
+import org.apache.beam.sdk.coders.IterableCoder;
+import org.apache.beam.sdk.coders.VarLongCoder;
+import org.apache.beam.sdk.schemas.Schema;
+import org.apache.beam.sdk.testing.PAssert;
+import org.apache.beam.sdk.testing.UsesPythonExpansionService;
+import org.apache.beam.sdk.testing.ValidatesRunner;
+import org.apache.beam.sdk.transforms.Create;
+import org.apache.beam.sdk.values.PCollection;
+import org.apache.beam.sdk.values.Row;
+import org.junit.Test;
+import org.junit.experimental.categories.Category;
+import org.junit.runner.RunWith;
+import org.junit.runners.JUnit4;
+
+@RunWith(JUnit4.class)
+public class RunInferenceTransformTest extends BaseExternalTest {
+ @Test
+ @Category({ValidatesRunner.class, UsesPythonExpansionService.class})
+ public void testRunInference() {
+ String stagingLocation =
+ Optional.ofNullable(System.getProperty("semiPersistDir")).orElse("/tmp");
+ Schema schema =
+ Schema.of(
+ Schema.Field.of("example", Schema.FieldType.array(Schema.FieldType.INT64)),
+ Schema.Field.of("inference", Schema.FieldType.INT32));
+ Row row0 = Row.withSchema(schema).addArray(0L, 0L).addValue(0).build();
+ Row row1 = Row.withSchema(schema).addArray(1L, 1L).addValue(1).build();
+ PCollection<Row> col =
+ testPipeline
+ .apply(Create.<Iterable<Long>>of(Arrays.asList(0L, 0L), Arrays.asList(1L, 1L)))
+ .setCoder(IterableCoder.of(VarLongCoder.of()))
+ .apply(
+ RunInference.of(
+ "apache_beam.ml.inference.sklearn_inference.SklearnModelHandlerNumpy",
+ schema)
+ .withKwarg(
+ // The test expansion service creates the test model and saves it to the
+ // returning external environment as a dependency.
+ // (sdks/python/apache_beam/runners/portability/expansion_service_test.py)
+ // The dependencies for Python SDK harness are supposed to be staged to
Review Comment:
Not sure if it's documented somewhere but this location is hard-coded in the container boot file. The path specified with stage_to role will be joined with this hard-coded base path when the container downloads the artifacts.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] pabloem commented on pull request #21809: [WIP/BEAM-14506] Adding testcases and examples for xlang Python RunInference
Posted by GitBox <gi...@apache.org>.
pabloem commented on PR #21809:
URL: https://github.com/apache/beam/pull/21809#issuecomment-1156980207
Run Java PreCommit
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] ihji commented on pull request #21809: [BEAM-14506] Adding testcases and examples for xlang Python RunInference
Posted by GitBox <gi...@apache.org>.
ihji commented on PR #21809:
URL: https://github.com/apache/beam/pull/21809#issuecomment-1164109347
Run XVR_Direct PostCommit
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] ihji commented on pull request #21809: [BEAM-14506] Adding testcases and examples for xlang Python RunInference
Posted by GitBox <gi...@apache.org>.
ihji commented on PR #21809:
URL: https://github.com/apache/beam/pull/21809#issuecomment-1166068377
Run Java PreCommit
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] ihji commented on pull request #21809: [BEAM-14506] Adding testcases and examples for xlang Python RunInference
Posted by GitBox <gi...@apache.org>.
ihji commented on PR #21809:
URL: https://github.com/apache/beam/pull/21809#issuecomment-1183567596
Run Portable_Python PreCommit
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] ihji commented on pull request #21809: [BEAM-14506] Adding testcases and examples for xlang Python RunInference
Posted by GitBox <gi...@apache.org>.
ihji commented on PR #21809:
URL: https://github.com/apache/beam/pull/21809#issuecomment-1183568677
Run Portable_Python PreCommit
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] TheNeuralBit commented on a diff in pull request #21809: [BEAM-14506] Adding testcases and examples for xlang Python RunInference
Posted by GitBox <gi...@apache.org>.
TheNeuralBit commented on code in PR #21809:
URL: https://github.com/apache/beam/pull/21809#discussion_r912236727
##########
sdks/java/extensions/python/src/main/java/org/apache/beam/sdk/extensions/python/transforms/RunInference.java:
##########
@@ -0,0 +1,102 @@
+/*
+ * 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.beam.sdk.extensions.python.transforms;
+
+import java.util.Map;
+import org.apache.beam.sdk.coders.RowCoder;
+import org.apache.beam.sdk.extensions.python.PythonExternalTransform;
+import org.apache.beam.sdk.schemas.Schema;
+import org.apache.beam.sdk.transforms.PTransform;
+import org.apache.beam.sdk.util.PythonCallableSource;
+import org.apache.beam.sdk.values.PCollection;
+import org.apache.beam.sdk.values.Row;
+import org.apache.beam.vendor.guava.v26_0_jre.com.google.common.collect.ImmutableMap;
+
+/** Wrapper for invoking external Python RunInference. */
+public class RunInference extends PTransform<PCollection<?>, PCollection<Row>> {
+ private final String modelLoader;
+ private final Schema schema;
+ private final Map<String, Object> kwargs;
+ private final String expansionService;
+
+ /**
+ * Instantiates a multi-language wrapper for a Python RunInference with a given model loader.
+ *
+ * @param modelLoader A Python lambda function for a model loader class object.
+ * @param exampleType A schema field type for the example column in output rows.
+ * @param inferenceType A schema field type for the inference column in output rows.
+ * @return A {@link RunInference} for the given model loader.
+ */
+ public static RunInference of(
+ String modelLoader, Schema.FieldType exampleType, Schema.FieldType inferenceType) {
+ Schema schema =
+ Schema.of(
+ Schema.Field.of("example", exampleType), Schema.Field.of("inference", inferenceType));
+ return new RunInference(modelLoader, schema, ImmutableMap.of(), "");
+ }
+
+ /**
+ * Instantiates a multi-language wrapper for a Python RunInference with a given model loader.
+ *
+ * @param modelLoader A Python lambda function for a model loader class object.
+ * @param schema A schema for output rows.
+ * @return A {@link RunInference} for the given model loader.
+ */
+ public static RunInference of(String modelLoader, Schema schema) {
+ return new RunInference(modelLoader, schema, ImmutableMap.of(), "");
+ }
+
+ /**
+ * Sets keyword arguments for RunInference constructor.
+ *
+ * @return A {@link RunInference} with keyword arguments.
+ */
+ public RunInference withKwarg(String key, Object arg) {
+ ImmutableMap.Builder<String, Object> builder =
+ ImmutableMap.<String, Object>builder().putAll(kwargs).put(key, arg);
+ return new RunInference(modelLoader, schema, builder.build(), expansionService);
+ }
+
+ /**
+ * Sets an expansion service endpoint for RunInference.
+ *
+ * @param expansionService A URL for a Python expansion service.
+ * @return A {@link RunInference} for the given expansion service endpoint.
+ */
+ public RunInference withExpansionService(String expansionService) {
+ return new RunInference(modelLoader, schema, kwargs, expansionService);
+ }
+
+ private RunInference(
+ String modelLoader, Schema schema, Map<String, Object> kwargs, String expansionService) {
+ this.modelLoader = modelLoader;
+ this.schema = schema;
+ this.kwargs = kwargs;
+ this.expansionService = expansionService;
+ }
+
+ @Override
+ public PCollection<Row> expand(PCollection<?> input) {
+ return input.apply(
+ PythonExternalTransform.<PCollection<?>, PCollection<Row>>from(
+ "apache_beam.ml.inference.base.RunInference.create", expansionService)
+ .withKwarg("model_handler_provider", PythonCallableSource.of(modelLoader))
+ .withKwargs(kwargs)
+ .withOutputCoder(RowCoder.of(schema)));
Review Comment:
I see. Could we then record the instances where the Python SDK is falling short in issues? It would be nice if there was a path toward deprecating and removing this workaround.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] ihji commented on a diff in pull request #21809: [BEAM-14506] Adding testcases and examples for xlang Python RunInference
Posted by GitBox <gi...@apache.org>.
ihji commented on code in PR #21809:
URL: https://github.com/apache/beam/pull/21809#discussion_r918465485
##########
runners/google-cloud-dataflow-java/build.gradle:
##########
@@ -413,6 +413,7 @@ createCrossLanguageValidatesRunnerTask(
classpath: configurations.validatesRunner,
numParallelTests: Integer.MAX_VALUE,
needsSdkLocation: true,
+ semiPersistDir: "/var/opt/google",
Review Comment:
We use different semi persistant dirs for open-source runners and Dataflow runner (don't know why).
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] ihji commented on pull request #21809: [BEAM-14506] Adding testcases and examples for xlang Python RunInference
Posted by GitBox <gi...@apache.org>.
ihji commented on PR #21809:
URL: https://github.com/apache/beam/pull/21809#issuecomment-1165826911
Run Java PreCommit
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] ihji commented on pull request #21809: [BEAM-14506] Adding testcases and examples for xlang Python RunInference
Posted by GitBox <gi...@apache.org>.
ihji commented on PR #21809:
URL: https://github.com/apache/beam/pull/21809#issuecomment-1166168209
Run SQL PreCommit
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] tvalentyn commented on a diff in pull request #21809: [BEAM-14506] Adding testcases and examples for xlang Python RunInference
Posted by GitBox <gi...@apache.org>.
tvalentyn commented on code in PR #21809:
URL: https://github.com/apache/beam/pull/21809#discussion_r919279012
##########
runners/google-cloud-dataflow-java/build.gradle:
##########
@@ -413,6 +413,7 @@ createCrossLanguageValidatesRunnerTask(
classpath: configurations.validatesRunner,
numParallelTests: Integer.MAX_VALUE,
needsSdkLocation: true,
+ semiPersistDir: "/var/opt/google",
Review Comment:
container code defines default values. i think runners may still be overriding the default values when configuring container pods, so switching to external containers might not remediate this.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] ihji commented on pull request #21809: [BEAM-14506] Adding testcases and examples for xlang Python RunInference
Posted by GitBox <gi...@apache.org>.
ihji commented on PR #21809:
URL: https://github.com/apache/beam/pull/21809#issuecomment-1182507742
Run PythonDocs PreCommit
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] ihji commented on a diff in pull request #21809: [BEAM-14506] Adding testcases and examples for xlang Python RunInference
Posted by GitBox <gi...@apache.org>.
ihji commented on code in PR #21809:
URL: https://github.com/apache/beam/pull/21809#discussion_r918453159
##########
sdks/java/extensions/python/src/main/java/org/apache/beam/sdk/extensions/python/transforms/RunInference.java:
##########
@@ -0,0 +1,102 @@
+/*
+ * 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.beam.sdk.extensions.python.transforms;
+
+import java.util.Map;
+import org.apache.beam.sdk.coders.RowCoder;
+import org.apache.beam.sdk.extensions.python.PythonExternalTransform;
+import org.apache.beam.sdk.schemas.Schema;
+import org.apache.beam.sdk.transforms.PTransform;
+import org.apache.beam.sdk.util.PythonCallableSource;
+import org.apache.beam.sdk.values.PCollection;
+import org.apache.beam.sdk.values.Row;
+import org.apache.beam.vendor.guava.v26_0_jre.com.google.common.collect.ImmutableMap;
+
+/** Wrapper for invoking external Python RunInference. */
+public class RunInference extends PTransform<PCollection<?>, PCollection<Row>> {
+ private final String modelLoader;
+ private final Schema schema;
+ private final Map<String, Object> kwargs;
+ private final String expansionService;
+
+ /**
+ * Instantiates a multi-language wrapper for a Python RunInference with a given model loader.
+ *
+ * @param modelLoader A Python lambda function for a model loader class object.
+ * @param exampleType A schema field type for the example column in output rows.
+ * @param inferenceType A schema field type for the inference column in output rows.
+ * @return A {@link RunInference} for the given model loader.
+ */
+ public static RunInference of(
+ String modelLoader, Schema.FieldType exampleType, Schema.FieldType inferenceType) {
+ Schema schema =
+ Schema.of(
+ Schema.Field.of("example", exampleType), Schema.Field.of("inference", inferenceType));
+ return new RunInference(modelLoader, schema, ImmutableMap.of(), "");
+ }
+
+ /**
+ * Instantiates a multi-language wrapper for a Python RunInference with a given model loader.
+ *
+ * @param modelLoader A Python lambda function for a model loader class object.
+ * @param schema A schema for output rows.
+ * @return A {@link RunInference} for the given model loader.
+ */
+ public static RunInference of(String modelLoader, Schema schema) {
+ return new RunInference(modelLoader, schema, ImmutableMap.of(), "");
+ }
+
+ /**
+ * Sets keyword arguments for RunInference constructor.
Review Comment:
Done.
##########
sdks/java/extensions/python/src/main/java/org/apache/beam/sdk/extensions/python/transforms/RunInference.java:
##########
@@ -0,0 +1,102 @@
+/*
+ * 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.beam.sdk.extensions.python.transforms;
+
+import java.util.Map;
+import org.apache.beam.sdk.coders.RowCoder;
+import org.apache.beam.sdk.extensions.python.PythonExternalTransform;
+import org.apache.beam.sdk.schemas.Schema;
+import org.apache.beam.sdk.transforms.PTransform;
+import org.apache.beam.sdk.util.PythonCallableSource;
+import org.apache.beam.sdk.values.PCollection;
+import org.apache.beam.sdk.values.Row;
+import org.apache.beam.vendor.guava.v26_0_jre.com.google.common.collect.ImmutableMap;
+
+/** Wrapper for invoking external Python RunInference. */
+public class RunInference extends PTransform<PCollection<?>, PCollection<Row>> {
+ private final String modelLoader;
+ private final Schema schema;
+ private final Map<String, Object> kwargs;
+ private final String expansionService;
+
+ /**
+ * Instantiates a multi-language wrapper for a Python RunInference with a given model loader.
+ *
+ * @param modelLoader A Python lambda function for a model loader class object.
+ * @param exampleType A schema field type for the example column in output rows.
+ * @param inferenceType A schema field type for the inference column in output rows.
+ * @return A {@link RunInference} for the given model loader.
+ */
+ public static RunInference of(
+ String modelLoader, Schema.FieldType exampleType, Schema.FieldType inferenceType) {
+ Schema schema =
+ Schema.of(
+ Schema.Field.of("example", exampleType), Schema.Field.of("inference", inferenceType));
+ return new RunInference(modelLoader, schema, ImmutableMap.of(), "");
+ }
+
+ /**
+ * Instantiates a multi-language wrapper for a Python RunInference with a given model loader.
+ *
+ * @param modelLoader A Python lambda function for a model loader class object.
+ * @param schema A schema for output rows.
+ * @return A {@link RunInference} for the given model loader.
+ */
+ public static RunInference of(String modelLoader, Schema schema) {
+ return new RunInference(modelLoader, schema, ImmutableMap.of(), "");
+ }
+
+ /**
+ * Sets keyword arguments for RunInference constructor.
+ *
+ * @return A {@link RunInference} with keyword arguments.
+ */
+ public RunInference withKwarg(String key, Object arg) {
+ ImmutableMap.Builder<String, Object> builder =
+ ImmutableMap.<String, Object>builder().putAll(kwargs).put(key, arg);
+ return new RunInference(modelLoader, schema, builder.build(), expansionService);
+ }
+
+ /**
+ * Sets an expansion service endpoint for RunInference.
+ *
+ * @param expansionService A URL for a Python expansion service.
+ * @return A {@link RunInference} for the given expansion service endpoint.
+ */
+ public RunInference withExpansionService(String expansionService) {
+ return new RunInference(modelLoader, schema, kwargs, expansionService);
+ }
+
+ private RunInference(
+ String modelLoader, Schema schema, Map<String, Object> kwargs, String expansionService) {
+ this.modelLoader = modelLoader;
+ this.schema = schema;
+ this.kwargs = kwargs;
+ this.expansionService = expansionService;
+ }
+
+ @Override
+ public PCollection<Row> expand(PCollection<?> input) {
+ return input.apply(
+ PythonExternalTransform.<PCollection<?>, PCollection<Row>>from(
+ "apache_beam.ml.inference.base.RunInference.create", expansionService)
Review Comment:
Done.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] robertwb commented on a diff in pull request #21809: [BEAM-14506] Adding testcases and examples for xlang Python RunInference
Posted by GitBox <gi...@apache.org>.
robertwb commented on code in PR #21809:
URL: https://github.com/apache/beam/pull/21809#discussion_r919223505
##########
sdks/java/extensions/python/src/test/java/org/apache/beam/sdk/extensions/python/transforms/RunInferenceTransformTest.java:
##########
@@ -0,0 +1,68 @@
+/*
+ * 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.beam.sdk.extensions.python.transforms;
+
+import java.util.Arrays;
+import java.util.Optional;
+import org.apache.beam.runners.core.construction.BaseExternalTest;
+import org.apache.beam.sdk.coders.IterableCoder;
+import org.apache.beam.sdk.coders.VarLongCoder;
+import org.apache.beam.sdk.schemas.Schema;
+import org.apache.beam.sdk.testing.PAssert;
+import org.apache.beam.sdk.testing.UsesPythonExpansionService;
+import org.apache.beam.sdk.testing.ValidatesRunner;
+import org.apache.beam.sdk.transforms.Create;
+import org.apache.beam.sdk.values.PCollection;
+import org.apache.beam.sdk.values.Row;
+import org.junit.Test;
+import org.junit.experimental.categories.Category;
+import org.junit.runner.RunWith;
+import org.junit.runners.JUnit4;
+
+@RunWith(JUnit4.class)
+public class RunInferenceTransformTest extends BaseExternalTest {
+ @Test
+ @Category({ValidatesRunner.class, UsesPythonExpansionService.class})
+ public void testRunInference() {
+ String stagingLocation =
+ Optional.ofNullable(System.getProperty("semiPersistDir")).orElse("/tmp");
+ Schema schema =
+ Schema.of(
+ Schema.Field.of("example", Schema.FieldType.array(Schema.FieldType.INT64)),
+ Schema.Field.of("inference", Schema.FieldType.INT32));
+ Row row0 = Row.withSchema(schema).addArray(0L, 0L).addValue(0).build();
+ Row row1 = Row.withSchema(schema).addArray(1L, 1L).addValue(1).build();
+ PCollection<Row> col =
+ testPipeline
+ .apply(Create.<Iterable<Long>>of(Arrays.asList(0L, 0L), Arrays.asList(1L, 1L)))
+ .setCoder(IterableCoder.of(VarLongCoder.of()))
+ .apply(
+ RunInference.of(
+ "apache_beam.ml.inference.sklearn_inference.SklearnModelHandlerNumpy",
+ schema)
+ .withKwarg(
+ // The test expansion service creates the test model and saves it to the
+ // returning external environment as a dependency.
+ // (sdks/python/apache_beam/runners/portability/expansion_service_test.py)
+ // The dependencies for Python SDK harness are supposed to be staged to
Review Comment:
Ack. This is good enough for a test, but we should probably do better if we expect users to use these mechanisms. (Alternatively, I suppose we could just stage the required files on GCS or similar.)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] robertwb commented on a diff in pull request #21809: [BEAM-14506] Adding testcases and examples for xlang Python RunInference
Posted by GitBox <gi...@apache.org>.
robertwb commented on code in PR #21809:
URL: https://github.com/apache/beam/pull/21809#discussion_r917155398
##########
sdks/java/extensions/python/src/main/java/org/apache/beam/sdk/extensions/python/transforms/RunInference.java:
##########
@@ -0,0 +1,102 @@
+/*
+ * 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.beam.sdk.extensions.python.transforms;
+
+import java.util.Map;
+import org.apache.beam.sdk.coders.RowCoder;
+import org.apache.beam.sdk.extensions.python.PythonExternalTransform;
+import org.apache.beam.sdk.schemas.Schema;
+import org.apache.beam.sdk.transforms.PTransform;
+import org.apache.beam.sdk.util.PythonCallableSource;
+import org.apache.beam.sdk.values.PCollection;
+import org.apache.beam.sdk.values.Row;
+import org.apache.beam.vendor.guava.v26_0_jre.com.google.common.collect.ImmutableMap;
+
+/** Wrapper for invoking external Python RunInference. */
+public class RunInference extends PTransform<PCollection<?>, PCollection<Row>> {
+ private final String modelLoader;
+ private final Schema schema;
+ private final Map<String, Object> kwargs;
+ private final String expansionService;
+
+ /**
+ * Instantiates a multi-language wrapper for a Python RunInference with a given model loader.
+ *
+ * @param modelLoader A Python lambda function for a model loader class object.
Review Comment:
s/lambda function/callable/ (and elsewhere)
##########
sdks/java/extensions/python/src/main/java/org/apache/beam/sdk/extensions/python/transforms/RunInference.java:
##########
@@ -0,0 +1,102 @@
+/*
+ * 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.beam.sdk.extensions.python.transforms;
+
+import java.util.Map;
+import org.apache.beam.sdk.coders.RowCoder;
+import org.apache.beam.sdk.extensions.python.PythonExternalTransform;
+import org.apache.beam.sdk.schemas.Schema;
+import org.apache.beam.sdk.transforms.PTransform;
+import org.apache.beam.sdk.util.PythonCallableSource;
+import org.apache.beam.sdk.values.PCollection;
+import org.apache.beam.sdk.values.Row;
+import org.apache.beam.vendor.guava.v26_0_jre.com.google.common.collect.ImmutableMap;
+
+/** Wrapper for invoking external Python RunInference. */
+public class RunInference extends PTransform<PCollection<?>, PCollection<Row>> {
+ private final String modelLoader;
+ private final Schema schema;
+ private final Map<String, Object> kwargs;
+ private final String expansionService;
+
+ /**
+ * Instantiates a multi-language wrapper for a Python RunInference with a given model loader.
+ *
+ * @param modelLoader A Python lambda function for a model loader class object.
+ * @param exampleType A schema field type for the example column in output rows.
+ * @param inferenceType A schema field type for the inference column in output rows.
+ * @return A {@link RunInference} for the given model loader.
+ */
+ public static RunInference of(
+ String modelLoader, Schema.FieldType exampleType, Schema.FieldType inferenceType) {
+ Schema schema =
+ Schema.of(
+ Schema.Field.of("example", exampleType), Schema.Field.of("inference", inferenceType));
+ return new RunInference(modelLoader, schema, ImmutableMap.of(), "");
+ }
+
+ /**
+ * Instantiates a multi-language wrapper for a Python RunInference with a given model loader.
+ *
+ * @param modelLoader A Python lambda function for a model loader class object.
+ * @param schema A schema for output rows.
+ * @return A {@link RunInference} for the given model loader.
+ */
+ public static RunInference of(String modelLoader, Schema schema) {
+ return new RunInference(modelLoader, schema, ImmutableMap.of(), "");
+ }
+
+ /**
+ * Sets keyword arguments for RunInference constructor.
+ *
+ * @return A {@link RunInference} with keyword arguments.
+ */
+ public RunInference withKwarg(String key, Object arg) {
+ ImmutableMap.Builder<String, Object> builder =
+ ImmutableMap.<String, Object>builder().putAll(kwargs).put(key, arg);
+ return new RunInference(modelLoader, schema, builder.build(), expansionService);
+ }
+
+ /**
+ * Sets an expansion service endpoint for RunInference.
+ *
+ * @param expansionService A URL for a Python expansion service.
+ * @return A {@link RunInference} for the given expansion service endpoint.
+ */
+ public RunInference withExpansionService(String expansionService) {
+ return new RunInference(modelLoader, schema, kwargs, expansionService);
+ }
+
+ private RunInference(
+ String modelLoader, Schema schema, Map<String, Object> kwargs, String expansionService) {
+ this.modelLoader = modelLoader;
+ this.schema = schema;
+ this.kwargs = kwargs;
+ this.expansionService = expansionService;
+ }
+
+ @Override
+ public PCollection<Row> expand(PCollection<?> input) {
+ return input.apply(
+ PythonExternalTransform.<PCollection<?>, PCollection<Row>>from(
+ "apache_beam.ml.inference.base.RunInference.create", expansionService)
Review Comment:
Generally "create" sounds like it's a synonym for the constructor. It should probably be renamed "from_callable" or something meaningful like that.
##########
buildSrc/src/main/groovy/org/apache/beam/gradle/BeamModulePlugin.groovy:
##########
@@ -2309,8 +2311,10 @@ class BeamModulePlugin implements Plugin<Project> {
throw new GradleException("unsupported java version.")
}
def setupTask = project.tasks.register(config.name+"Setup", Exec) {
- dependsOn ':sdks:java:container:'+javaContainerSuffix+':docker'
- dependsOn ':sdks:python:container:py'+pythonContainerSuffix+':docker'
+ if (!project.hasProperty('skipContainerBuilds')) {
Review Comment:
When would this be desirable? Seems dangerous to make tests non-hermetic.
##########
sdks/java/extensions/python/src/main/java/org/apache/beam/sdk/extensions/python/transforms/RunInference.java:
##########
@@ -0,0 +1,102 @@
+/*
+ * 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.beam.sdk.extensions.python.transforms;
+
+import java.util.Map;
+import org.apache.beam.sdk.coders.RowCoder;
+import org.apache.beam.sdk.extensions.python.PythonExternalTransform;
+import org.apache.beam.sdk.schemas.Schema;
+import org.apache.beam.sdk.transforms.PTransform;
+import org.apache.beam.sdk.util.PythonCallableSource;
+import org.apache.beam.sdk.values.PCollection;
+import org.apache.beam.sdk.values.Row;
+import org.apache.beam.vendor.guava.v26_0_jre.com.google.common.collect.ImmutableMap;
+
+/** Wrapper for invoking external Python RunInference. */
+public class RunInference extends PTransform<PCollection<?>, PCollection<Row>> {
+ private final String modelLoader;
+ private final Schema schema;
+ private final Map<String, Object> kwargs;
+ private final String expansionService;
+
+ /**
+ * Instantiates a multi-language wrapper for a Python RunInference with a given model loader.
+ *
+ * @param modelLoader A Python lambda function for a model loader class object.
+ * @param exampleType A schema field type for the example column in output rows.
+ * @param inferenceType A schema field type for the inference column in output rows.
+ * @return A {@link RunInference} for the given model loader.
+ */
+ public static RunInference of(
+ String modelLoader, Schema.FieldType exampleType, Schema.FieldType inferenceType) {
+ Schema schema =
+ Schema.of(
+ Schema.Field.of("example", exampleType), Schema.Field.of("inference", inferenceType));
+ return new RunInference(modelLoader, schema, ImmutableMap.of(), "");
+ }
+
+ /**
+ * Instantiates a multi-language wrapper for a Python RunInference with a given model loader.
+ *
+ * @param modelLoader A Python lambda function for a model loader class object.
+ * @param schema A schema for output rows.
+ * @return A {@link RunInference} for the given model loader.
+ */
+ public static RunInference of(String modelLoader, Schema schema) {
+ return new RunInference(modelLoader, schema, ImmutableMap.of(), "");
+ }
+
+ /**
+ * Sets keyword arguments for RunInference constructor.
Review Comment:
Sets keyword arguments for the model loader.
##########
sdks/java/extensions/python/src/test/java/org/apache/beam/sdk/extensions/python/transforms/RunInferenceTransformTest.java:
##########
@@ -0,0 +1,63 @@
+/*
+ * 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.beam.sdk.extensions.python.transforms;
+
+import java.util.Arrays;
+import java.util.Optional;
+import org.apache.beam.runners.core.construction.BaseExternalTest;
+import org.apache.beam.sdk.coders.IterableCoder;
+import org.apache.beam.sdk.coders.VarLongCoder;
+import org.apache.beam.sdk.schemas.Schema;
+import org.apache.beam.sdk.testing.PAssert;
+import org.apache.beam.sdk.testing.UsesPythonExpansionService;
+import org.apache.beam.sdk.testing.ValidatesRunner;
+import org.apache.beam.sdk.transforms.Create;
+import org.apache.beam.sdk.values.PCollection;
+import org.apache.beam.sdk.values.Row;
+import org.junit.Test;
+import org.junit.experimental.categories.Category;
+import org.junit.runner.RunWith;
+import org.junit.runners.JUnit4;
+
+@RunWith(JUnit4.class)
+public class RunInferenceTransformTest extends BaseExternalTest {
+ @Test
+ @Category({ValidatesRunner.class, UsesPythonExpansionService.class})
+ public void testRunInference() {
+ String stagingLocation =
+ Optional.ofNullable(System.getProperty("semiPersistDir")).orElse("/tmp");
+ Schema schema =
+ Schema.of(
+ Schema.Field.of("example", Schema.FieldType.array(Schema.FieldType.INT64)),
+ Schema.Field.of("inference", Schema.FieldType.INT32));
+ Row row0 = Row.withSchema(schema).addArray(0L, 0L).addValue(0).build();
+ Row row1 = Row.withSchema(schema).addArray(1L, 1L).addValue(1).build();
+ PCollection<Row> col =
+ testPipeline
+ .apply(Create.<Iterable<Long>>of(Arrays.asList(0L, 0L), Arrays.asList(1L, 1L)))
+ .setCoder(IterableCoder.of(VarLongCoder.of()))
+ .apply(
+ RunInference.of(
+ "apache_beam.ml.inference.sklearn_inference.SklearnModelHandlerNumpy",
+ schema)
+ .withKwarg(
+ "model_uri", String.format("%s/staged/sklearn_model", stagingLocation))
Review Comment:
Where does the "staged" come from? I think we need some comments here about who creates the model and how it gets saved for this test.
##########
sdks/java/extensions/python/src/main/java/org/apache/beam/sdk/extensions/python/transforms/RunInference.java:
##########
@@ -0,0 +1,102 @@
+/*
+ * 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.beam.sdk.extensions.python.transforms;
+
+import java.util.Map;
+import org.apache.beam.sdk.coders.RowCoder;
+import org.apache.beam.sdk.extensions.python.PythonExternalTransform;
+import org.apache.beam.sdk.schemas.Schema;
+import org.apache.beam.sdk.transforms.PTransform;
+import org.apache.beam.sdk.util.PythonCallableSource;
+import org.apache.beam.sdk.values.PCollection;
+import org.apache.beam.sdk.values.Row;
+import org.apache.beam.vendor.guava.v26_0_jre.com.google.common.collect.ImmutableMap;
+
+/** Wrapper for invoking external Python RunInference. */
+public class RunInference extends PTransform<PCollection<?>, PCollection<Row>> {
+ private final String modelLoader;
+ private final Schema schema;
+ private final Map<String, Object> kwargs;
+ private final String expansionService;
+
+ /**
+ * Instantiates a multi-language wrapper for a Python RunInference with a given model loader.
+ *
+ * @param modelLoader A Python lambda function for a model loader class object.
+ * @param exampleType A schema field type for the example column in output rows.
+ * @param inferenceType A schema field type for the inference column in output rows.
+ * @return A {@link RunInference} for the given model loader.
+ */
+ public static RunInference of(
+ String modelLoader, Schema.FieldType exampleType, Schema.FieldType inferenceType) {
+ Schema schema =
+ Schema.of(
+ Schema.Field.of("example", exampleType), Schema.Field.of("inference", inferenceType));
+ return new RunInference(modelLoader, schema, ImmutableMap.of(), "");
+ }
+
+ /**
+ * Instantiates a multi-language wrapper for a Python RunInference with a given model loader.
+ *
+ * @param modelLoader A Python lambda function for a model loader class object.
+ * @param schema A schema for output rows.
+ * @return A {@link RunInference} for the given model loader.
+ */
+ public static RunInference of(String modelLoader, Schema schema) {
+ return new RunInference(modelLoader, schema, ImmutableMap.of(), "");
+ }
+
+ /**
+ * Sets keyword arguments for RunInference constructor.
+ *
+ * @return A {@link RunInference} with keyword arguments.
+ */
+ public RunInference withKwarg(String key, Object arg) {
+ ImmutableMap.Builder<String, Object> builder =
+ ImmutableMap.<String, Object>builder().putAll(kwargs).put(key, arg);
+ return new RunInference(modelLoader, schema, builder.build(), expansionService);
+ }
+
+ /**
+ * Sets an expansion service endpoint for RunInference.
+ *
+ * @param expansionService A URL for a Python expansion service.
+ * @return A {@link RunInference} for the given expansion service endpoint.
+ */
+ public RunInference withExpansionService(String expansionService) {
+ return new RunInference(modelLoader, schema, kwargs, expansionService);
+ }
+
+ private RunInference(
+ String modelLoader, Schema schema, Map<String, Object> kwargs, String expansionService) {
+ this.modelLoader = modelLoader;
+ this.schema = schema;
+ this.kwargs = kwargs;
+ this.expansionService = expansionService;
+ }
+
+ @Override
+ public PCollection<Row> expand(PCollection<?> input) {
+ return input.apply(
+ PythonExternalTransform.<PCollection<?>, PCollection<Row>>from(
+ "apache_beam.ml.inference.base.RunInference.create", expansionService)
+ .withKwarg("model_handler_provider", PythonCallableSource.of(modelLoader))
+ .withKwargs(kwargs)
+ .withOutputCoder(RowCoder.of(schema)));
Review Comment:
I agree this is unfortunate, but there are lots of cases where the type inferencer can't figure this out, and it doesn't matter (much) in Python as Object is just fine (e.g. it gets fused away or we can use FastPrimitives/PickleCoder), but for cross language this information must be provided explicitly.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org