beam-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From rober...@apache.org
Subject [1/3] incubator-beam git commit: Implement add_input for all CombineFns.
Date Wed, 03 Aug 2016 20:53:25 GMT
Repository: incubator-beam
Updated Branches:
  refs/heads/python-sdk e834fa82b -> 65152cab8


Implement add_input for all CombineFns.


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

Branch: refs/heads/python-sdk
Commit: 3ebf28c6e0d17af3720076e33f88a0f126a89059
Parents: e834fa8
Author: Robert Bradshaw <robertwb@gmail.com>
Authored: Tue Jul 26 01:15:55 2016 -0700
Committer: Robert Bradshaw <robertwb@gmail.com>
Committed: Tue Aug 2 15:52:28 2016 -0700

----------------------------------------------------------------------
 sdks/python/apache_beam/transforms/combiners.py | 16 ++++++++--------
 sdks/python/apache_beam/transforms/core.py      |  6 +++---
 2 files changed, 11 insertions(+), 11 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/3ebf28c6/sdks/python/apache_beam/transforms/combiners.py
----------------------------------------------------------------------
diff --git a/sdks/python/apache_beam/transforms/combiners.py b/sdks/python/apache_beam/transforms/combiners.py
index 155dcc6..c3f0da1 100644
--- a/sdks/python/apache_beam/transforms/combiners.py
+++ b/sdks/python/apache_beam/transforms/combiners.py
@@ -132,6 +132,9 @@ class CountCombineFn(core.CombineFn):
   def create_accumulator(self):
     return 0
 
+  def add_input(self, accumulator, element):
+    return accumulator + 1
+
   def add_inputs(self, accumulator, elements):
     return accumulator + len(elements)
 
@@ -425,9 +428,9 @@ class _TupleCombineFnBase(core.CombineFn):
 
 class TupleCombineFn(_TupleCombineFnBase):
 
-  def add_inputs(self, accumulator, elements):
-    return [c.add_inputs(a, e)
-            for c, a, e in zip(self._combiners, accumulator, zip(*elements))]
+  def add_input(self, accumulator, element):
+    return [c.add_input(a, e)
+            for c, a, e in zip(self._combiners, accumulator, element)]
 
   def with_common_input(self):
     return SingleInputTupleCombineFn(*self._combiners)
@@ -435,8 +438,8 @@ class TupleCombineFn(_TupleCombineFnBase):
 
 class SingleInputTupleCombineFn(_TupleCombineFnBase):
 
-  def add_inputs(self, accumulator, elements):
-    return [c.add_inputs(a, elements)
+  def add_input(self, accumulator, element):
+    return [c.add_input(a, element)
             for c, a in zip(self._combiners, accumulator)]
 
 
@@ -522,9 +525,6 @@ def curry_combine_fn(fn, args, kwargs):
       def add_input(self, accumulator, element):
         return fn.add_input(accumulator, element, *args, **kwargs)
 
-      def add_inputs(self, accumulator, elements):
-        return fn.add_inputs(accumulator, elements, *args, **kwargs)
-
       def merge_accumulators(self, accumulators):
         return fn.merge_accumulators(accumulators, *args, **kwargs)
 

http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/3ebf28c6/sdks/python/apache_beam/transforms/core.py
----------------------------------------------------------------------
diff --git a/sdks/python/apache_beam/transforms/core.py b/sdks/python/apache_beam/transforms/core.py
index 38b9cd2..da26205 100644
--- a/sdks/python/apache_beam/transforms/core.py
+++ b/sdks/python/apache_beam/transforms/core.py
@@ -270,7 +270,7 @@ class CombineFn(WithTypeHints):
   1. Input values are partitioned into one or more batches.
   2. For each batch, the create_accumulator method is invoked to create a fresh
      initial "accumulator" value representing the combination of zero values.
-  3. For each input value in the batch, the add_inputs method is invoked to
+  3. For each input value in the batch, the add_input method is invoked to
      combine more values with the accumulator for that batch.
   4. The merge_accumulators method is invoked to combine accumulators from
      separate batches into a single combined output accumulator value, once all
@@ -296,7 +296,7 @@ class CombineFn(WithTypeHints):
   def add_input(self, accumulator, element, *args, **kwargs):
     """Return result of folding element into accumulator.
 
-    CombineFn implementors must override either add_input or add_inputs.
+    CombineFn implementors must override add_input.
 
     Args:
       accumulator: the current accumulator
@@ -420,7 +420,7 @@ class CallableWrapperCombineFn(CombineFn):
     if accumulator is self._EMPTY:
       return self._fn(elements, *args, **kwargs)
     elif isinstance(elements, (list, tuple)):
-      return self._fn([accumulator] + elements, *args, **kwargs)
+      return self._fn([accumulator] + list(elements), *args, **kwargs)
     else:
       def union():
         yield accumulator


Mime
View raw message