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/10/18 20:35:51 UTC
[1/9] incubator-beam git commit: Fix tests expecting list from AsIter.
Repository: incubator-beam
Updated Branches:
refs/heads/python-sdk 6e6d89d48 -> 8e1793caf
Fix tests expecting list from AsIter.
Project: http://git-wip-us.apache.org/repos/asf/incubator-beam/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-beam/commit/66b4c2f1
Tree: http://git-wip-us.apache.org/repos/asf/incubator-beam/tree/66b4c2f1
Diff: http://git-wip-us.apache.org/repos/asf/incubator-beam/diff/66b4c2f1
Branch: refs/heads/python-sdk
Commit: 66b4c2f160b5fdce6aca9d5ea50b406fcb07bb51
Parents: 29a7378
Author: Robert Bradshaw <ro...@google.com>
Authored: Thu Oct 13 15:15:06 2016 -0700
Committer: Robert Bradshaw <ro...@google.com>
Committed: Tue Oct 18 12:17:15 2016 -0700
----------------------------------------------------------------------
.../python/apache_beam/examples/cookbook/bigquery_side_input.py | 2 +-
sdks/python/apache_beam/transforms/util.py | 5 +++--
sdks/python/apache_beam/transforms/write_ptransform_test.py | 2 +-
3 files changed, 5 insertions(+), 4 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/66b4c2f1/sdks/python/apache_beam/examples/cookbook/bigquery_side_input.py
----------------------------------------------------------------------
diff --git a/sdks/python/apache_beam/examples/cookbook/bigquery_side_input.py b/sdks/python/apache_beam/examples/cookbook/bigquery_side_input.py
index 2099e48..ffba786 100644
--- a/sdks/python/apache_beam/examples/cookbook/bigquery_side_input.py
+++ b/sdks/python/apache_beam/examples/cookbook/bigquery_side_input.py
@@ -72,7 +72,7 @@ def create_groups(group_ids, corpus, word, ignore_corpus, ignore_word):
| beam.FlatMap(
'attach word',
attach_word_fn,
- AsIter(word),
+ AsList(word),
AsSingleton(ignore_word)))
http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/66b4c2f1/sdks/python/apache_beam/transforms/util.py
----------------------------------------------------------------------
diff --git a/sdks/python/apache_beam/transforms/util.py b/sdks/python/apache_beam/transforms/util.py
index aeac0d9..ebe6ba9 100644
--- a/sdks/python/apache_beam/transforms/util.py
+++ b/sdks/python/apache_beam/transforms/util.py
@@ -20,7 +20,7 @@
from __future__ import absolute_import
-from apache_beam.pvalue import AsIter as AllOf
+from apache_beam.pvalue import AsList
from apache_beam.transforms import core
from apache_beam.transforms import window
from apache_beam.transforms.core import CombinePerKey, Create, Flatten, GroupByKey, Map
@@ -192,6 +192,7 @@ def equal_to(expected):
def is_empty():
def _empty(actual):
+ actual = list(actual)
if actual:
raise DataflowAssertException(
'Failed assert: [] == %r' % actual)
@@ -224,7 +225,7 @@ def assert_that(actual, matcher, label='assert_that'):
def apply(self, pipeline):
return pipeline | 'singleton' >> Create([None]) | Map(
match,
- AllOf(actual | core.WindowInto(window.GlobalWindows())))
+ AsList(actual | core.WindowInto(window.GlobalWindows())))
def default_label(self):
return label
http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/66b4c2f1/sdks/python/apache_beam/transforms/write_ptransform_test.py
----------------------------------------------------------------------
diff --git a/sdks/python/apache_beam/transforms/write_ptransform_test.py b/sdks/python/apache_beam/transforms/write_ptransform_test.py
index af3668c..e7cdbd4 100644
--- a/sdks/python/apache_beam/transforms/write_ptransform_test.py
+++ b/sdks/python/apache_beam/transforms/write_ptransform_test.py
@@ -100,7 +100,7 @@ class WriteTest(unittest.TestCase):
write_to_test_sink = WriteToTestSink(return_init_result,
return_write_results)
p = Pipeline(options=PipelineOptions([]))
- result = p | 'start' >> beam.Create(data) | write_to_test_sink
+ result = p | beam.Create(data) | write_to_test_sink | beam.Map(list)
assert_that(result, is_empty())
p.run()
[5/9] incubator-beam git commit: Implement windowed side inputs for
InProcess runner.
Posted by ro...@apache.org.
Implement windowed side inputs for InProcess runner.
Project: http://git-wip-us.apache.org/repos/asf/incubator-beam/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-beam/commit/7386bcca
Tree: http://git-wip-us.apache.org/repos/asf/incubator-beam/tree/7386bcca
Diff: http://git-wip-us.apache.org/repos/asf/incubator-beam/diff/7386bcca
Branch: refs/heads/python-sdk
Commit: 7386bcca10d1e7905ea65ba68a53e78b6185d876
Parents: 66b4c2f
Author: Robert Bradshaw <ro...@google.com>
Authored: Wed Oct 12 17:37:52 2016 -0700
Committer: Robert Bradshaw <ro...@google.com>
Committed: Tue Oct 18 12:17:16 2016 -0700
----------------------------------------------------------------------
.../inprocess/inprocess_evaluation_context.py | 20 +-------------------
1 file changed, 1 insertion(+), 19 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/7386bcca/sdks/python/apache_beam/runners/inprocess/inprocess_evaluation_context.py
----------------------------------------------------------------------
diff --git a/sdks/python/apache_beam/runners/inprocess/inprocess_evaluation_context.py b/sdks/python/apache_beam/runners/inprocess/inprocess_evaluation_context.py
index 883be99..c6bd41f 100644
--- a/sdks/python/apache_beam/runners/inprocess/inprocess_evaluation_context.py
+++ b/sdks/python/apache_beam/runners/inprocess/inprocess_evaluation_context.py
@@ -106,25 +106,7 @@ class _InProcessSideInputsContainer(object):
Raises:
ValueError: If values cannot be converted into the requested form.
"""
- if isinstance(view, SingletonPCollectionView):
- if len(values) == 0:
- # pylint: disable=protected-access
- result = view._view_options().get('default', EmptySideInput())
- elif len(values) == 1:
- result = values[0].value
- else:
- raise ValueError(
- ('PCollection with more than one element accessed as '
- 'a singleton view: %s.') % view)
- elif isinstance(view, IterablePCollectionView):
- result = [v.value for v in values]
- elif isinstance(view, ListPCollectionView):
- result = [v.value for v in values]
- elif isinstance(view, DictPCollectionView):
- result = dict(v.value for v in values)
- else:
- raise NotImplementedError
- return result
+ return sideinputs.SideInputMap(type(view), view._view_options(), values)
class InProcessEvaluationContext(object):
[7/9] incubator-beam git commit: Optimize globally windowed side
input case
Posted by ro...@apache.org.
Optimize globally windowed side input case
Project: http://git-wip-us.apache.org/repos/asf/incubator-beam/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-beam/commit/82f553e1
Tree: http://git-wip-us.apache.org/repos/asf/incubator-beam/tree/82f553e1
Diff: http://git-wip-us.apache.org/repos/asf/incubator-beam/diff/82f553e1
Branch: refs/heads/python-sdk
Commit: 82f553e1a777a8f88ca163e07d631d3df3e1f321
Parents: 989a189
Author: Robert Bradshaw <ro...@google.com>
Authored: Thu Oct 13 18:17:30 2016 -0700
Committer: Robert Bradshaw <ro...@google.com>
Committed: Tue Oct 18 12:17:16 2016 -0700
----------------------------------------------------------------------
.../examples/cookbook/bigquery_side_input.py | 1 -
sdks/python/apache_beam/pvalue.py | 4 ++--
sdks/python/apache_beam/runners/common.py | 16 ++++++++++++----
sdks/python/apache_beam/runners/direct_runner.py | 5 -----
.../inprocess/inprocess_evaluation_context.py | 5 -----
sdks/python/apache_beam/transforms/sideinputs.py | 9 ++++++---
.../apache_beam/transforms/sideinputs_test.py | 2 +-
7 files changed, 21 insertions(+), 21 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/82f553e1/sdks/python/apache_beam/examples/cookbook/bigquery_side_input.py
----------------------------------------------------------------------
diff --git a/sdks/python/apache_beam/examples/cookbook/bigquery_side_input.py b/sdks/python/apache_beam/examples/cookbook/bigquery_side_input.py
index ffba786..8a53637 100644
--- a/sdks/python/apache_beam/examples/cookbook/bigquery_side_input.py
+++ b/sdks/python/apache_beam/examples/cookbook/bigquery_side_input.py
@@ -33,7 +33,6 @@ from random import randrange
import apache_beam as beam
-from apache_beam.pvalue import AsIter
from apache_beam.pvalue import AsList
from apache_beam.pvalue import AsSingleton
from apache_beam.utils.options import PipelineOptions
http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/82f553e1/sdks/python/apache_beam/pvalue.py
----------------------------------------------------------------------
diff --git a/sdks/python/apache_beam/pvalue.py b/sdks/python/apache_beam/pvalue.py
index f8b7feb..e40d746 100644
--- a/sdks/python/apache_beam/pvalue.py
+++ b/sdks/python/apache_beam/pvalue.py
@@ -276,8 +276,8 @@ class SingletonPCollectionView(PCollectionView):
return head[0]
else:
raise ValueError(
- 'PCollection with more than one element accessed as '
- 'a singleton view.')
+ 'PCollection with more than one element accessed as '
+ 'a singleton view.')
class IterablePCollectionView(PCollectionView):
http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/82f553e1/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 cd06879..e500fd8 100644
--- a/sdks/python/apache_beam/runners/common.py
+++ b/sdks/python/apache_beam/runners/common.py
@@ -76,17 +76,25 @@ class DoFnRunner(Receiver):
self.dofn = fn
self.dofn_process = fn.process
else:
+ global_window = window.GlobalWindow()
# TODO(robertwb): Remove when all runners pass side input maps.
- # TODO(robertwb): Optimize for global windows case.
- side_inputs = [side_input if isinstance(side_input, sideinputs.SideInputMap)
- else {window.GlobalWindow(): side_input}
+ side_inputs = [side_input
+ if isinstance(side_input, sideinputs.SideInputMap)
+ else {global_window: side_input}
for side_input in side_inputs]
+ if side_inputs and all(side_input.is_globally_windowed()
+ for side_input in side_inputs):
+ args, kwargs = util.insert_values_in_args(
+ args, kwargs, [side_input[global_window]
+ for side_input in side_inputs])
+ side_inputs = []
if side_inputs:
self.has_windowed_side_inputs = True
+
def process(context):
w = context.windows[0]
cur_args, cur_kwargs = util.insert_values_in_args(
- args, kwargs, [side_input[w] for side_input in side_inputs])
+ args, kwargs, [side_input[w] for side_input in side_inputs])
return fn.process(context, *cur_args, **cur_kwargs)
self.dofn_process = process
elif kwargs:
http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/82f553e1/sdks/python/apache_beam/runners/direct_runner.py
----------------------------------------------------------------------
diff --git a/sdks/python/apache_beam/runners/direct_runner.py b/sdks/python/apache_beam/runners/direct_runner.py
index 936043c..c4c52b3 100644
--- a/sdks/python/apache_beam/runners/direct_runner.py
+++ b/sdks/python/apache_beam/runners/direct_runner.py
@@ -30,11 +30,6 @@ import logging
from apache_beam import coders
from apache_beam import error
-from apache_beam.pvalue import DictPCollectionView
-from apache_beam.pvalue import EmptySideInput
-from apache_beam.pvalue import IterablePCollectionView
-from apache_beam.pvalue import ListPCollectionView
-from apache_beam.pvalue import SingletonPCollectionView
from apache_beam.runners.common import DoFnRunner
from apache_beam.runners.common import DoFnState
from apache_beam.runners.runner import PipelineResult
http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/82f553e1/sdks/python/apache_beam/runners/inprocess/inprocess_evaluation_context.py
----------------------------------------------------------------------
diff --git a/sdks/python/apache_beam/runners/inprocess/inprocess_evaluation_context.py b/sdks/python/apache_beam/runners/inprocess/inprocess_evaluation_context.py
index c6bd41f..7af1608 100644
--- a/sdks/python/apache_beam/runners/inprocess/inprocess_evaluation_context.py
+++ b/sdks/python/apache_beam/runners/inprocess/inprocess_evaluation_context.py
@@ -22,11 +22,6 @@ from __future__ import absolute_import
import collections
import threading
-from apache_beam.pvalue import DictPCollectionView
-from apache_beam.pvalue import EmptySideInput
-from apache_beam.pvalue import IterablePCollectionView
-from apache_beam.pvalue import ListPCollectionView
-from apache_beam.pvalue import SingletonPCollectionView
from apache_beam.transforms import sideinputs
from apache_beam.runners.inprocess.clock import Clock
from apache_beam.runners.inprocess.inprocess_watermark_manager import InProcessWatermarkManager
http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/82f553e1/sdks/python/apache_beam/transforms/sideinputs.py
----------------------------------------------------------------------
diff --git a/sdks/python/apache_beam/transforms/sideinputs.py b/sdks/python/apache_beam/transforms/sideinputs.py
index 2079c93..182179a 100644
--- a/sdks/python/apache_beam/transforms/sideinputs.py
+++ b/sdks/python/apache_beam/transforms/sideinputs.py
@@ -170,7 +170,7 @@ def default_window_mapping_fn(target_window_fn):
else:
def map_via_end(source_window):
return list(target_window_fn.assign(
- window.WindowFn.AssignContext(source_window.max_timestamp())))[-1]
+ window.WindowFn.AssignContext(source_window.max_timestamp())))[-1]
return map_via_end
@@ -188,14 +188,17 @@ class SideInputMap(object):
if window not in self._cache:
target_window = self._window_mapping_fn(window)
self._cache[window] = self._view_class.from_iterable(
- _FilteringIterable(self._iterable, target_window), self._view_options)
+ _FilteringIterable(self._iterable, target_window), self._view_options)
return self._cache[window]
+ def is_globally_windowed(self):
+ return self._window_mapping_fn == _global_window_mapping_fn
+
class _FilteringIterable(object):
"""An iterable containing only those values in the given window.
"""
-
+
def __init__(self, iterable, target_window):
self._iterable = iterable
self._target_window = target_window
http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/82f553e1/sdks/python/apache_beam/transforms/sideinputs_test.py
----------------------------------------------------------------------
diff --git a/sdks/python/apache_beam/transforms/sideinputs_test.py b/sdks/python/apache_beam/transforms/sideinputs_test.py
index f84ff57..28e324a 100644
--- a/sdks/python/apache_beam/transforms/sideinputs_test.py
+++ b/sdks/python/apache_beam/transforms/sideinputs_test.py
@@ -101,7 +101,7 @@ class SideInputsTest(unittest.TestCase):
[1, 2, 11],
window.FixedWindows(10),
side_input_type=beam.pvalue.AsIter,
- expected=[(1, [1, 2]), (2, [1, 2]), (11, [11])])
+ expected=[(1, [1, 2]), (2, [1, 2]), (11, [11])])
def test_windowed_singleton(self):
self.run_windowed_side_inputs(
[6/9] incubator-beam git commit: More complicated window tests.
Posted by ro...@apache.org.
More complicated window tests.
Project: http://git-wip-us.apache.org/repos/asf/incubator-beam/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-beam/commit/989a189d
Tree: http://git-wip-us.apache.org/repos/asf/incubator-beam/tree/989a189d
Diff: http://git-wip-us.apache.org/repos/asf/incubator-beam/diff/989a189d
Branch: refs/heads/python-sdk
Commit: 989a189d6495163bb4f80e310834ef927dbefef5
Parents: 7386bcc
Author: Robert Bradshaw <ro...@google.com>
Authored: Thu Oct 13 18:02:12 2016 -0700
Committer: Robert Bradshaw <ro...@google.com>
Committed: Tue Oct 18 12:17:16 2016 -0700
----------------------------------------------------------------------
sdks/python/apache_beam/runners/common.py | 2 +-
.../python/apache_beam/transforms/sideinputs.py | 2 +-
.../apache_beam/transforms/sideinputs_test.py | 69 +++++++++++++++++++-
3 files changed, 69 insertions(+), 4 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/989a189d/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 86fd819..cd06879 100644
--- a/sdks/python/apache_beam/runners/common.py
+++ b/sdks/python/apache_beam/runners/common.py
@@ -150,7 +150,7 @@ class DoFnRunner(Receiver):
if self.has_windowed_side_inputs and len(element.windows) > 1:
for w in element.windows:
self.context.set_element(
- WindowedValue(element.value, element.timestamp, w))
+ WindowedValue(element.value, element.timestamp, (w,)))
self._process_outputs(element, self.dofn_process(self.context))
else:
self.context.set_element(element)
http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/989a189d/sdks/python/apache_beam/transforms/sideinputs.py
----------------------------------------------------------------------
diff --git a/sdks/python/apache_beam/transforms/sideinputs.py b/sdks/python/apache_beam/transforms/sideinputs.py
index 00c2852..2079c93 100644
--- a/sdks/python/apache_beam/transforms/sideinputs.py
+++ b/sdks/python/apache_beam/transforms/sideinputs.py
@@ -170,7 +170,7 @@ def default_window_mapping_fn(target_window_fn):
else:
def map_via_end(source_window):
return list(target_window_fn.assign(
- window.WindowFn.AssignContext(source_window.max_timestamp())))[0]
+ window.WindowFn.AssignContext(source_window.max_timestamp())))[-1]
return map_via_end
http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/989a189d/sdks/python/apache_beam/transforms/sideinputs_test.py
----------------------------------------------------------------------
diff --git a/sdks/python/apache_beam/transforms/sideinputs_test.py b/sdks/python/apache_beam/transforms/sideinputs_test.py
index 8b5b4e0..f84ff57 100644
--- a/sdks/python/apache_beam/transforms/sideinputs_test.py
+++ b/sdks/python/apache_beam/transforms/sideinputs_test.py
@@ -38,9 +38,13 @@ class SideInputsTest(unittest.TestCase):
main = pcoll | 'WindowMain' >> beam.WindowInto(main_window_fn)
side = pcoll | 'WindowSide' >> beam.WindowInto(
side_window_fn or main_window_fn)
+ kw = {}
if combine_fn is not None:
- side |= beam.CombineGlobally(combine_fn)
- res = main | beam.Map(lambda x, s: (x, s), side_input_type(side))
+ side |= beam.CombineGlobally(combine_fn).without_defaults()
+ kw['default_value'] = 0
+ elif side_input_type == beam.pvalue.AsDict:
+ side |= beam.Map(lambda x: ('k%s' % x, 'v%s' % x))
+ res = main | beam.Map(lambda x, s: (x, s), side_input_type(side, **kw))
if side_input_type in (beam.pvalue.AsIter, beam.pvalue.AsList):
res |= beam.Map(lambda (x, s): (x, sorted(s)))
assert_that(res, equal_to(expected))
@@ -57,6 +61,67 @@ class SideInputsTest(unittest.TestCase):
window.FixedWindows(10),
expected=[(1, [1, 2]), (2, [1, 2]), (11, [11])])
+ def test_different_fixed_windows(self):
+ self.run_windowed_side_inputs(
+ [1, 2, 11, 21, 31],
+ window.FixedWindows(10),
+ window.FixedWindows(20),
+ expected=[(1, [1, 2, 11]), (2, [1, 2, 11]), (11, [1, 2, 11]),
+ (21, [21, 31]), (31, [21, 31])])
+
+ def test_fixed_global_window(self):
+ self.run_windowed_side_inputs(
+ [1, 2, 11],
+ window.FixedWindows(10),
+ window.GlobalWindows(),
+ expected=[(1, [1, 2, 11]), (2, [1, 2, 11]), (11, [1, 2, 11])])
+
+ def test_sliding_windows(self):
+ self.run_windowed_side_inputs(
+ [1, 2, 4],
+ window.SlidingWindows(size=6, period=2),
+ window.SlidingWindows(size=6, period=2),
+ expected=[
+ # Element 1 falls in three windows
+ (1, [1]), # [-4, 2)
+ (1, [1, 2]), # [-2, 4)
+ (1, [1, 2, 4]), # [0, 6)
+ # as does 2,
+ (2, [1, 2]), # [-2, 4)
+ (2, [1, 2, 4]), # [0, 6)
+ (2, [2, 4]), # [2, 8)
+ # and 4.
+ (4, [1, 2, 4]), # [0, 6)
+ (4, [2, 4]), # [2, 8)
+ (4, [4]), # [4, 10)
+ ])
+
+ def test_windowed_iter(self):
+ self.run_windowed_side_inputs(
+ [1, 2, 11],
+ window.FixedWindows(10),
+ side_input_type=beam.pvalue.AsIter,
+ expected=[(1, [1, 2]), (2, [1, 2]), (11, [11])])
+
+ def test_windowed_singleton(self):
+ self.run_windowed_side_inputs(
+ [1, 2, 11],
+ window.FixedWindows(10),
+ side_input_type=beam.pvalue.AsSingleton,
+ combine_fn=sum,
+ expected=[(1, 3), (2, 3), (11, 11)])
+
+ def test_windowed_dict(self):
+ self.run_windowed_side_inputs(
+ [1, 2, 11],
+ window.FixedWindows(10),
+ side_input_type=beam.pvalue.AsDict,
+ expected=[
+ (1, {'k1': 'v1', 'k2': 'v2'}),
+ (2, {'k1': 'v1', 'k2': 'v2'}),
+ (11, {'k11': 'v11'}),
+ ])
+
if __name__ == '__main__':
logging.getLogger().setLevel(logging.DEBUG)
[4/9] incubator-beam git commit: Minor fixups for better testing
Posted by ro...@apache.org.
Minor fixups for better testing
Project: http://git-wip-us.apache.org/repos/asf/incubator-beam/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-beam/commit/a50efccc
Tree: http://git-wip-us.apache.org/repos/asf/incubator-beam/tree/a50efccc
Diff: http://git-wip-us.apache.org/repos/asf/incubator-beam/diff/a50efccc
Branch: refs/heads/python-sdk
Commit: a50efccc4b79e95cedf49d2680df67dd2b587927
Parents: 82f553e
Author: Robert Bradshaw <ro...@google.com>
Authored: Mon Oct 17 17:50:32 2016 -0700
Committer: Robert Bradshaw <ro...@google.com>
Committed: Tue Oct 18 12:17:16 2016 -0700
----------------------------------------------------------------------
sdks/python/apache_beam/runners/common.py | 5 +++--
sdks/python/apache_beam/transforms/sideinputs.py | 7 ++++++-
sdks/python/apache_beam/transforms/sideinputs_test.py | 5 ++++-
3 files changed, 13 insertions(+), 4 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/a50efccc/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 e500fd8..cc834ba 100644
--- a/sdks/python/apache_beam/runners/common.py
+++ b/sdks/python/apache_beam/runners/common.py
@@ -82,8 +82,9 @@ class DoFnRunner(Receiver):
if isinstance(side_input, sideinputs.SideInputMap)
else {global_window: side_input}
for side_input in side_inputs]
- if side_inputs and all(side_input.is_globally_windowed()
- for side_input in side_inputs):
+ if side_inputs and all(
+ isinstance(side_input, dict) or side_input.is_globally_windowed()
+ for side_input in side_inputs):
args, kwargs = util.insert_values_in_args(
args, kwargs, [side_input[global_window]
for side_input in side_inputs])
http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/a50efccc/sdks/python/apache_beam/transforms/sideinputs.py
----------------------------------------------------------------------
diff --git a/sdks/python/apache_beam/transforms/sideinputs.py b/sdks/python/apache_beam/transforms/sideinputs.py
index 182179a..f3a7178 100644
--- a/sdks/python/apache_beam/transforms/sideinputs.py
+++ b/sdks/python/apache_beam/transforms/sideinputs.py
@@ -178,7 +178,8 @@ class SideInputMap(object):
"""Represents a mapping of windows to side input values."""
def __init__(self, view_class, view_options, iterable):
- self._window_mapping_fn = view_options['window_mapping_fn']
+ self._window_mapping_fn = view_options.get(
+ 'window_mapping_fn', _global_window_mapping_fn)
self._view_class = view_class
self._view_options = view_options
self._iterable = iterable
@@ -207,3 +208,7 @@ class _FilteringIterable(object):
for wv in self._iterable:
if self._target_window in wv.windows:
yield wv.value
+
+ def __reduce__(self):
+ # Pickle self as an already filtered list.
+ return list, (list(self),)
http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/a50efccc/sdks/python/apache_beam/transforms/sideinputs_test.py
----------------------------------------------------------------------
diff --git a/sdks/python/apache_beam/transforms/sideinputs_test.py b/sdks/python/apache_beam/transforms/sideinputs_test.py
index 28e324a..caf3652 100644
--- a/sdks/python/apache_beam/transforms/sideinputs_test.py
+++ b/sdks/python/apache_beam/transforms/sideinputs_test.py
@@ -27,12 +27,15 @@ from apache_beam.transforms.util import assert_that, equal_to
class SideInputsTest(unittest.TestCase):
+ def create_pipeline(self):
+ return beam.Pipeline('DirectPipelineRunner')
+
def run_windowed_side_inputs(self, elements, main_window_fn,
side_window_fn=None,
side_input_type=beam.pvalue.AsList,
combine_fn=None,
expected=None):
- with beam.Pipeline('DirectPipelineRunner') as p:
+ with self.create_pipeline() as p:
pcoll = p | beam.Create(elements) | beam.Map(
lambda t: window.TimestampedValue(t, t))
main = pcoll | 'WindowMain' >> beam.WindowInto(main_window_fn)
[8/9] incubator-beam git commit: Rename from_iterable to avoid
confusion.
Posted by ro...@apache.org.
Rename from_iterable to avoid confusion.
Project: http://git-wip-us.apache.org/repos/asf/incubator-beam/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-beam/commit/cf15ab51
Tree: http://git-wip-us.apache.org/repos/asf/incubator-beam/tree/cf15ab51
Diff: http://git-wip-us.apache.org/repos/asf/incubator-beam/diff/cf15ab51
Branch: refs/heads/python-sdk
Commit: cf15ab51efa9c8bb21f00eee255f0c57e41a8e7b
Parents: a50efcc
Author: Robert Bradshaw <ro...@google.com>
Authored: Tue Oct 18 12:31:30 2016 -0700
Committer: Robert Bradshaw <ro...@google.com>
Committed: Tue Oct 18 12:31:30 2016 -0700
----------------------------------------------------------------------
sdks/python/apache_beam/pvalue.py | 8 ++++----
sdks/python/apache_beam/transforms/sideinputs.py | 2 +-
2 files changed, 5 insertions(+), 5 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/cf15ab51/sdks/python/apache_beam/pvalue.py
----------------------------------------------------------------------
diff --git a/sdks/python/apache_beam/pvalue.py b/sdks/python/apache_beam/pvalue.py
index e40d746..4cfdfbe 100644
--- a/sdks/python/apache_beam/pvalue.py
+++ b/sdks/python/apache_beam/pvalue.py
@@ -268,7 +268,7 @@ class SingletonPCollectionView(PCollectionView):
return base
@staticmethod
- def from_iterable(it, options):
+ def _from_runtime_iterable(it, options):
head = list(itertools.islice(it, 2))
if len(head) == 0:
return options.get('default', EmptySideInput())
@@ -284,7 +284,7 @@ class IterablePCollectionView(PCollectionView):
"""A PCollectionView that can be treated as an iterable."""
@staticmethod
- def from_iterable(it, options):
+ def _from_runtime_iterable(it, options):
return it
@@ -292,7 +292,7 @@ class ListPCollectionView(PCollectionView):
"""A PCollectionView that can be treated as a list."""
@staticmethod
- def from_iterable(it, options):
+ def _from_runtime_iterable(it, options):
return list(it)
@@ -300,7 +300,7 @@ class DictPCollectionView(PCollectionView):
"""A PCollectionView that can be treated as a dict."""
@staticmethod
- def from_iterable(it, options):
+ def _from_runtime_iterable(it, options):
return dict(it)
http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/cf15ab51/sdks/python/apache_beam/transforms/sideinputs.py
----------------------------------------------------------------------
diff --git a/sdks/python/apache_beam/transforms/sideinputs.py b/sdks/python/apache_beam/transforms/sideinputs.py
index f3a7178..05ba6ab 100644
--- a/sdks/python/apache_beam/transforms/sideinputs.py
+++ b/sdks/python/apache_beam/transforms/sideinputs.py
@@ -188,7 +188,7 @@ class SideInputMap(object):
def __getitem__(self, window):
if window not in self._cache:
target_window = self._window_mapping_fn(window)
- self._cache[window] = self._view_class.from_iterable(
+ self._cache[window] = self._view_class._from_runtime_iterable(
_FilteringIterable(self._iterable, target_window), self._view_options)
return self._cache[window]
[2/9] incubator-beam git commit: Implement windowed side inputs for
direct runner.
Posted by ro...@apache.org.
Implement windowed side inputs for direct runner.
Project: http://git-wip-us.apache.org/repos/asf/incubator-beam/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-beam/commit/29a73789
Tree: http://git-wip-us.apache.org/repos/asf/incubator-beam/tree/29a73789
Diff: http://git-wip-us.apache.org/repos/asf/incubator-beam/diff/29a73789
Branch: refs/heads/python-sdk
Commit: 29a73789c787588dd35edc7964b398961c627cdf
Parents: 9007376
Author: Robert Bradshaw <ro...@google.com>
Authored: Tue Oct 11 15:27:58 2016 -0700
Committer: Robert Bradshaw <ro...@google.com>
Committed: Tue Oct 18 12:17:15 2016 -0700
----------------------------------------------------------------------
sdks/python/apache_beam/pvalue.py | 44 ++++++++++---
sdks/python/apache_beam/runners/common.pxd | 1 +
sdks/python/apache_beam/runners/common.py | 42 ++++++++----
.../python/apache_beam/runners/direct_runner.py | 25 +-------
.../inprocess/inprocess_evaluation_context.py | 8 +--
.../python/apache_beam/transforms/sideinputs.py | 67 +++++++++++++++++---
.../apache_beam/transforms/sideinputs_test.py | 8 ---
sdks/python/apache_beam/transforms/timeutil.py | 4 ++
sdks/python/apache_beam/transforms/window.py | 3 +
9 files changed, 139 insertions(+), 63 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/29a73789/sdks/python/apache_beam/pvalue.py
----------------------------------------------------------------------
diff --git a/sdks/python/apache_beam/pvalue.py b/sdks/python/apache_beam/pvalue.py
index 063d0b5..f8b7feb 100644
--- a/sdks/python/apache_beam/pvalue.py
+++ b/sdks/python/apache_beam/pvalue.py
@@ -27,6 +27,7 @@ produced when the pipeline gets executed.
from __future__ import absolute_import
import collections
+import itertools
class PValue(object):
@@ -227,9 +228,10 @@ class SideOutputValue(object):
class PCollectionView(PValue):
"""An immutable view of a PCollection that can be used as a side input."""
- def __init__(self, pipeline):
+ def __init__(self, pipeline, window_mapping_fn):
"""Initializes a PCollectionView. Do not call directly."""
super(PCollectionView, self).__init__(pipeline)
+ self._window_mapping_fn = window_mapping_fn
@property
def windowing(self):
@@ -246,34 +248,60 @@ class PCollectionView(PValue):
Returns:
Tuple of options for the given view.
"""
- return ()
+ return {'window_mapping_fn': self._window_mapping_fn}
class SingletonPCollectionView(PCollectionView):
"""A PCollectionView that contains a single object."""
- def __init__(self, pipeline, has_default, default_value):
- super(SingletonPCollectionView, self).__init__(pipeline)
+ def __init__(self, pipeline, has_default, default_value,
+ window_mapping_fn):
+ super(SingletonPCollectionView, self).__init__(pipeline, window_mapping_fn)
self.has_default = has_default
self.default_value = default_value
def _view_options(self):
- return (self.has_default, self.default_value)
+ base = super(SingletonPCollectionView, self)._view_options()
+ if self.has_default:
+ return dict(base, default=self.default_value)
+ else:
+ return base
+
+ @staticmethod
+ def from_iterable(it, options):
+ head = list(itertools.islice(it, 2))
+ if len(head) == 0:
+ return options.get('default', EmptySideInput())
+ elif len(head) == 1:
+ return head[0]
+ else:
+ raise ValueError(
+ 'PCollection with more than one element accessed as '
+ 'a singleton view.')
class IterablePCollectionView(PCollectionView):
"""A PCollectionView that can be treated as an iterable."""
- pass
+
+ @staticmethod
+ def from_iterable(it, options):
+ return it
class ListPCollectionView(PCollectionView):
"""A PCollectionView that can be treated as a list."""
- pass
+
+ @staticmethod
+ def from_iterable(it, options):
+ return list(it)
class DictPCollectionView(PCollectionView):
"""A PCollectionView that can be treated as a dict."""
- pass
+
+ @staticmethod
+ def from_iterable(it, options):
+ return dict(it)
def _get_cached_view(pipeline, key):
http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/29a73789/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 5cd4cf8..085fd11 100644
--- a/sdks/python/apache_beam/runners/common.pxd
+++ b/sdks/python/apache_beam/runners/common.pxd
@@ -36,6 +36,7 @@ cdef class DoFnRunner(Receiver):
cdef object tagged_receivers
cdef LoggingContext logging_context
cdef object step_name
+ cdef bint has_windowed_side_inputs
cdef Receiver main_receivers
http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/29a73789/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 67277c3..86fd819 100644
--- a/sdks/python/apache_beam/runners/common.py
+++ b/sdks/python/apache_beam/runners/common.py
@@ -24,6 +24,8 @@ import sys
from apache_beam.internal import util
from apache_beam.pvalue import SideOutputValue
from apache_beam.transforms import core
+from apache_beam.transforms import sideinputs
+from apache_beam.transforms import window
from apache_beam.transforms.window import TimestampedValue
from apache_beam.transforms.window import WindowFn
from apache_beam.utils.windowed_value import WindowedValue
@@ -69,24 +71,36 @@ class DoFnRunner(Receiver):
# Preferred alternative to context
# TODO(robertwb): Remove once all runners are updated.
state=None):
+ self.has_windowed_side_inputs = False # Set to True in one case below.
if not args and not kwargs:
self.dofn = fn
self.dofn_process = fn.process
else:
- args, kwargs = util.insert_values_in_args(args, kwargs, side_inputs)
+ # TODO(robertwb): Remove when all runners pass side input maps.
+ # TODO(robertwb): Optimize for global windows case.
+ side_inputs = [side_input if isinstance(side_input, sideinputs.SideInputMap)
+ else {window.GlobalWindow(): side_input}
+ for side_input in side_inputs]
+ if side_inputs:
+ self.has_windowed_side_inputs = True
+ def process(context):
+ w = context.windows[0]
+ cur_args, cur_kwargs = util.insert_values_in_args(
+ args, kwargs, [side_input[w] for side_input in side_inputs])
+ return fn.process(context, *cur_args, **cur_kwargs)
+ self.dofn_process = process
+ elif kwargs:
+ self.dofn_process = lambda context: fn.process(context, *args, **kwargs)
+ else:
+ self.dofn_process = lambda context: fn.process(context, *args)
class CurriedFn(core.DoFn):
- def start_bundle(self, context):
- return fn.start_bundle(context)
-
- def process(self, context):
- return fn.process(context, *args, **kwargs)
+ start_bundle = staticmethod(fn.start_bundle)
+ process = staticmethod(self.dofn_process)
+ finish_bundle = staticmethod(fn.finish_bundle)
- def finish_bundle(self, context):
- return fn.finish_bundle(context)
self.dofn = CurriedFn()
- self.dofn_process = lambda context: fn.process(context, *args, **kwargs)
self.window_fn = windowing.windowfn
self.tagged_receivers = tagged_receivers
@@ -133,8 +147,14 @@ class DoFnRunner(Receiver):
def process(self, element):
try:
self.logging_context.enter()
- self.context.set_element(element)
- self._process_outputs(element, self.dofn_process(self.context))
+ if self.has_windowed_side_inputs and len(element.windows) > 1:
+ for w in element.windows:
+ self.context.set_element(
+ WindowedValue(element.value, element.timestamp, w))
+ self._process_outputs(element, self.dofn_process(self.context))
+ else:
+ self.context.set_element(element)
+ self._process_outputs(element, self.dofn_process(self.context))
except BaseException as exn:
self.reraise_augmented(exn)
finally:
http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/29a73789/sdks/python/apache_beam/runners/direct_runner.py
----------------------------------------------------------------------
diff --git a/sdks/python/apache_beam/runners/direct_runner.py b/sdks/python/apache_beam/runners/direct_runner.py
index 656cc91..936043c 100644
--- a/sdks/python/apache_beam/runners/direct_runner.py
+++ b/sdks/python/apache_beam/runners/direct_runner.py
@@ -41,6 +41,7 @@ from apache_beam.runners.runner import PipelineResult
from apache_beam.runners.runner import PipelineRunner
from apache_beam.runners.runner import PipelineState
from apache_beam.runners.runner import PValueCache
+from apache_beam.transforms import sideinputs
from apache_beam.transforms.window import GlobalWindows
from apache_beam.transforms.window import WindowedValue
from apache_beam.typehints.typecheck import OutputCheckWrapperDoFn
@@ -107,29 +108,7 @@ class DirectPipelineRunner(PipelineRunner):
transform = transform_node.transform
view = transform.view
values = self._cache.get_pvalue(transform_node.inputs[0])
- if isinstance(view, SingletonPCollectionView):
- has_default, default_value = view._view_options() # pylint: disable=protected-access
- if len(values) == 0:
- if has_default:
- result = default_value
- else:
- result = EmptySideInput()
- elif len(values) == 1:
- # TODO(ccy): Figure out whether side inputs should ever be given as
- # windowed values
- result = values[0].value
- else:
- raise ValueError(('PCollection with more than one element accessed as '
- 'a singleton view: %s.') % view)
- elif isinstance(view, IterablePCollectionView):
- result = [v.value for v in values]
- elif isinstance(view, ListPCollectionView):
- result = [v.value for v in values]
- elif isinstance(view, DictPCollectionView):
- result = dict(v.value for v in values)
- else:
- raise NotImplementedError
-
+ result = sideinputs.SideInputMap(type(view), view._view_options(), values)
self._cache.cache_output(transform_node, result)
@skip_if_cached
http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/29a73789/sdks/python/apache_beam/runners/inprocess/inprocess_evaluation_context.py
----------------------------------------------------------------------
diff --git a/sdks/python/apache_beam/runners/inprocess/inprocess_evaluation_context.py b/sdks/python/apache_beam/runners/inprocess/inprocess_evaluation_context.py
index 9c8b695..883be99 100644
--- a/sdks/python/apache_beam/runners/inprocess/inprocess_evaluation_context.py
+++ b/sdks/python/apache_beam/runners/inprocess/inprocess_evaluation_context.py
@@ -27,6 +27,7 @@ from apache_beam.pvalue import EmptySideInput
from apache_beam.pvalue import IterablePCollectionView
from apache_beam.pvalue import ListPCollectionView
from apache_beam.pvalue import SingletonPCollectionView
+from apache_beam.transforms import sideinputs
from apache_beam.runners.inprocess.clock import Clock
from apache_beam.runners.inprocess.inprocess_watermark_manager import InProcessWatermarkManager
from apache_beam.runners.inprocess.inprocess_executor import TransformExecutor
@@ -106,12 +107,9 @@ class _InProcessSideInputsContainer(object):
ValueError: If values cannot be converted into the requested form.
"""
if isinstance(view, SingletonPCollectionView):
- has_default, default_value = view._view_options() # pylint: disable=protected-access
if len(values) == 0:
- if has_default:
- result = default_value
- else:
- result = EmptySideInput()
+ # pylint: disable=protected-access
+ result = view._view_options().get('default', EmptySideInput())
elif len(values) == 1:
result = values[0].value
else:
http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/29a73789/sdks/python/apache_beam/transforms/sideinputs.py
----------------------------------------------------------------------
diff --git a/sdks/python/apache_beam/transforms/sideinputs.py b/sdks/python/apache_beam/transforms/sideinputs.py
index 6c698da..00c2852 100644
--- a/sdks/python/apache_beam/transforms/sideinputs.py
+++ b/sdks/python/apache_beam/transforms/sideinputs.py
@@ -27,6 +27,7 @@ from __future__ import absolute_import
from apache_beam import pvalue
from apache_beam import typehints
from apache_beam.transforms.ptransform import PTransform
+from apache_beam.transforms import window
# Type variables
K = typehints.TypeVariable('K')
@@ -50,10 +51,6 @@ class CreatePCollectionView(PTransform):
return input_type
def apply(self, pcoll):
- if not pcoll.windowing.is_default():
- raise ValueError(
- "Side inputs only supported for global windows, default triggering. "
- "Found %s" % pcoll.windowing)
return self.view
@@ -78,7 +75,8 @@ class ViewAsSingleton(PTransform):
return (pcoll
| CreatePCollectionView(
pvalue.SingletonPCollectionView(
- pcoll.pipeline, self.has_default, self.default_value))
+ pcoll.pipeline, self.has_default, self.default_value,
+ default_window_mapping_fn(pcoll.windowing.windowfn)))
.with_input_types(input_type)
.with_output_types(output_type))
@@ -101,7 +99,9 @@ class ViewAsIterable(PTransform):
output_type = typehints.Iterable[input_type]
return (pcoll
| CreatePCollectionView(
- pvalue.IterablePCollectionView(pcoll.pipeline))
+ pvalue.IterablePCollectionView(
+ pcoll.pipeline,
+ default_window_mapping_fn(pcoll.windowing.windowfn)))
.with_input_types(input_type)
.with_output_types(output_type))
@@ -123,7 +123,9 @@ class ViewAsList(PTransform):
input_type = pcoll.element_type
output_type = typehints.List[input_type]
return (pcoll
- | CreatePCollectionView(pvalue.ListPCollectionView(pcoll.pipeline))
+ | CreatePCollectionView(pvalue.ListPCollectionView(
+ pcoll.pipeline,
+ default_window_mapping_fn(pcoll.windowing.windowfn)))
.with_input_types(input_type)
.with_output_types(output_type))
@@ -150,6 +152,55 @@ class ViewAsDict(PTransform):
output_type = typehints.Dict[key_type, value_type]
return (pcoll
| CreatePCollectionView(
- pvalue.DictPCollectionView(pcoll.pipeline))
+ pvalue.DictPCollectionView(
+ pcoll.pipeline,
+ default_window_mapping_fn(pcoll.windowing.windowfn)))
.with_input_types(input_type)
.with_output_types(output_type))
+
+
+# Top-level function so we can identify it later.
+def _global_window_mapping_fn(w, global_window=window.GlobalWindow()):
+ return global_window
+
+
+def default_window_mapping_fn(target_window_fn):
+ if target_window_fn == window.GlobalWindows():
+ return _global_window_mapping_fn
+ else:
+ def map_via_end(source_window):
+ return list(target_window_fn.assign(
+ window.WindowFn.AssignContext(source_window.max_timestamp())))[0]
+ return map_via_end
+
+
+class SideInputMap(object):
+ """Represents a mapping of windows to side input values."""
+
+ def __init__(self, view_class, view_options, iterable):
+ self._window_mapping_fn = view_options['window_mapping_fn']
+ self._view_class = view_class
+ self._view_options = view_options
+ self._iterable = iterable
+ self._cache = {}
+
+ def __getitem__(self, window):
+ if window not in self._cache:
+ target_window = self._window_mapping_fn(window)
+ self._cache[window] = self._view_class.from_iterable(
+ _FilteringIterable(self._iterable, target_window), self._view_options)
+ return self._cache[window]
+
+
+class _FilteringIterable(object):
+ """An iterable containing only those values in the given window.
+ """
+
+ def __init__(self, iterable, target_window):
+ self._iterable = iterable
+ self._target_window = target_window
+
+ def __iter__(self):
+ for wv in self._iterable:
+ if self._target_window in wv.windows:
+ yield wv.value
http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/29a73789/sdks/python/apache_beam/transforms/sideinputs_test.py
----------------------------------------------------------------------
diff --git a/sdks/python/apache_beam/transforms/sideinputs_test.py b/sdks/python/apache_beam/transforms/sideinputs_test.py
index 68deba8..8b5b4e0 100644
--- a/sdks/python/apache_beam/transforms/sideinputs_test.py
+++ b/sdks/python/apache_beam/transforms/sideinputs_test.py
@@ -27,14 +27,6 @@ from apache_beam.transforms.util import assert_that, equal_to
class SideInputsTest(unittest.TestCase):
- # TODO(BEAM-733): Actually support this.
- def test_no_sideinput_windowing(self):
- p = beam.Pipeline('DirectPipelineRunner')
- pc = p | beam.Create([0, 1]) | beam.WindowInto(window.FixedWindows(10))
- with self.assertRaises(ValueError):
- # pylint: disable=expression-not-assigned
- pc | beam.Map(lambda x, side: None, side=beam.pvalue.AsIter(pc))
-
def run_windowed_side_inputs(self, elements, main_window_fn,
side_window_fn=None,
side_input_type=beam.pvalue.AsList,
http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/29a73789/sdks/python/apache_beam/transforms/timeutil.py
----------------------------------------------------------------------
diff --git a/sdks/python/apache_beam/transforms/timeutil.py b/sdks/python/apache_beam/transforms/timeutil.py
index 4092b60..f026d83 100644
--- a/sdks/python/apache_beam/transforms/timeutil.py
+++ b/sdks/python/apache_beam/transforms/timeutil.py
@@ -58,6 +58,10 @@ class Timestamp(object):
return seconds
return Timestamp(seconds)
+ def predecessor(self):
+ """Returns the largest timestamp smaller than self."""
+ return Timestamp(micros=self.micros - 1)
+
def __repr__(self):
micros = self.micros
sign = ''
http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/29a73789/sdks/python/apache_beam/transforms/window.py
----------------------------------------------------------------------
diff --git a/sdks/python/apache_beam/transforms/window.py b/sdks/python/apache_beam/transforms/window.py
index e07814d..9485032 100644
--- a/sdks/python/apache_beam/transforms/window.py
+++ b/sdks/python/apache_beam/transforms/window.py
@@ -143,6 +143,9 @@ class BoundedWindow(object):
def __init__(self, end):
self.end = Timestamp.of(end)
+ def max_timestamp(self):
+ return self.end.predecessor()
+
def __cmp__(self, other):
# Order first by endpoint, then arbitrarily.
return cmp(self.end, other.end) or cmp(hash(self), hash(other))
[9/9] incubator-beam git commit: Closes #1100
Posted by ro...@apache.org.
Closes #1100
Project: http://git-wip-us.apache.org/repos/asf/incubator-beam/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-beam/commit/8e1793ca
Tree: http://git-wip-us.apache.org/repos/asf/incubator-beam/tree/8e1793ca
Diff: http://git-wip-us.apache.org/repos/asf/incubator-beam/diff/8e1793ca
Branch: refs/heads/python-sdk
Commit: 8e1793cafa1839456ae007937bc581764749a006
Parents: 6e6d89d cf15ab5
Author: Robert Bradshaw <ro...@google.com>
Authored: Tue Oct 18 13:12:56 2016 -0700
Committer: Robert Bradshaw <ro...@google.com>
Committed: Tue Oct 18 13:12:56 2016 -0700
----------------------------------------------------------------------
.../examples/cookbook/bigquery_side_input.py | 3 +-
sdks/python/apache_beam/pvalue.py | 44 ++++++--
sdks/python/apache_beam/runners/common.pxd | 1 +
sdks/python/apache_beam/runners/common.py | 51 +++++++--
.../python/apache_beam/runners/direct_runner.py | 30 +-----
.../inprocess/inprocess_evaluation_context.py | 29 +----
.../python/apache_beam/transforms/sideinputs.py | 75 +++++++++++--
.../apache_beam/transforms/sideinputs_test.py | 105 +++++++++++++++++--
sdks/python/apache_beam/transforms/timeutil.py | 4 +
sdks/python/apache_beam/transforms/util.py | 5 +-
sdks/python/apache_beam/transforms/window.py | 3 +
.../transforms/write_ptransform_test.py | 2 +-
12 files changed, 258 insertions(+), 94 deletions(-)
----------------------------------------------------------------------
[3/9] incubator-beam git commit: Windowed side input test.
Posted by ro...@apache.org.
Windowed side input test.
Project: http://git-wip-us.apache.org/repos/asf/incubator-beam/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-beam/commit/90073768
Tree: http://git-wip-us.apache.org/repos/asf/incubator-beam/tree/90073768
Diff: http://git-wip-us.apache.org/repos/asf/incubator-beam/diff/90073768
Branch: refs/heads/python-sdk
Commit: 90073768985ad492e5c0dccf1114deefcae06b13
Parents: 6e6d89d
Author: Robert Bradshaw <ro...@google.com>
Authored: Tue Oct 11 16:43:41 2016 -0700
Committer: Robert Bradshaw <ro...@google.com>
Committed: Tue Oct 18 12:17:15 2016 -0700
----------------------------------------------------------------------
.../apache_beam/transforms/sideinputs_test.py | 31 ++++++++++++++++++++
1 file changed, 31 insertions(+)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/90073768/sdks/python/apache_beam/transforms/sideinputs_test.py
----------------------------------------------------------------------
diff --git a/sdks/python/apache_beam/transforms/sideinputs_test.py b/sdks/python/apache_beam/transforms/sideinputs_test.py
index 8e292e3..68deba8 100644
--- a/sdks/python/apache_beam/transforms/sideinputs_test.py
+++ b/sdks/python/apache_beam/transforms/sideinputs_test.py
@@ -22,6 +22,7 @@ import unittest
import apache_beam as beam
from apache_beam.transforms import window
+from apache_beam.transforms.util import assert_that, equal_to
class SideInputsTest(unittest.TestCase):
@@ -34,6 +35,36 @@ class SideInputsTest(unittest.TestCase):
# pylint: disable=expression-not-assigned
pc | beam.Map(lambda x, side: None, side=beam.pvalue.AsIter(pc))
+ def run_windowed_side_inputs(self, elements, main_window_fn,
+ side_window_fn=None,
+ side_input_type=beam.pvalue.AsList,
+ combine_fn=None,
+ expected=None):
+ with beam.Pipeline('DirectPipelineRunner') as p:
+ pcoll = p | beam.Create(elements) | beam.Map(
+ lambda t: window.TimestampedValue(t, t))
+ main = pcoll | 'WindowMain' >> beam.WindowInto(main_window_fn)
+ side = pcoll | 'WindowSide' >> beam.WindowInto(
+ side_window_fn or main_window_fn)
+ if combine_fn is not None:
+ side |= beam.CombineGlobally(combine_fn)
+ res = main | beam.Map(lambda x, s: (x, s), side_input_type(side))
+ if side_input_type in (beam.pvalue.AsIter, beam.pvalue.AsList):
+ res |= beam.Map(lambda (x, s): (x, sorted(s)))
+ assert_that(res, equal_to(expected))
+
+ def test_global_global_windows(self):
+ self.run_windowed_side_inputs(
+ [1, 2, 3],
+ window.GlobalWindows(),
+ expected=[(1, [1, 2, 3]), (2, [1, 2, 3]), (3, [1, 2, 3])])
+
+ def test_same_fixed_windows(self):
+ self.run_windowed_side_inputs(
+ [1, 2, 11],
+ window.FixedWindows(10),
+ expected=[(1, [1, 2]), (2, [1, 2]), (11, [11])])
+
if __name__ == '__main__':
logging.getLogger().setLevel(logging.DEBUG)