beam-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From rober...@apache.org
Subject [1/2] beam git commit: BEAM-1443: Preserve window duplication on window assignment.
Date Fri, 10 Feb 2017 17:48:16 GMT
Repository: beam
Updated Branches:
  refs/heads/master 9c4a784bb -> caaa64dc6


BEAM-1443: Preserve window duplication on window assignment.


Project: http://git-wip-us.apache.org/repos/asf/beam/repo
Commit: http://git-wip-us.apache.org/repos/asf/beam/commit/71264671
Tree: http://git-wip-us.apache.org/repos/asf/beam/tree/71264671
Diff: http://git-wip-us.apache.org/repos/asf/beam/diff/71264671

Branch: refs/heads/master
Commit: 712646714b8efec6b024ebd2ba0f0e066280f36e
Parents: 9c4a784
Author: Robert Bradshaw <robertwb@google.com>
Authored: Wed Feb 8 17:58:32 2017 -0800
Committer: Robert Bradshaw <robertwb@gmail.com>
Committed: Fri Feb 10 09:47:47 2017 -0800

----------------------------------------------------------------------
 sdks/python/apache_beam/runners/common.py         | 13 +++++++++----
 sdks/python/apache_beam/transforms/window_test.py | 16 ++++++++++++++++
 2 files changed, 25 insertions(+), 4 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/beam/blob/71264671/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 6f86ca0..2c1032d 100644
--- a/sdks/python/apache_beam/runners/common.py
+++ b/sdks/python/apache_beam/runners/common.py
@@ -224,7 +224,7 @@ class DoFnRunner(Receiver):
     # 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
     # otherwise as none of the arguments are changing
-    if self.has_windowed_inputs and len(element.windows) > 1:
+    if self.has_windowed_inputs and len(element.windows) != 1:
       for w in element.windows:
         self._dofn_per_window_invoker(
             WindowedValue(element.value, element.timestamp, (w,)))
@@ -280,7 +280,7 @@ class DoFnRunner(Receiver):
     else:
       raise
 
-  def _process_outputs(self, element, results):
+  def _process_outputs(self, windowed_input_element, results):
     """Dispatch the result of computation to the appropriate receivers.
 
     A value wrapped in a SideOutputValue object will be unwrapped and
@@ -297,7 +297,10 @@ class DoFnRunner(Receiver):
         result = result.value
       if isinstance(result, WindowedValue):
         windowed_value = result
-      elif element is None:
+        if (windowed_input_element is not None
+            and len(windowed_input_element.windows) != 1):
+          windowed_value.windows *= len(windowed_input_element.windows)
+      elif windowed_input_element is None:
         # Start and finish have no element from which to grab context,
         # but may emit elements.
         if isinstance(result, TimestampedValue):
@@ -315,8 +318,10 @@ class DoFnRunner(Receiver):
         windowed_value = WindowedValue(
             result.value, result.timestamp,
             self.window_fn.assign(assign_context))
+        if len(windowed_input_element.windows) != 1:
+          windowed_value.windows *= len(windowed_input_element.windows)
       else:
-        windowed_value = element.with_value(result)
+        windowed_value = windowed_input_element.with_value(result)
       if tag is None:
         self.main_receivers.receive(windowed_value)
       else:

http://git-wip-us.apache.org/repos/asf/beam/blob/71264671/sdks/python/apache_beam/transforms/window_test.py
----------------------------------------------------------------------
diff --git a/sdks/python/apache_beam/transforms/window_test.py b/sdks/python/apache_beam/transforms/window_test.py
index 1a21709..c4072ac 100644
--- a/sdks/python/apache_beam/transforms/window_test.py
+++ b/sdks/python/apache_beam/transforms/window_test.py
@@ -176,6 +176,22 @@ class WindowTest(unittest.TestCase):
                                   ('key', [5, 6, 7, 8, 9])]))
     p.run()
 
+  def test_rewindow(self):
+    p = TestPipeline()
+    result = (p
+              | Create([(k, k) for k in range(10)])
+              | Map(lambda (x, t): TimestampedValue(x, t))
+              | 'window' >> WindowInto(SlidingWindows(period=2, size=6))
+              # Per the model, each element is now duplicated across
+              # three windows. Rewindowing must preserve this duplication.
+              | 'rewindow' >> WindowInto(FixedWindows(5))
+              | 'rewindow2' >> WindowInto(FixedWindows(5))
+              | Map(lambda v: ('key', v))
+              | GroupByKey())
+    assert_that(result, equal_to([('key', sorted([0, 1, 2, 3, 4] * 3)),
+                                  ('key', sorted([5, 6, 7, 8, 9] * 3))]))
+    p.run()
+
   def test_timestamped_with_combiners(self):
     p = TestPipeline()
     result = (p


Mime
View raw message