[
https://issues.apache.org/jira/browse/BEAM-1630?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=16312125#comment-16312125
]
ASF GitHub Bot commented on BEAM-1630:
--------------------------------------
chamikaramj closed pull request #4342: [BEAM-1630] Updates common.py due to a
Cython bug.
URL: https://github.com/apache/beam/pull/4342
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/common.pxd
b/sdks/python/apache_beam/runners/common.pxd
index dd7f3e45953..fed0c2c95c2 100644
--- a/sdks/python/apache_beam/runners/common.pxd
+++ b/sdks/python/apache_beam/runners/common.pxd
@@ -47,10 +47,11 @@ cdef class DoFnSignature(object):
cdef class DoFnInvoker(object):
cdef public DoFnSignature signature
- cdef _OutputProcessor output_processor
+ cdef OutputProcessor output_processor
cpdef invoke_process(self, WindowedValue windowed_value,
- restriction_tracker=*, output_processor=*)
+ restriction_tracker=*,
+ OutputProcessor output_processor=*)
cpdef invoke_start_bundle(self)
cpdef invoke_finish_bundle(self)
cpdef invoke_split(self, element, restriction)
@@ -85,16 +86,15 @@ cdef class DoFnRunner(Receiver):
cdef class OutputProcessor(object):
- pass
+ @cython.locals(windowed_value=WindowedValue)
+ cpdef process_outputs(self, WindowedValue element, results)
+
cdef class _OutputProcessor(OutputProcessor):
cdef object window_fn
cdef Receiver main_receivers
cdef object tagged_receivers
- @cython.locals(windowed_value=WindowedValue)
- cpdef process_outputs(self, WindowedValue element, results)
-
cdef class DoFnContext(object):
cdef object label
diff --git a/sdks/python/apache_beam/runners/common.py
b/sdks/python/apache_beam/runners/common.py
index 57c9f389cd9..d5ca68307aa 100644
--- a/sdks/python/apache_beam/runners/common.py
+++ b/sdks/python/apache_beam/runners/common.py
@@ -264,7 +264,8 @@ def __init__(self, output_processor, signature):
def invoke_process(self, windowed_value, restriction_tracker=None,
output_processor=None):
- output_processor = output_processor or self.output_processor
+ if not output_processor:
+ output_processor = self.output_processor
output_processor.process_outputs(
windowed_value, self.process_method(windowed_value.value))
@@ -349,7 +350,8 @@ def __init__(self, placeholder):
def invoke_process(self, windowed_value, restriction_tracker=None,
output_processor=None):
- output_processor = output_processor or self.output_processor
+ if not output_processor:
+ output_processor = self.output_processor
self.context.set_element(windowed_value)
# Call for the process function for each window if has windowed side inputs
# or if the process accesses the window parameter. We can just call it once
----------------------------------------------------------------
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]
> Add Splittable DoFn to Python SDK
> ---------------------------------
>
> Key: BEAM-1630
> URL: https://issues.apache.org/jira/browse/BEAM-1630
> Project: Beam
> Issue Type: Improvement
> Components: sdk-py-core
> Reporter: Chamikara Jayalath
> Assignee: Chamikara Jayalath
>
> Splittable DoFn [1] is currently being implemented for Java SDK [2]. We
> should add this to Python SDK as well.
> Following document proposes an API for this.
> https://docs.google.com/document/d/1h_zprJrOilivK2xfvl4L42vaX4DMYGfH1YDmi-s_ozM/edit?usp=sharing
> [1] https://s.apache.org/splittable-do-fn
> [2]
> https://lists.apache.org/thread.html/0ce61ac162460a149d5c93cdface37cc383f8030fe86ca09e5699b18@%3Cdev.beam.apache.org%3E
--
This message was sent by Atlassian JIRA
(v6.4.14#64029)
