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/20 20:06:47 UTC
[1/5] incubator-beam git commit: Closes #691
Repository: incubator-beam
Updated Branches:
refs/heads/python-sdk a643e2009 -> 37f426faf
Closes #691
Project: http://git-wip-us.apache.org/repos/asf/incubator-beam/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-beam/commit/37f426fa
Tree: http://git-wip-us.apache.org/repos/asf/incubator-beam/tree/37f426fa
Diff: http://git-wip-us.apache.org/repos/asf/incubator-beam/diff/37f426fa
Branch: refs/heads/python-sdk
Commit: 37f426faf5c47d5b99d100292fd0cde6a640939e
Parents: a643e20 9d1fc85
Author: Robert Bradshaw <ro...@google.com>
Authored: Wed Jul 20 13:06:21 2016 -0700
Committer: Robert Bradshaw <ro...@google.com>
Committed: Wed Jul 20 13:06:21 2016 -0700
----------------------------------------------------------------------
sdks/python/apache_beam/pipeline_test.py | 18 +--
sdks/python/apache_beam/runners/common.pxd | 20 ++-
sdks/python/apache_beam/runners/common.py | 35 +++++-
sdks/python/apache_beam/transforms/window.py | 37 +-----
.../python/apache_beam/utils/windowed_value.pxd | 38 ++++++
sdks/python/apache_beam/utils/windowed_value.py | 122 +++++++++++++++++++
sdks/python/setup.py | 1 +
7 files changed, 223 insertions(+), 48 deletions(-)
----------------------------------------------------------------------
[5/5] incubator-beam git commit: Add Cython DoFnContext and Receiver
stubs.
Posted by ro...@apache.org.
Add Cython DoFnContext and Receiver stubs.
Project: http://git-wip-us.apache.org/repos/asf/incubator-beam/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-beam/commit/8efc231c
Tree: http://git-wip-us.apache.org/repos/asf/incubator-beam/tree/8efc231c
Diff: http://git-wip-us.apache.org/repos/asf/incubator-beam/diff/8efc231c
Branch: refs/heads/python-sdk
Commit: 8efc231c867fcf5267ba21a8a33ee306b87d9945
Parents: dc42467
Author: Robert Bradshaw <ro...@google.com>
Authored: Tue Jul 19 12:22:08 2016 -0700
Committer: Robert Bradshaw <ro...@google.com>
Committed: Wed Jul 20 13:06:21 2016 -0700
----------------------------------------------------------------------
sdks/python/apache_beam/runners/common.pxd | 20 ++++++++++++---
sdks/python/apache_beam/runners/common.py | 34 ++++++++++++++++++++++++-
2 files changed, 50 insertions(+), 4 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/8efc231c/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 480c056..e855376 100644
--- a/sdks/python/apache_beam/runners/common.pxd
+++ b/sdks/python/apache_beam/runners/common.pxd
@@ -15,17 +15,31 @@
# limitations under the License.
#
-cdef type SideOutputValue, TimestampedValue, WindowedValue
+from apache_beam.utils.windowed_value cimport WindowedValue
+
+cdef type SideOutputValue, TimestampedValue
+
cdef class DoFnRunner(object):
cdef object dofn
cdef object window_fn
- cdef object context
+ cdef object context # TODO(robertwb): Make this a DoFnContext
cdef object tagged_receivers
cdef object logger
cdef object step_name
- cdef object main_receivers
+ cdef object main_receivers # TODO(robertwb): Make this a Receiver
cpdef _process_outputs(self, element, results)
+
+
+cdef class DoFnContext(object):
+ cdef object label
+ cdef object state
+ cdef WindowedValue windowed_value
+ cdef set_element(self, WindowedValue windowed_value)
+
+
+cdef class Receiver(object):
+ cdef receive(self, WindowedValue windowed_value)
http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/8efc231c/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 3c0c3f6..059359c 100644
--- a/sdks/python/apache_beam/runners/common.py
+++ b/sdks/python/apache_beam/runners/common.py
@@ -25,8 +25,8 @@ from apache_beam.internal import util
from apache_beam.pvalue import SideOutputValue
from apache_beam.transforms import core
from apache_beam.transforms.window import TimestampedValue
-from apache_beam.transforms.window import WindowedValue
from apache_beam.transforms.window import WindowFn
+from apache_beam.utils.windowed_value import WindowedValue
class FakeLogger(object):
@@ -188,3 +188,35 @@ class DoFnState(object):
"""Looks up the counter for this aggregator, creating one if necessary."""
return self._counter_factory.get_aggregator_counter(
self.step_name, aggregator)
+
+
+class DoFnContext(object):
+
+ def __init__(self, label, element=None, state=None):
+ self.label = label
+ self.state = state
+ if element is not None:
+ self.set_element(element)
+
+ def set_element(self, windowed_value):
+ self.windowed_value = windowed_value
+
+ @property
+ def element(self):
+ return self.windowed_value.value
+
+ @property
+ def timestamp(self):
+ return self.windowed_value.timestamp
+
+ @property
+ def windows(self):
+ return self.windowed_value.windows
+
+ 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/5] incubator-beam git commit: Reduce the number of elements in the
pvalue caching test.
Posted by ro...@apache.org.
Reduce the number of elements in the pvalue caching test.
It seems this test is causing travis-ci to time out,
as the non-compiled version got slightly slower.
100,000 elements should be sufficient to see the effects
of not caching.
Project: http://git-wip-us.apache.org/repos/asf/incubator-beam/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-beam/commit/27689819
Tree: http://git-wip-us.apache.org/repos/asf/incubator-beam/tree/27689819
Diff: http://git-wip-us.apache.org/repos/asf/incubator-beam/diff/27689819
Branch: refs/heads/python-sdk
Commit: 276898195819329a1b88002a41a109df61530a26
Parents: 8efc231
Author: Robert Bradshaw <ro...@google.com>
Authored: Wed Jul 20 01:37:49 2016 -0700
Committer: Robert Bradshaw <ro...@google.com>
Committed: Wed Jul 20 13:06:21 2016 -0700
----------------------------------------------------------------------
sdks/python/apache_beam/pipeline_test.py | 18 ++++++++++--------
1 file changed, 10 insertions(+), 8 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/27689819/sdks/python/apache_beam/pipeline_test.py
----------------------------------------------------------------------
diff --git a/sdks/python/apache_beam/pipeline_test.py b/sdks/python/apache_beam/pipeline_test.py
index c1db5cb..86ae45f 100644
--- a/sdks/python/apache_beam/pipeline_test.py
+++ b/sdks/python/apache_beam/pipeline_test.py
@@ -207,11 +207,13 @@ class PipelineTest(unittest.TestCase):
yield o
yield SideOutputValue('side', o)
+ num_elements = 100000
+
pipeline = Pipeline('DirectPipelineRunner')
gc.collect()
count_threshold = len(gc.get_objects()) + 10000
- biglist = pipeline | Create('oom:create', ['x'] * 1000000)
+ biglist = pipeline | Create('oom:create', ['x'] * num_elements)
dupes = (
biglist
| Map('oom:addone', lambda x: (x, 1))
@@ -223,17 +225,17 @@ class PipelineTest(unittest.TestCase):
| CombinePerKey('oom:combine', sum)
| Map('oom:check', check_memory, count_threshold))
- assert_that(result, equal_to([('x', 3000000)]))
+ assert_that(result, equal_to([('x', 3 * num_elements)]))
pipeline.run()
self.assertEqual(
pipeline.runner.debug_counters['element_counts'],
{
- 'oom:flatten': 3000000,
- ('oom:combine/GroupByKey/reify_windows', None): 3000000,
- ('oom:dupes/oom:dupes', 'side'): 1000000,
- ('oom:dupes/oom:dupes', None): 1000000,
- 'oom:create': 1000000,
- ('oom:addone', None): 1000000,
+ 'oom:flatten': 3 * num_elements,
+ ('oom:combine/GroupByKey/reify_windows', None): 3 * num_elements,
+ ('oom:dupes/oom:dupes', 'side'): num_elements,
+ ('oom:dupes/oom:dupes', None): num_elements,
+ 'oom:create': num_elements,
+ ('oom:addone', None): num_elements,
'oom:combine/GroupByKey/group_by_key': 1,
('oom:check', None): 1,
'assert_that/singleton': 1,
[3/5] incubator-beam git commit: Cythonize WindowedValue class.
Posted by ro...@apache.org.
Cythonize WindowedValue 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/dc42467f
Tree: http://git-wip-us.apache.org/repos/asf/incubator-beam/tree/dc42467f
Diff: http://git-wip-us.apache.org/repos/asf/incubator-beam/diff/dc42467f
Branch: refs/heads/python-sdk
Commit: dc42467f98a31c21b02abbbbe9851c6272d40d13
Parents: a643e20
Author: Robert Bradshaw <ro...@google.com>
Authored: Tue Jul 19 12:19:47 2016 -0700
Committer: Robert Bradshaw <ro...@google.com>
Committed: Wed Jul 20 13:06:21 2016 -0700
----------------------------------------------------------------------
sdks/python/apache_beam/transforms/window.py | 37 +-----
.../python/apache_beam/utils/windowed_value.pxd | 38 +++++++
sdks/python/apache_beam/utils/windowed_value.py | 114 +++++++++++++++++++
sdks/python/setup.py | 1 +
4 files changed, 154 insertions(+), 36 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/dc42467f/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 eec89c0..e07814d 100644
--- a/sdks/python/apache_beam/transforms/window.py
+++ b/sdks/python/apache_beam/transforms/window.py
@@ -55,6 +55,7 @@ from apache_beam.transforms.timeutil import Duration
from apache_beam.transforms.timeutil import MAX_TIMESTAMP
from apache_beam.transforms.timeutil import MIN_TIMESTAMP
from apache_beam.transforms.timeutil import Timestamp
+from apache_beam.utils.windowed_value import WindowedValue
# TODO(ccy): revisit naming and semantics once Java Apache Beam finalizes their
@@ -185,42 +186,6 @@ class IntervalWindow(BoundedWindow):
min(self.start, other.start), max(self.end, other.end))
-class WindowedValue(object):
- """A windowed value having a value, a timestamp and set of windows.
-
- Attributes:
- value: The underlying value of a windowed value.
- timestamp: Timestamp associated with the value as seconds since Unix epoch.
- windows: A set (iterable) of window objects for the value. The window
- object are descendants of the BoundedWindow class.
- """
-
- def __init__(self, value, timestamp, windows):
- self.value = value
- self.timestamp = Timestamp.of(timestamp)
- self.windows = windows
-
- def __repr__(self):
- return '(%s, %s, %s)' % (
- repr(self.value),
- 'MIN_TIMESTAMP' if self.timestamp == MIN_TIMESTAMP else
- 'MAX_TIMESTAMP' if self.timestamp == MAX_TIMESTAMP else
- float(self.timestamp),
- self.windows)
-
- def __hash__(self):
- return hash((self.value, self.timestamp, self.windows))
-
- def __eq__(self, other):
- return (type(self) == type(other)
- and self.value == other.value
- and self.timestamp == other.timestamp
- and self.windows == other.windows)
-
- def with_value(self, new_value):
- return WindowedValue(new_value, self.timestamp, self.windows)
-
-
class TimestampedValue(object):
"""A timestamped value having a value and a timestamp.
http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/dc42467f/sdks/python/apache_beam/utils/windowed_value.pxd
----------------------------------------------------------------------
diff --git a/sdks/python/apache_beam/utils/windowed_value.pxd b/sdks/python/apache_beam/utils/windowed_value.pxd
new file mode 100644
index 0000000..8799914
--- /dev/null
+++ b/sdks/python/apache_beam/utils/windowed_value.pxd
@@ -0,0 +1,38 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+
+cimport cython
+from libc.stdint cimport int64_t
+
+
+cdef type Timestamp
+
+@cython.final
+cdef class WindowedValue(object):
+ cdef public object value
+ cdef public object windows
+ cdef public int64_t timestamp_micros
+ cdef object timestamp_object
+
+ cpdef WindowedValue with_value(self, new_value)
+
+ @staticmethod
+ cdef inline bint _typed_eq(WindowedValue left, WindowedValue right) except? -2
+
+@cython.locals(wv=WindowedValue)
+cdef inline WindowedValue create(
+ object value, int64_t timestamp_micros, object windows)
http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/dc42467f/sdks/python/apache_beam/utils/windowed_value.py
----------------------------------------------------------------------
diff --git a/sdks/python/apache_beam/utils/windowed_value.py b/sdks/python/apache_beam/utils/windowed_value.py
new file mode 100644
index 0000000..5172b36
--- /dev/null
+++ b/sdks/python/apache_beam/utils/windowed_value.py
@@ -0,0 +1,114 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+
+"""Core windowing data structures.
+"""
+
+# This module is carefully crafted to have optimal performance when
+# compiled whiel still being valid Python. Care needs to be taken when
+# editing this file as WindowedValues are created for every element for
+# every step in a Beam pipeline.
+
+#cython: profile=True
+
+from apache_beam.transforms.timeutil import MAX_TIMESTAMP
+from apache_beam.transforms.timeutil import MIN_TIMESTAMP
+from apache_beam.transforms.timeutil import Timestamp
+
+
+class WindowedValue(object):
+ """A windowed value having a value, a timestamp and set of windows.
+
+ Attributes:
+ value: The underlying value of a windowed value.
+ timestamp: Timestamp associated with the value as seconds since Unix epoch.
+ windows: A set (iterable) of window objects for the value. The window
+ object are descendants of the BoundedWindow class.
+ """
+
+ def __init__(self, value, timestamp, windows):
+ # For performance reasons, only timestamp_micros is stored by default
+ # (as a C int). The Timestamp object is created on demand below.
+ self.value = value
+ if isinstance(timestamp, int):
+ self.timestamp_micros = timestamp * 1000000
+ else:
+ self.timestamp_object = (timestamp if isinstance(timestamp, Timestamp)
+ else Timestamp.of(timestamp))
+ self.timestamp_micros = self.timestamp_object.micros
+ self.windows = windows
+
+ @property
+ def timestamp(self):
+ if self.timestamp_object is None:
+ self.timestamp_object = Timestamp(0, self.timestamp_micros)
+ return self.timestamp_object
+
+ def __repr__(self):
+ return '(%s, %s, %s)' % (
+ repr(self.value),
+ 'MIN_TIMESTAMP' if self.timestamp == MIN_TIMESTAMP else
+ 'MAX_TIMESTAMP' if self.timestamp == MAX_TIMESTAMP else
+ float(self.timestamp),
+ self.windows)
+
+ def __hash__(self):
+ return hash(self.value) + 3 * self.timestamp_micros + 7 * hash(self.windows)
+
+ # We'd rather implement __eq__, but Cython supports that via __richcmp__
+ # instead. Fortunately __cmp__ is understood by both (but not by Python 3).
+ def __cmp__(left, right): # pylint: disable=no-self-argument
+ if type(left) is not type(right):
+ return cmp(type(left), type(right))
+ else:
+ # Don't bother paying the cost of a total ordering.
+ # TODO(robertwb): Avoid the type checks?
+ return not WindowedValue._typed_eq(left, right)
+
+ @staticmethod
+ def _typed_eq(left, right):
+ if (left.timestamp_micros == right.timestamp_micros
+ and left.value == right.value
+ and left.windows == right.windows):
+ return True
+ else:
+ return False
+
+ def with_value(self, new_value):
+ """Creates a new WindowedValue with the same timestamps and windows as this.
+
+ This is the fasted way to create a new WindowedValue.
+ """
+ return create(new_value, self.timestamp_micros, self.windows)
+
+ def __reduce__(self):
+ return WindowedValue, (self.value, self.timestamp, self.windows)
+
+
+# TODO(robertwb): Move this to a static method.
+def create(value, timestamp_micros, windows):
+ wv = WindowedValue.__new__(WindowedValue)
+ wv.value = value
+ wv.timestamp_micros = timestamp_micros
+ wv.windows = windows
+ return wv
+
+
+try:
+ WindowedValue.timestamp_object = None
+except TypeError:
+ pass # Cythonized class already has this default value.
http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/dc42467f/sdks/python/setup.py
----------------------------------------------------------------------
diff --git a/sdks/python/setup.py b/sdks/python/setup.py
index 2287b8e..da2df87 100644
--- a/sdks/python/setup.py
+++ b/sdks/python/setup.py
@@ -77,6 +77,7 @@ setuptools.setup(
'apache_beam/runners/common.py',
'apache_beam/transforms/cy_combiners.py',
'apache_beam/utils/counters.py',
+ 'apache_beam/utils/windowed_value.py',
]),
install_requires=REQUIRED_PACKAGES,
package_data={'': ['*.pyx', '*.pxd']},
[4/5] incubator-beam git commit: Clarifying comments.
Posted by ro...@apache.org.
Clarifying comments.
Project: http://git-wip-us.apache.org/repos/asf/incubator-beam/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-beam/commit/9d1fc85d
Tree: http://git-wip-us.apache.org/repos/asf/incubator-beam/tree/9d1fc85d
Diff: http://git-wip-us.apache.org/repos/asf/incubator-beam/diff/9d1fc85d
Branch: refs/heads/python-sdk
Commit: 9d1fc85d5352f16ce8f0f092865decc6b296dbb8
Parents: 2768981
Author: Robert Bradshaw <ro...@google.com>
Authored: Wed Jul 20 12:24:18 2016 -0700
Committer: Robert Bradshaw <ro...@google.com>
Committed: Wed Jul 20 13:06:21 2016 -0700
----------------------------------------------------------------------
sdks/python/apache_beam/runners/common.py | 1 +
sdks/python/apache_beam/utils/windowed_value.py | 12 ++++++++++--
2 files changed, 11 insertions(+), 2 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/9d1fc85d/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 059359c..ef28c63 100644
--- a/sdks/python/apache_beam/runners/common.py
+++ b/sdks/python/apache_beam/runners/common.py
@@ -190,6 +190,7 @@ class DoFnState(object):
self.step_name, aggregator)
+# TODO(robertwb): Replace core.DoFnContext with this.
class DoFnContext(object):
def __init__(self, label, element=None, state=None):
http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/9d1fc85d/sdks/python/apache_beam/utils/windowed_value.py
----------------------------------------------------------------------
diff --git a/sdks/python/apache_beam/utils/windowed_value.py b/sdks/python/apache_beam/utils/windowed_value.py
index 5172b36..4c50e72 100644
--- a/sdks/python/apache_beam/utils/windowed_value.py
+++ b/sdks/python/apache_beam/utils/windowed_value.py
@@ -72,11 +72,16 @@ class WindowedValue(object):
# We'd rather implement __eq__, but Cython supports that via __richcmp__
# instead. Fortunately __cmp__ is understood by both (but not by Python 3).
def __cmp__(left, right): # pylint: disable=no-self-argument
+ """Compares left and right for equality.
+
+ For performance reasons, doesn't actually impose an ordering
+ on unequal values (always returning 1).
+ """
if type(left) is not type(right):
return cmp(type(left), type(right))
else:
- # Don't bother paying the cost of a total ordering.
# TODO(robertwb): Avoid the type checks?
+ # Returns False (0) if equal, and True (1) if not.
return not WindowedValue._typed_eq(left, right)
@staticmethod
@@ -111,4 +116,7 @@ def create(value, timestamp_micros, windows):
try:
WindowedValue.timestamp_object = None
except TypeError:
- pass # Cythonized class already has this default value.
+ # When we're compiled, we can't dynamically add attributes to
+ # the cdef class, but in this case it's OK as it's already present
+ # on each instance.
+ pass