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/09/16 22:50:31 UTC
[1/3] incubator-beam git commit: Allow pickling of UnwindowedValues
instances
Repository: incubator-beam
Updated Branches:
refs/heads/python-sdk a60b58a94 -> ebae225ed
Allow pickling of UnwindowedValues instances
These are trivial wrappers produced when "windowing" by the global window, but may need to be materialized in some cases.
Project: http://git-wip-us.apache.org/repos/asf/incubator-beam/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-beam/commit/63904e01
Tree: http://git-wip-us.apache.org/repos/asf/incubator-beam/tree/63904e01
Diff: http://git-wip-us.apache.org/repos/asf/incubator-beam/diff/63904e01
Branch: refs/heads/python-sdk
Commit: 63904e01f52be498ba3c8c9e1989bcffb02dfcd6
Parents: a60b58a
Author: Robert Bradshaw <ro...@gmail.com>
Authored: Thu Sep 15 17:00:42 2016 -0700
Committer: GitHub <no...@github.com>
Committed: Thu Sep 15 17:00:42 2016 -0700
----------------------------------------------------------------------
sdks/python/apache_beam/transforms/trigger.py | 3 +++
1 file changed, 3 insertions(+)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/63904e01/sdks/python/apache_beam/transforms/trigger.py
----------------------------------------------------------------------
diff --git a/sdks/python/apache_beam/transforms/trigger.py b/sdks/python/apache_beam/transforms/trigger.py
index 587384a..58b6154 100644
--- a/sdks/python/apache_beam/transforms/trigger.py
+++ b/sdks/python/apache_beam/transforms/trigger.py
@@ -734,6 +734,9 @@ class DefaultGlobalBatchTriggerDriver(TriggerDriver):
def __repr__(self):
return '<UnwindowedValues of %s>' % windowed_values
+
+ def __reduce__(self):
+ return list, (list(self),)
unwindowed = UnwindowedValues()
yield WindowedValue(unwindowed, MIN_TIMESTAMP, self.GLOBAL_WINDOW_TUPLE)
[2/3] incubator-beam git commit: Add unit test for unwindowed
iterator picking.
Posted by ro...@apache.org.
Add unit test for unwindowed iterator picking.
Also lifted this out to a top-level class rather than defining
it on every element now that it's not longer a simple generator
statement.
Project: http://git-wip-us.apache.org/repos/asf/incubator-beam/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-beam/commit/e9b1e412
Tree: http://git-wip-us.apache.org/repos/asf/incubator-beam/tree/e9b1e412
Diff: http://git-wip-us.apache.org/repos/asf/incubator-beam/diff/e9b1e412
Branch: refs/heads/python-sdk
Commit: e9b1e41240d5032cdaa2b745c95f94da45475f34
Parents: 63904e0
Author: Robert Bradshaw <ro...@google.com>
Authored: Fri Sep 16 15:41:22 2016 -0700
Committer: Robert Bradshaw <ro...@google.com>
Committed: Fri Sep 16 15:41:22 2016 -0700
----------------------------------------------------------------------
sdks/python/apache_beam/transforms/trigger.py | 34 ++++++++++++--------
.../apache_beam/transforms/trigger_test.py | 12 +++++++
2 files changed, 33 insertions(+), 13 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/e9b1e412/sdks/python/apache_beam/transforms/trigger.py
----------------------------------------------------------------------
diff --git a/sdks/python/apache_beam/transforms/trigger.py b/sdks/python/apache_beam/transforms/trigger.py
index 58b6154..8c23873 100644
--- a/sdks/python/apache_beam/transforms/trigger.py
+++ b/sdks/python/apache_beam/transforms/trigger.py
@@ -713,6 +713,26 @@ class TriggerDriver(object):
pass
+class _UnwindowedValues(observable.ObservableMixin):
+ """Exposes iterable of windowed values as interable of unwindowed values."""
+
+ def __init__(self, windowed_values):
+ super(_UnwindowedValues, self).__init__()
+ self._windowed_values = windowed_values
+
+ def __iter__(self):
+ for wv in self._windowed_values:
+ unwindowed_value = wv.value
+ self.notify_observers(unwindowed_value)
+ yield unwindowed_value
+
+ def __repr__(self):
+ return '<_UnwindowedValues of %s>' % self._windowed_values
+
+ def __reduce__(self):
+ return list, (list(self),)
+
+
class DefaultGlobalBatchTriggerDriver(TriggerDriver):
"""Breaks a bundles into window (pane)s according to the default triggering.
"""
@@ -725,19 +745,7 @@ class DefaultGlobalBatchTriggerDriver(TriggerDriver):
if isinstance(windowed_values, list):
unwindowed = [wv.value for wv in windowed_values]
else:
- class UnwindowedValues(observable.ObservableMixin):
- def __iter__(self):
- for wv in windowed_values:
- unwindowed_value = wv.value
- self.notify_observers(unwindowed_value)
- yield unwindowed_value
-
- def __repr__(self):
- return '<UnwindowedValues of %s>' % windowed_values
-
- def __reduce__(self):
- return list, (list(self),)
- unwindowed = UnwindowedValues()
+ unwindowed = _UnwindowedValues(windowed_values)
yield WindowedValue(unwindowed, MIN_TIMESTAMP, self.GLOBAL_WINDOW_TUPLE)
def process_timer(self, window_id, name, time_domain, timestamp, state):
http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/e9b1e412/sdks/python/apache_beam/transforms/trigger_test.py
----------------------------------------------------------------------
diff --git a/sdks/python/apache_beam/transforms/trigger_test.py b/sdks/python/apache_beam/transforms/trigger_test.py
index a3ad8d8..c37d4ae 100644
--- a/sdks/python/apache_beam/transforms/trigger_test.py
+++ b/sdks/python/apache_beam/transforms/trigger_test.py
@@ -19,12 +19,14 @@
import collections
import os.path
+import pickle
import unittest
import yaml
import apache_beam as beam
from apache_beam.pipeline import Pipeline
+from apache_beam.transforms import trigger
from apache_beam.transforms.core import Windowing
from apache_beam.transforms.trigger import AccumulationMode
from apache_beam.transforms.trigger import AfterAll
@@ -366,6 +368,16 @@ class TriggerTest(unittest.TestCase):
IntervalWindow(0, 17): [set('abcdefgh')]},
2)
+ def test_picklable_output(self):
+ global_window = trigger.GlobalWindow(),
+ driver = trigger.DefaultGlobalBatchTriggerDriver()
+ unpicklable = (WindowedValue(k, 0, global_window)
+ for k in range(10))
+ with self.assertRaises(TypeError):
+ pickle.dumps(unpicklable)
+ for unwindowed in driver.process_elements(None, unpicklable, None):
+ self.assertEqual(pickle.loads(pickle.dumps(unwindowed)).value,
+ range(10))
class TriggerPipelineTest(unittest.TestCase):
[3/3] incubator-beam git commit: Closes #964
Posted by ro...@apache.org.
Closes #964
Project: http://git-wip-us.apache.org/repos/asf/incubator-beam/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-beam/commit/ebae225e
Tree: http://git-wip-us.apache.org/repos/asf/incubator-beam/tree/ebae225e
Diff: http://git-wip-us.apache.org/repos/asf/incubator-beam/diff/ebae225e
Branch: refs/heads/python-sdk
Commit: ebae225ed30854296862c46c497d4b3e6e7de9f6
Parents: a60b58a e9b1e41
Author: Robert Bradshaw <ro...@google.com>
Authored: Fri Sep 16 15:50:09 2016 -0700
Committer: Robert Bradshaw <ro...@google.com>
Committed: Fri Sep 16 15:50:09 2016 -0700
----------------------------------------------------------------------
sdks/python/apache_beam/transforms/trigger.py | 31 +++++++++++++-------
.../apache_beam/transforms/trigger_test.py | 12 ++++++++
2 files changed, 33 insertions(+), 10 deletions(-)
----------------------------------------------------------------------