You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@beam.apache.org by ch...@apache.org on 2019/03/30 01:29:58 UTC
[beam] branch master updated: Fixes a bug in external transform to
runner API proto conversion.
This is an automated email from the ASF dual-hosted git repository.
chamikara pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/beam.git
The following commit(s) were added to refs/heads/master by this push:
new b64c400 Fixes a bug in external transform to runner API proto conversion.
new 00c7810 Merge pull request #8165: [BEAM-6894] Fixes a bug in external transform and adds tests
b64c400 is described below
commit b64c40064837aee515926b32a74837747f7f70ae
Author: chamikara@google.com <ch...@google.com>
AuthorDate: Thu Mar 28 18:56:43 2019 -0700
Fixes a bug in external transform to runner API proto conversion.
Also, adds a unit test and an integration test for python to python external transform expansion.
---
sdks/python/apache_beam/transforms/external.py | 1 +
.../python/apache_beam/transforms/external_test.py | 39 ++++++++++++
.../apache_beam/transforms/external_test_it.py | 70 ++++++++++++++++++++++
3 files changed, 110 insertions(+)
diff --git a/sdks/python/apache_beam/transforms/external.py b/sdks/python/apache_beam/transforms/external.py
index 71577d5..cd670b6 100644
--- a/sdks/python/apache_beam/transforms/external.py
+++ b/sdks/python/apache_beam/transforms/external.py
@@ -184,6 +184,7 @@ class ExternalTransform(ptransform.PTransform):
if id.startswith(self._namespace):
context.environments.put_proto(id, proto)
for id, proto in self._expanded_components.pcollections.items():
+ id = pcoll_renames.get(id, id)
if id not in context.pcollections._id_to_obj.keys():
context.pcollections.put_proto(id, proto)
diff --git a/sdks/python/apache_beam/transforms/external_test.py b/sdks/python/apache_beam/transforms/external_test.py
index 3432017..c3448c6 100644
--- a/sdks/python/apache_beam/transforms/external_test.py
+++ b/sdks/python/apache_beam/transforms/external_test.py
@@ -28,6 +28,7 @@ import grpc
from past.builtins import unicode
import apache_beam as beam
+from apache_beam import Pipeline
from apache_beam.io.external.generate_sequence import GenerateSequence
from apache_beam.portability import python_urns
from apache_beam.runners.portability import expansion_service
@@ -41,6 +42,44 @@ class ExternalTransformTest(unittest.TestCase):
# This will be overwritten if set via a flag.
expansion_service_jar = None
+ def test_pipeline_generation(self):
+
+ @ptransform.PTransform.register_urn('simple', None)
+ class SimpleTransform(ptransform.PTransform):
+ def expand(self, pcoll):
+ return pcoll | 'TestLabel' >> beam.Map(lambda x: 'Simple(%s)' % x)
+
+ def to_runner_api_parameter(self, unused_context):
+ return 'simple', None
+
+ @staticmethod
+ def from_runner_api_parameter(unused_parameter, unused_context):
+ return SimpleTransform()
+
+ pipeline = beam.Pipeline()
+ res = (pipeline
+ | beam.Create(['a', 'b'])
+ | beam.ExternalTransform(
+ 'simple',
+ None,
+ expansion_service.ExpansionServiceServicer()))
+ assert_that(res, equal_to(['Simple(a)', 'Simple(b)']))
+
+ proto, _ = pipeline.to_runner_api(
+ return_context=True)
+ pipeline_from_proto = Pipeline.from_runner_api(
+ proto, pipeline.runner, pipeline._options)
+
+ # Original pipeline has the un-expanded external transform
+ self.assertEqual([], pipeline.transforms_stack[0].parts[1].parts)
+
+ # new pipeline has the expanded external transform
+ self.assertNotEqual(
+ [], pipeline_from_proto.transforms_stack[0].parts[1].parts)
+ self.assertEqual(
+ u'ExternalTransform(simple)/TestLabel',
+ pipeline_from_proto.transforms_stack[0].parts[1].parts[0].full_label)
+
def test_simple(self):
@ptransform.PTransform.register_urn('simple', None)
diff --git a/sdks/python/apache_beam/transforms/external_test_it.py b/sdks/python/apache_beam/transforms/external_test_it.py
new file mode 100644
index 0000000..97f857c
--- /dev/null
+++ b/sdks/python/apache_beam/transforms/external_test_it.py
@@ -0,0 +1,70 @@
+#
+# 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.
+#
+
+"""Integration tests for cross-language transform expansion."""
+
+from __future__ import absolute_import
+
+import unittest
+
+from nose.plugins.attrib import attr
+
+import apache_beam as beam
+from apache_beam import Pipeline
+from apache_beam.runners.portability import expansion_service
+from apache_beam.testing.test_pipeline import TestPipeline
+from apache_beam.testing.util import assert_that
+from apache_beam.testing.util import equal_to
+from apache_beam.transforms import ptransform
+
+
+class ExternalTransformIT(unittest.TestCase):
+
+ @attr('IT')
+ def test_job_python_from_python_it(self):
+ @ptransform.PTransform.register_urn('simple', None)
+ class SimpleTransform(ptransform.PTransform):
+ def expand(self, pcoll):
+ return pcoll | beam.Map(lambda x: 'Simple(%s)' % x)
+
+ def to_runner_api_parameter(self, unused_context):
+ return 'simple', None
+
+ @staticmethod
+ def from_runner_api_parameter(_1, _2):
+ return SimpleTransform()
+
+ pipeline = TestPipeline(is_integration_test=True)
+
+ res = (
+ pipeline
+ | beam.Create(['a', 'b'])
+ | beam.ExternalTransform(
+ 'simple',
+ None,
+ expansion_service.ExpansionServiceServicer()))
+ assert_that(res, equal_to(['Simple(a)', 'Simple(b)']))
+
+ proto_pipeline, _ = pipeline.to_runner_api(
+ return_context=True)
+ pipeline_from_proto = Pipeline.from_runner_api(
+ proto_pipeline, pipeline.runner, pipeline._options)
+ pipeline_from_proto.run().wait_until_finish()
+
+
+if __name__ == '__main__':
+ unittest.main()