You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@beam.apache.org by al...@apache.org on 2019/02/14 23:32:42 UTC
[beam] branch master updated: [BEAM-5731] Only accept key operator
in Python 3. (#7842)
This is an automated email from the ASF dual-hosted git repository.
altay pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/beam.git
The following commit(s) were added to refs/heads/master by this push:
new e712870 [BEAM-5731] Only accept key operator in Python 3. (#7842)
e712870 is described below
commit e7128701a4d195a1ba70750b393ecf218989289c
Author: Robert Bradshaw <ro...@gmail.com>
AuthorDate: Fri Feb 15 00:32:29 2019 +0100
[BEAM-5731] Only accept key operator in Python 3. (#7842)
* Only accept key operator in Python 3.
This is in line with Python's change to its sorting operations.
---
sdks/python/apache_beam/transforms/combiners.py | 58 +++++++++++++++++++++-
.../apache_beam/transforms/combiners_test.py | 36 ++++++++++----
.../apache_beam/transforms/ptransform_test.py | 10 ++--
3 files changed, 87 insertions(+), 17 deletions(-)
diff --git a/sdks/python/apache_beam/transforms/combiners.py b/sdks/python/apache_beam/transforms/combiners.py
index 3976a8e..9ac6ef2 100644
--- a/sdks/python/apache_beam/transforms/combiners.py
+++ b/sdks/python/apache_beam/transforms/combiners.py
@@ -23,6 +23,8 @@ from __future__ import division
import heapq
import operator
import random
+import sys
+import warnings
from builtins import object
from builtins import zip
@@ -166,7 +168,7 @@ class Top(object):
function supplied as the compare argument.
"""
- def __init__(self, n, compare=None, *args, **kwargs):
+ def _py2__init__(self, n, compare=None, *args, **kwargs):
"""Initializer.
compare should be an implementation of "a < b" taking at least two
@@ -183,6 +185,9 @@ class Top(object):
*args: as described above.
**kwargs: as described above.
"""
+ if compare:
+ warnings.warn('Compare not available in Python 3, use key instead.',
+ DeprecationWarning)
self._n = n
self._compare = compare
self._key = kwargs.pop('key', None)
@@ -190,6 +195,29 @@ class Top(object):
self._args = args
self._kwargs = kwargs
+ def _py3__init__(self, n, **kwargs):
+ """Creates a global Top operation.
+
+ The arguments 'key' and 'reverse' may be passed as keyword arguments,
+ and have the same meaning as for Python's sort functions.
+
+ Args:
+ pcoll: PCollection to process.
+ n: number of elements to extract from pcoll.
+ **kwargs: may contain 'key' and/or 'reverse'
+ """
+ unknown_kwargs = set(kwargs.keys()) - set(['key', 'reverse'])
+ if unknown_kwargs:
+ raise ValueError(
+ 'Unknown keyword arguments: ' + ', '.join(unknown_kwargs))
+ self._py2__init__(n, None, **kwargs)
+
+ # Python 3 sort does not accept a comparison operator, and nor do we.
+ if sys.version_info[0] < 3:
+ __init__ = _py2__init__
+ else:
+ __init__ = _py3__init__
+
def default_label(self):
return 'Top(%d)' % self._n
@@ -227,7 +255,7 @@ class Top(object):
"greatest" is determined by the comparator function supplied as the compare
argument in the initializer.
"""
- def __init__(self, n, compare=None, *args, **kwargs):
+ def _py2__init__(self, n, compare=None, *args, **kwargs):
"""Initializer.
compare should be an implementation of "a < b" taking at least two
@@ -244,6 +272,9 @@ class Top(object):
*args: as described above.
**kwargs: as described above.
"""
+ if compare:
+ warnings.warn('Compare not available in Python 3, use key instead.',
+ DeprecationWarning)
self._n = n
self._compare = compare
self._key = kwargs.pop('key', None)
@@ -251,6 +282,29 @@ class Top(object):
self._args = args
self._kwargs = kwargs
+ def _py3__init__(self, n, **kwargs):
+ """Creates a per-key Top operation.
+
+ The arguments 'key' and 'reverse' may be passed as keyword arguments,
+ and have the same meaning as for Python's sort functions.
+
+ Args:
+ pcoll: PCollection to process.
+ n: number of elements to extract from pcoll.
+ **kwargs: may contain 'key' and/or 'reverse'
+ """
+ unknown_kwargs = set(kwargs.keys()) - set(['key', 'reverse'])
+ if unknown_kwargs:
+ raise ValueError(
+ 'Unknown keyword arguments: ' + ', '.join(unknown_kwargs))
+ self._py2__init__(n, None, **kwargs)
+
+ # Python 3 sort does not accept a comparison operator, and nor do we.
+ if sys.version_info[0] < 3:
+ __init__ = _py2__init__
+ else:
+ __init__ = _py3__init__
+
def default_label(self):
return 'TopPerKey(%d)' % self._n
diff --git a/sdks/python/apache_beam/transforms/combiners_test.py b/sdks/python/apache_beam/transforms/combiners_test.py
index f070c32..8dad3ba 100644
--- a/sdks/python/apache_beam/transforms/combiners_test.py
+++ b/sdks/python/apache_beam/transforms/combiners_test.py
@@ -21,6 +21,7 @@ from __future__ import division
import itertools
import random
+import sys
import unittest
import hamcrest as hc
@@ -66,6 +67,28 @@ class CombineTest(unittest.TestCase):
def test_top(self):
pipeline = TestPipeline()
+ # First for global combines.
+ pcoll = pipeline | 'start' >> Create([6, 3, 1, 1, 9, 1, 5, 2, 0, 6])
+ result_top = pcoll | 'top' >> combine.Top.Largest(5)
+ result_bot = pcoll | 'bot' >> combine.Top.Smallest(4)
+ assert_that(result_top, equal_to([[9, 6, 6, 5, 3]]), label='assert:top')
+ assert_that(result_bot, equal_to([[0, 1, 1, 1]]), label='assert:bot')
+
+ # Again for per-key combines.
+ pcoll = pipeline | 'start-perkye' >> Create(
+ [('a', x) for x in [6, 3, 1, 1, 9, 1, 5, 2, 0, 6]])
+ result_key_top = pcoll | 'top-perkey' >> combine.Top.LargestPerKey(5)
+ result_key_bot = pcoll | 'bot-perkey' >> combine.Top.SmallestPerKey(4)
+ assert_that(result_key_top, equal_to([('a', [9, 6, 6, 5, 3])]),
+ label='key:top')
+ assert_that(result_key_bot, equal_to([('a', [0, 1, 1, 1])]),
+ label='key:bot')
+ pipeline.run()
+
+ @unittest.skipIf(sys.version_info[0] > 2, 'deprecated comparator')
+ def test_top_py2(self):
+ pipeline = TestPipeline()
+
# A parameter we'll be sharing with a custom comparator.
names = {0: 'zo',
1: 'one',
@@ -77,8 +100,7 @@ class CombineTest(unittest.TestCase):
# First for global combines.
pcoll = pipeline | 'start' >> Create([6, 3, 1, 1, 9, 1, 5, 2, 0, 6])
- result_top = pcoll | 'top' >> combine.Top.Largest(5)
- result_bot = pcoll | 'bot' >> combine.Top.Smallest(4)
+
result_cmp = pcoll | 'cmp' >> combine.Top.Of(
6,
lambda a, b, names: len(names[a]) < len(names[b]),
@@ -88,24 +110,16 @@ class CombineTest(unittest.TestCase):
lambda a, b, names: len(names[a]) < len(names[b]),
names, # Note parameter passed to comparator.
reverse=True)
- assert_that(result_top, equal_to([[9, 6, 6, 5, 3]]), label='assert:top')
- assert_that(result_bot, equal_to([[0, 1, 1, 1]]), label='assert:bot')
assert_that(result_cmp, equal_to([[9, 6, 6, 5, 3, 2]]), label='assert:cmp')
assert_that(result_cmp_rev, equal_to([[0, 1, 1]]), label='assert:cmp_rev')
# Again for per-key combines.
pcoll = pipeline | 'start-perkye' >> Create(
[('a', x) for x in [6, 3, 1, 1, 9, 1, 5, 2, 0, 6]])
- result_key_top = pcoll | 'top-perkey' >> combine.Top.LargestPerKey(5)
- result_key_bot = pcoll | 'bot-perkey' >> combine.Top.SmallestPerKey(4)
result_key_cmp = pcoll | 'cmp-perkey' >> combine.Top.PerKey(
6,
lambda a, b, names: len(names[a]) < len(names[b]),
names) # Note parameter passed to comparator.
- assert_that(result_key_top, equal_to([('a', [9, 6, 6, 5, 3])]),
- label='key:top')
- assert_that(result_key_bot, equal_to([('a', [0, 1, 1, 1])]),
- label='key:bot')
assert_that(result_key_cmp, equal_to([('a', [9, 6, 6, 5, 3, 2])]),
label='key:cmp')
pipeline.run()
@@ -134,6 +148,8 @@ class CombineTest(unittest.TestCase):
['aa', 'bbb', 'c', 'dddd'] | combine.Top.Of(3, key=len, reverse=True),
[['c', 'aa', 'bbb']])
+ @unittest.skipIf(sys.version_info[0] > 2, 'deprecated comparator')
+ def test_top_key_py2(self):
# The largest elements compared by their length mod 5.
self.assertEqual(
['aa', 'bbbb', 'c', 'ddddd', 'eee', 'ffffff'] | combine.Top.Of(
diff --git a/sdks/python/apache_beam/transforms/ptransform_test.py b/sdks/python/apache_beam/transforms/ptransform_test.py
index ffb4331..75a0102 100644
--- a/sdks/python/apache_beam/transforms/ptransform_test.py
+++ b/sdks/python/apache_beam/transforms/ptransform_test.py
@@ -1852,7 +1852,7 @@ class PTransformTypeCheckTestCase(TypeHintTestCase):
def test_top_of_pipeline_checking_satisfied(self):
d = (self.p
| beam.Create(range(5, 11)).with_output_types(int)
- | 'Top 3' >> combine.Top.Of(3, lambda x, y: x < y))
+ | 'Top 3' >> combine.Top.Of(3))
self.assertCompatible(typehints.Iterable[int],
d.element_type)
@@ -1864,7 +1864,7 @@ class PTransformTypeCheckTestCase(TypeHintTestCase):
d = (self.p
| beam.Create(list('testing')).with_output_types(str)
- | 'AciiTop' >> combine.Top.Of(3, lambda x, y: x < y))
+ | 'AciiTop' >> combine.Top.Of(3))
self.assertCompatible(typehints.Iterable[str], d.element_type)
assert_that(d, equal_to([['t', 't', 's']]))
@@ -1875,7 +1875,7 @@ class PTransformTypeCheckTestCase(TypeHintTestCase):
(self.p
| beam.Create(range(100)).with_output_types(int)
| 'Num + 1' >> beam.Map(lambda x: x + 1).with_output_types(int)
- | 'TopMod' >> combine.Top.PerKey(1, lambda a, b: a < b))
+ | 'TopMod' >> combine.Top.PerKey(1))
self.assertEqual(
"Type hint violation for 'CombinePerKey(TopCombineFn)': "
@@ -1888,7 +1888,7 @@ class PTransformTypeCheckTestCase(TypeHintTestCase):
| beam.Create(range(100)).with_output_types(int)
| ('GroupMod 3' >> beam.Map(lambda x: (x % 3, x))
.with_output_types(typehints.KV[int, int]))
- | 'TopMod' >> combine.Top.PerKey(1, lambda a, b: a < b))
+ | 'TopMod' >> combine.Top.PerKey(1))
self.assertCompatible(typehints.Tuple[int, typehints.Iterable[int]],
d.element_type)
@@ -1902,7 +1902,7 @@ class PTransformTypeCheckTestCase(TypeHintTestCase):
| beam.Create(range(21))
| ('GroupMod 3' >> beam.Map(lambda x: (x % 3, x))
.with_output_types(typehints.KV[int, int]))
- | 'TopMod' >> combine.Top.PerKey(1, lambda a, b: a < b))
+ | 'TopMod' >> combine.Top.PerKey(1))
self.assertCompatible(typehints.KV[int, typehints.Iterable[int]],
d.element_type)