Align names with those produced by the dataflow runner harness. These will be unused once the runner harness produces the correct transform payloads.
Project: http://git-wip-us.apache.org/repos/asf/beam/repo Commit: http://git-wip-us.apache.org/repos/asf/beam/commit/a375b2e1 Tree: http://git-wip-us.apache.org/repos/asf/beam/tree/a375b2e1 Diff: http://git-wip-us.apache.org/repos/asf/beam/diff/a375b2e1 Branch: refs/heads/release-2.2.0 Commit: a375b2e1a498dbae827ae2398523283aacb51827 Parents: c2e0306 Author: Robert Bradshaw <[email protected]> Authored: Wed Oct 4 13:57:01 2017 -0700 Committer: Ahmet Altay <[email protected]> Committed: Mon Oct 30 14:39:58 2017 -0700 ---------------------------------------------------------------------- sdks/python/apache_beam/runners/worker/bundle_processor.py | 8 ++++---- 1 file changed, 4 insertions(+), 4 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/beam/blob/a375b2e1/sdks/python/apache_beam/runners/worker/bundle_processor.py ---------------------------------------------------------------------- diff --git a/sdks/python/apache_beam/runners/worker/bundle_processor.py b/sdks/python/apache_beam/runners/worker/bundle_processor.py index b69d002..69e4ade 100644 --- a/sdks/python/apache_beam/runners/worker/bundle_processor.py +++ b/sdks/python/apache_beam/runners/worker/bundle_processor.py @@ -58,8 +58,8 @@ IDENTITY_DOFN_URN = 'urn:org.apache.beam:dofn:identity:0.1' PYTHON_ITERABLE_VIEWFN_URN = 'urn:org.apache.beam:viewfn:iterable:python:0.1' PYTHON_CODER_URN = 'urn:org.apache.beam:coder:python:0.1' # TODO(vikasrk): Fix this once runner sends appropriate python urns. -PYTHON_DOFN_URN = 'urn:org.apache.beam:dofn:java:0.1' -PYTHON_SOURCE_URN = 'urn:org.apache.beam:source:java:0.1' +OLD_DATAFLOW_RUNNER_HARNESS_PARDO_URN = 'urn:beam:dofn:javasdk:0.1' +OLD_DATAFLOW_RUNNER_HARNESS_READ_URN = 'urn:org.apache.beam:source:java:0.1' def side_input_tag(transform_id, tag): @@ -358,7 +358,7 @@ def create(factory, transform_id, transform_proto, grpc_port, consumers): data_channel=factory.data_channel_factory.create_data_channel(grpc_port)) [email protected]_urn(PYTHON_SOURCE_URN, None) [email protected]_urn(OLD_DATAFLOW_RUNNER_HARNESS_READ_URN, None) def create(factory, transform_id, transform_proto, parameter, consumers): # The Dataflow runner harness strips the base64 encoding. source = pickler.loads(base64.b64encode(parameter)) @@ -393,7 +393,7 @@ def create(factory, transform_id, transform_proto, parameter, consumers): consumers) [email protected]_urn(PYTHON_DOFN_URN, None) [email protected]_urn(OLD_DATAFLOW_RUNNER_HARNESS_PARDO_URN, None) def create(factory, transform_id, transform_proto, parameter, consumers): dofn_data = pickler.loads(parameter) if len(dofn_data) == 2:
