[
https://issues.apache.org/jira/browse/BEAM-2717?focusedWorklogId=163850&page=com.atlassian.jira.plugin.system.issuetabpanels:worklog-tabpanel#worklog-163850
]
ASF GitHub Bot logged work on BEAM-2717:
----------------------------------------
Author: ASF GitHub Bot
Created on: 08/Nov/18 09:26
Start Date: 08/Nov/18 09:26
Worklog Time Spent: 10m
Work Description: robertwb closed pull request #6959: [BEAM-2717] Cleanup
portable runner.
URL: https://github.com/apache/beam/pull/6959
This is a PR merged from a forked repository.
As GitHub hides the original diff on merge, it is displayed below for
the sake of provenance:
As this is a foreign pull request (from a fork), the diff is supplied
below (as it won't show otherwise due to GitHub magic):
diff --git a/sdks/python/apache_beam/runners/portability/portable_runner.py
b/sdks/python/apache_beam/runners/portability/portable_runner.py
index 756e01a04bf..0673b96a358 100644
--- a/sdks/python/apache_beam/runners/portability/portable_runner.py
+++ b/sdks/python/apache_beam/runners/portability/portable_runner.py
@@ -58,10 +58,6 @@ class PortableRunner(runner.PipelineRunner):
This runner schedules the job on a job service. The responsibility of
running and managing the job lies with the job service used.
"""
-
- def __init__(self, is_embedded_fnapi_runner=False):
- self.is_embedded_fnapi_runner = is_embedded_fnapi_runner
-
@staticmethod
def default_docker_image():
if 'USER' in os.environ:
@@ -115,23 +111,7 @@ def run_pipeline(self, pipeline):
docker = DockerizedJobServer()
job_endpoint = docker.start()
- proto_context = pipeline_context.PipelineContext(
- default_environment=PortableRunner._create_environment(
- portable_options))
- proto_pipeline = pipeline.to_runner_api(context=proto_context)
-
- if not self.is_embedded_fnapi_runner:
- # Java has different expectations about coders
- # (windowed in Fn API, but *un*windowed in runner API), whereas the
- # embedded FnApiRunner treats them consistently, so we must guard this
- # for now, until FnApiRunner is fixed.
- # See also BEAM-2717.
- for pcoll in proto_pipeline.components.pcollections.values():
- if pcoll.coder_id not in proto_context.coders:
- # This is not really a coder id, but a pickled coder.
- coder = coders.registry.get_coder(pickler.loads(pcoll.coder_id))
- pcoll.coder_id = proto_context.coders.get_id(coder)
- proto_context.coders.populate_map(proto_pipeline.components.coders)
+ proto_pipeline = pipeline.to_runner_api()
# Some runners won't detect the GroupByKey transform unless it has no
# subtransforms. Remove all sub-transforms until BEAM-4605 is resolved.
diff --git
a/sdks/python/apache_beam/runners/portability/portable_runner_test.py
b/sdks/python/apache_beam/runners/portability/portable_runner_test.py
index 2e2dfaa46dc..68ada95ac5c 100644
--- a/sdks/python/apache_beam/runners/portability/portable_runner_test.py
+++ b/sdks/python/apache_beam/runners/portability/portable_runner_test.py
@@ -141,7 +141,7 @@ def _create_job_endpoint(cls):
@classmethod
def get_runner(cls):
- return portable_runner.PortableRunner(is_embedded_fnapi_runner=True)
+ return portable_runner.PortableRunner()
@classmethod
def tearDownClass(cls):
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
[email protected]
Issue Time Tracking
-------------------
Worklog Id: (was: 163850)
Time Spent: 2h 10m (was: 2h)
> Infer coders in SDK prior to handing off pipeline to Runner
> -----------------------------------------------------------
>
> Key: BEAM-2717
> URL: https://issues.apache.org/jira/browse/BEAM-2717
> Project: Beam
> Issue Type: Bug
> Components: sdk-py-core
> Reporter: Robert Bradshaw
> Priority: Minor
> Time Spent: 2h 10m
> Remaining Estimate: 0h
>
> Currently all runners have to duplicate this work, and there's also a hack
> storing the element type rather than the coder in the Runner protos.
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
