You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@beam.apache.org by ro...@apache.org on 2016/07/25 19:04:07 UTC
[1/4] incubator-beam git commit: Make DoFnRunner a Receiver.
Repository: incubator-beam
Updated Branches:
refs/heads/python-sdk 38d9dea2e -> 153916fe9
Make DoFnRunner a Receiver.
Project: http://git-wip-us.apache.org/repos/asf/incubator-beam/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-beam/commit/ac3249fe
Tree: http://git-wip-us.apache.org/repos/asf/incubator-beam/tree/ac3249fe
Diff: http://git-wip-us.apache.org/repos/asf/incubator-beam/diff/ac3249fe
Branch: refs/heads/python-sdk
Commit: ac3249fe51e9674149ae9cccd48c7290aa8bf34d
Parents: 38d9dea
Author: Robert Bradshaw <ro...@google.com>
Authored: Fri Jul 22 17:33:57 2016 -0700
Committer: Robert Bradshaw <ro...@google.com>
Committed: Mon Jul 25 11:56:52 2016 -0700
----------------------------------------------------------------------
sdks/python/apache_beam/runners/common.pxd | 15 +++++++++------
sdks/python/apache_beam/runners/common.py | 20 +++++++++++++++-----
2 files changed, 24 insertions(+), 11 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/ac3249fe/sdks/python/apache_beam/runners/common.pxd
----------------------------------------------------------------------
diff --git a/sdks/python/apache_beam/runners/common.pxd b/sdks/python/apache_beam/runners/common.pxd
index e1bf461..c256ded 100644
--- a/sdks/python/apache_beam/runners/common.pxd
+++ b/sdks/python/apache_beam/runners/common.pxd
@@ -20,7 +20,11 @@ from apache_beam.utils.windowed_value cimport WindowedValue
cdef type SideOutputValue, TimestampedValue
-cdef class DoFnRunner(object):
+cdef class Receiver(object):
+ cpdef receive(self, WindowedValue windowed_value)
+
+
+cdef class DoFnRunner(Receiver):
cdef object dofn
cdef object dofn_process
@@ -32,7 +36,10 @@ cdef class DoFnRunner(object):
cdef object main_receivers # TODO(robertwb): Make this a Receiver
- cpdef _process_outputs(self, element, results)
+ cpdef process(self, WindowedValue element)
+
+ @cython.locals(windowed_value=WindowedValue)
+ cpdef _process_outputs(self, WindowedValue element, results)
cdef class DoFnContext(object):
@@ -42,10 +49,6 @@ cdef class DoFnContext(object):
cpdef set_element(self, WindowedValue windowed_value)
-cdef class Receiver(object):
- cpdef receive(self, WindowedValue windowed_value)
-
-
cdef class LoggingContext(object):
# TODO(robertwb): Optimize "with [cdef class]"
cpdef enter(self)
http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/ac3249fe/sdks/python/apache_beam/runners/common.py
----------------------------------------------------------------------
diff --git a/sdks/python/apache_beam/runners/common.py b/sdks/python/apache_beam/runners/common.py
index c6b27bc..fec8a69 100644
--- a/sdks/python/apache_beam/runners/common.py
+++ b/sdks/python/apache_beam/runners/common.py
@@ -38,7 +38,18 @@ class LoggingContext(object):
pass
-class DoFnRunner(object):
+class Receiver(object):
+ """An object that consumes a WindowedValue.
+
+ This class can be efficiently used to pass values between the
+ sdk and worker harnesses.
+ """
+
+ def receive(self, windowed_value):
+ raise NotImplementedError
+
+
+class DoFnRunner(Receiver):
"""A helper class for executing ParDo operations.
"""
@@ -81,6 +92,9 @@ class DoFnRunner(object):
# Optimize for the common case.
self.main_receivers = tagged_receivers[None]
+ def receive(self, windowed_value):
+ self.process(windowed_value)
+
def start(self):
self.context.set_element(None)
try:
@@ -236,7 +250,3 @@ class DoFnContext(object):
def aggregate_to(self, aggregator, input_value):
self.state.counter_for(aggregator).update(input_value)
-
-class Receiver(object):
- def receive(self, windowed_value):
- raise NotImplementedError
[2/4] incubator-beam git commit: Receiver and LoggingContext adapters.
Posted by ro...@apache.org.
Receiver and LoggingContext adapters.
This allows for cdef calls iff the right type is passed in.
Project: http://git-wip-us.apache.org/repos/asf/incubator-beam/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-beam/commit/d20cf642
Tree: http://git-wip-us.apache.org/repos/asf/incubator-beam/tree/d20cf642
Diff: http://git-wip-us.apache.org/repos/asf/incubator-beam/diff/d20cf642
Branch: refs/heads/python-sdk
Commit: d20cf64287fdd22b1efa1a3afe346a0a1d5cf924
Parents: ac3249f
Author: Robert Bradshaw <ro...@google.com>
Authored: Sat Jul 23 00:37:42 2016 -0700
Committer: Robert Bradshaw <ro...@google.com>
Committed: Mon Jul 25 11:57:36 2016 -0700
----------------------------------------------------------------------
sdks/python/apache_beam/runners/common.pxd | 15 ++++++--
sdks/python/apache_beam/runners/common.py | 48 ++++++++++++++++++++++---
2 files changed, 57 insertions(+), 6 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/d20cf642/sdks/python/apache_beam/runners/common.pxd
----------------------------------------------------------------------
diff --git a/sdks/python/apache_beam/runners/common.pxd b/sdks/python/apache_beam/runners/common.pxd
index c256ded..7acd049 100644
--- a/sdks/python/apache_beam/runners/common.pxd
+++ b/sdks/python/apache_beam/runners/common.pxd
@@ -15,8 +15,11 @@
# limitations under the License.
#
+cimport cython
+
from apache_beam.utils.windowed_value cimport WindowedValue
+
cdef type SideOutputValue, TimestampedValue
@@ -31,10 +34,10 @@ cdef class DoFnRunner(Receiver):
cdef object window_fn
cdef object context # TODO(robertwb): Make this a DoFnContext
cdef object tagged_receivers
- cdef object logging_context # TODO(robertwb): Make this a LoggingContext
+ cdef LoggingContext logging_context
cdef object step_name
- cdef object main_receivers # TODO(robertwb): Make this a Receiver
+ cdef Receiver main_receivers
cpdef process(self, WindowedValue element)
@@ -53,3 +56,11 @@ cdef class LoggingContext(object):
# TODO(robertwb): Optimize "with [cdef class]"
cpdef enter(self)
cpdef exit(self)
+
+
+cdef class _LoggingContextAdapter(LoggingContext):
+ cdef object underlying
+
+
+cdef class _ReceiverAdapter(Receiver):
+ cdef object underlying
http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/d20cf642/sdks/python/apache_beam/runners/common.py
----------------------------------------------------------------------
diff --git a/sdks/python/apache_beam/runners/common.py b/sdks/python/apache_beam/runners/common.py
index fec8a69..9c9ab22 100644
--- a/sdks/python/apache_beam/runners/common.py
+++ b/sdks/python/apache_beam/runners/common.py
@@ -85,12 +85,11 @@ class DoFnRunner(Receiver):
self.window_fn = windowing.windowfn
self.context = context
self.tagged_receivers = tagged_receivers
- self.logging_context = (logger.PerThreadLoggingContext(step_name=step_name)
- if logger else LoggingContext())
+ self.logging_context = get_logging_context(logger, step_name=step_name)
self.step_name = step_name
# Optimize for the common case.
- self.main_receivers = tagged_receivers[None]
+ self.main_receivers = as_receiver(tagged_receivers[None])
def receive(self, windowed_value):
self.process(windowed_value)
@@ -175,7 +174,7 @@ class DoFnRunner(Receiver):
else:
windowed_value = element.with_value(result)
if tag is None:
- self.main_receivers.output(windowed_value)
+ self.main_receivers.receive(windowed_value)
else:
self.tagged_receivers[tag].output(windowed_value)
@@ -250,3 +249,44 @@ class DoFnContext(object):
def aggregate_to(self, aggregator, input_value):
self.state.counter_for(aggregator).update(input_value)
+
+# TODO(robertwb): Remove all these adapters once service is updated out.
+
+
+class _LoggingContextAdapter(LoggingContext):
+
+ def __init__(self, underlying):
+ self.underlying = underlying
+
+ def enter(self):
+ self.underlying.enter()
+
+ def exit(self):
+ self.underlying.exit()
+
+
+def get_logging_context(maybe_logger, **kwargs):
+ if maybe_logger:
+ maybe_context = maybe_logger.PerThreadLoggingContext(**kwargs)
+ if isinstance(maybe_context, LoggingContext):
+ return maybe_context
+ else:
+ return _LoggingContextAdapter(maybe_context)
+ else:
+ return LoggingContext()
+
+
+class _ReceiverAdapter(Receiver):
+
+ def __init__(self, underlying):
+ self.underlying = underlying
+
+ def receive(self, windowed_value):
+ self.underlying.output(windowed_value)
+
+
+def as_receiver(maybe_receiver):
+ if isinstance(maybe_receiver, Receiver):
+ return maybe_receiver
+ else:
+ return _ReceiverAdapter(maybe_receiver)
[3/4] incubator-beam git commit: Allow passing logging context
directly.
Posted by ro...@apache.org.
Allow passing logging context directly.
This is better than passing a logger module with a specific class.
Project: http://git-wip-us.apache.org/repos/asf/incubator-beam/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-beam/commit/de35f282
Tree: http://git-wip-us.apache.org/repos/asf/incubator-beam/tree/de35f282
Diff: http://git-wip-us.apache.org/repos/asf/incubator-beam/diff/de35f282
Branch: refs/heads/python-sdk
Commit: de35f28294dfb4f47b429635eff843b38279e240
Parents: d20cf64
Author: Robert Bradshaw <ro...@google.com>
Authored: Sat Jul 23 00:54:11 2016 -0700
Committer: Robert Bradshaw <ro...@google.com>
Committed: Mon Jul 25 11:57:42 2016 -0700
----------------------------------------------------------------------
sdks/python/apache_beam/runners/common.py | 10 ++++++++--
1 file changed, 8 insertions(+), 2 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/de35f282/sdks/python/apache_beam/runners/common.py
----------------------------------------------------------------------
diff --git a/sdks/python/apache_beam/runners/common.py b/sdks/python/apache_beam/runners/common.py
index 9c9ab22..c017704 100644
--- a/sdks/python/apache_beam/runners/common.py
+++ b/sdks/python/apache_beam/runners/common.py
@@ -62,7 +62,9 @@ class DoFnRunner(Receiver):
context,
tagged_receivers,
logger=None,
- step_name=None):
+ step_name=None,
+ # Preferred alternative to logger
+ logging_context=None):
if not args and not kwargs:
self.dofn = fn
self.dofn_process = fn.process
@@ -85,9 +87,13 @@ class DoFnRunner(Receiver):
self.window_fn = windowing.windowfn
self.context = context
self.tagged_receivers = tagged_receivers
- self.logging_context = get_logging_context(logger, step_name=step_name)
self.step_name = step_name
+ if logging_context:
+ self.logging_context = logging_context
+ else:
+ self.logging_context = get_logging_context(logger, step_name=step_name)
+
# Optimize for the common case.
self.main_receivers = as_receiver(tagged_receivers[None])
[4/4] incubator-beam git commit: Closes #721
Posted by ro...@apache.org.
Closes #721
Project: http://git-wip-us.apache.org/repos/asf/incubator-beam/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-beam/commit/153916fe
Tree: http://git-wip-us.apache.org/repos/asf/incubator-beam/tree/153916fe
Diff: http://git-wip-us.apache.org/repos/asf/incubator-beam/diff/153916fe
Branch: refs/heads/python-sdk
Commit: 153916fe9f641f5fe5e8d473caef67f449ee6bca
Parents: 38d9dea de35f28
Author: Robert Bradshaw <ro...@google.com>
Authored: Mon Jul 25 12:02:02 2016 -0700
Committer: Robert Bradshaw <ro...@google.com>
Committed: Mon Jul 25 12:02:02 2016 -0700
----------------------------------------------------------------------
sdks/python/apache_beam/runners/common.pxd | 30 ++++++++---
sdks/python/apache_beam/runners/common.py | 72 ++++++++++++++++++++++---
2 files changed, 86 insertions(+), 16 deletions(-)
----------------------------------------------------------------------