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 2024/01/19 18:21:23 UTC
(beam) branch master updated: [YAML] add time unit suffix to windowing (#30035)
This is an automated email from the ASF dual-hosted git repository.
robertwb 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 81688eb57b3 [YAML] add time unit suffix to windowing (#30035)
81688eb57b3 is described below
commit 81688eb57b36da36fb272224d1a5eda79d8f09e6
Author: Jeff Kinard <je...@thekinards.com>
AuthorDate: Fri Jan 19 13:21:15 2024 -0500
[YAML] add time unit suffix to windowing (#30035)
* [YAML] add time unit suffix to windowing
Signed-off-by: Jeffrey Kinard <je...@thekinards.com>
* add ms and regex
Signed-off-by: Jeffrey Kinard <je...@thekinards.com>
* add unit tests
Signed-off-by: Jeffrey Kinard <je...@thekinards.com>
---------
Signed-off-by: Jeffrey Kinard <je...@thekinards.com>
---
sdks/python/apache_beam/yaml/README.md | 18 ++---
sdks/python/apache_beam/yaml/yaml_provider.py | 93 +++++++++++++++++-----
.../apache_beam/yaml/yaml_provider_unit_test.py | 68 ++++++++++++++++
.../python/apache_beam/yaml/yaml_transform_test.py | 12 +--
.../apache_beam/yaml/yaml_transform_unit_test.py | 2 +-
5 files changed, 155 insertions(+), 38 deletions(-)
diff --git a/sdks/python/apache_beam/yaml/README.md b/sdks/python/apache_beam/yaml/README.md
index caf3a542b2b..75fe5c11d12 100644
--- a/sdks/python/apache_beam/yaml/README.md
+++ b/sdks/python/apache_beam/yaml/README.md
@@ -264,7 +264,7 @@ In order to meaningfully aggregate elements in a streaming pipeline,
some kind of windowing is typically required. Beam's
[windowing](https://beam.apache.org/documentation/programming-guide/#windowing)
and [triggering](https://beam.apache.org/documentation/programming-guide/#triggers)
-can be be declared using the same WindowInto transform available in all other
+can be declared using the same WindowInto transform available in all other
SDKs.
```
@@ -284,7 +284,7 @@ pipeline:
- type: WindowInto
windowing:
type: fixed
- size: 60
+ size: 60s
- type: SomeAggregation
- type: WriteToPubSub
config:
@@ -308,8 +308,8 @@ pipeline:
- type: SomeAggregation
windowing:
type: sliding
- size: 60
- period: 10
+ size: 60s
+ period: 10s
- type: WriteToPubSub
config:
topic: anotherPubSubTopic
@@ -334,7 +334,7 @@ pipeline:
query: "select col1, count(*) as c from PCOLLECTION"
windowing:
type: sessions
- gap: 60
+ gap: 60s
- type: WriteToPubSub
config:
topic: anotherPubSubTopic
@@ -369,7 +369,7 @@ pipeline:
right: ReadRight
windowing:
type: fixed
- size: 60
+ size: 60s
```
For a transform with no inputs, the specified windowing is instead applied to
@@ -387,7 +387,7 @@ pipeline:
schema: ...
windowing:
type: fixed
- size: 60
+ size: 60s
- type: Sql
config:
query: "select col1, count(*) as c from PCOLLECTION"
@@ -438,7 +438,7 @@ pipeline:
schema: ...
windowing:
type: fixed
- size: 10
+ size: 10s
transforms:
- type: Sql
@@ -451,7 +451,7 @@ pipeline:
path: /path/to/output.json
windowing:
type: fixed
- size: 300
+ size: 5m
```
diff --git a/sdks/python/apache_beam/yaml/yaml_provider.py b/sdks/python/apache_beam/yaml/yaml_provider.py
index fdfcf69ba66..99c98f891de 100755
--- a/sdks/python/apache_beam/yaml/yaml_provider.py
+++ b/sdks/python/apache_beam/yaml/yaml_provider.py
@@ -208,8 +208,9 @@ class ExternalProvider(Provider):
urns = SafeLineLoader.strip_metadata(spec['transforms'])
type = spec['type']
config = SafeLineLoader.strip_metadata(spec.get('config', {}))
- extra_params = set(SafeLineLoader.strip_metadata(spec).keys()) - set(
- ['transforms', 'type', 'config'])
+ extra_params = set(SafeLineLoader.strip_metadata(spec).keys()) - {
+ 'transforms', 'type', 'config'
+ }
if extra_params:
raise ValueError(
f'Unexpected parameters in provider of type {type} '
@@ -260,6 +261,7 @@ def maven_jar(
urns,
lambda: subprocess_server.JavaJarServer.path_to_maven_jar(
artifact_id=artifact_id,
+ group_id=group_id,
version=version,
repository=repository,
classifier=classifier,
@@ -550,7 +552,8 @@ def dicts_to_rows(o):
return o
-def create_builtin_provider():
+class YamlProviders:
+ @staticmethod
def create(elements: Iterable[Any], reshuffle: Optional[bool] = True):
"""Creates a collection containing a specified set of elements.
@@ -565,7 +568,7 @@ def create_builtin_provider():
Args:
elements: The set of elements that should belong to the PCollection.
YAML/JSON-style mappings will be interpreted as Beam rows.
- reshuffle (optional): Whether to introduce a reshuffle (to possibly
+ reshuffle: (optional) Whether to introduce a reshuffle (to possibly
redistribute the work) if there is more than one element in the
collection. Defaults to True.
"""
@@ -574,6 +577,7 @@ def create_builtin_provider():
# Or should this be posargs, args?
# pylint: disable=dangerous-default-value
+ @staticmethod
def fully_qualified_named_transform(
constructor: str,
args: Optional[Iterable[Any]] = (),
@@ -585,13 +589,13 @@ def create_builtin_provider():
via a YAML interface. Note, however, that conversion may be required if this
transform does not accept or produce Beam Rows.
- For example,
+ For example::
type: PyTransform
config:
- constructor: apache_beam.pkg.mod.SomeClass
- args: [1, 'foo']
- kwargs:
+ constructor: apache_beam.pkg.mod.SomeClass
+ args: [1, 'foo']
+ kwargs:
baz: 3
can be used to access the transform
@@ -651,6 +655,17 @@ def create_builtin_provider():
See [the Beam documentation on windowing](https://beam.apache.org/documentation/programming-guide/#windowing)
for more details.
+ Sizes, offsets, periods and gaps (where applicable) must be defined using
+ a time unit suffix 'ms', 's', 'm', 'h' or 'd' for milliseconds, seconds,
+ minutes, hours or days, respectively. If a time unit is not specified, it
+ will default to 's'.
+
+ For example::
+
+ windowing:
+ type: fixed
+ size: 30s
+
Note that any Yaml transform can have a
[windowing parameter](https://github.com/apache/beam/blob/master/sdks/python/apache_beam/yaml/README.md#windowing),
which is applied to its inputs (if any) or outputs (if there are no inputs)
@@ -665,6 +680,29 @@ def create_builtin_provider():
def expand(self, pcoll):
return pcoll | self._window_transform
+ @staticmethod
+ def _parse_duration(value, name):
+ time_units = {
+ 'ms': 0.001, 's': 1, 'm': 60, 'h': 60 * 60, 'd': 60 * 60 * 12
+ }
+ value, suffix = re.match(r'^(.*?)([^\d]*)$', str(value)).groups()
+ # Default to seconds if time unit suffix is not defined
+ if not suffix:
+ suffix = 's'
+ if not value:
+ raise ValueError(
+ f"Invalid windowing {name} value "
+ f"'{suffix if not value else value}'. "
+ f"Must provide numeric value.")
+ if suffix not in time_units:
+ raise ValueError((
+ "Invalid windowing {} time unit '{}'. " +
+ "Valid time units are {}.").format(
+ name,
+ suffix,
+ ', '.join("'{}'".format(k) for k in time_units.keys())))
+ return float(value) * time_units[suffix]
+
@staticmethod
def _parse_window_spec(spec):
spec = dict(spec)
@@ -673,18 +711,27 @@ def create_builtin_provider():
if window_type == 'global':
window_fn = window.GlobalWindows()
elif window_type == 'fixed':
- window_fn = window.FixedWindows(spec.pop('size'), spec.pop('offset', 0))
+ window_fn = window.FixedWindows(
+ YamlProviders.WindowInto._parse_duration(spec.pop('size'), 'size'),
+ YamlProviders.WindowInto._parse_duration(
+ spec.pop('offset', 0), 'offset'))
elif window_type == 'sliding':
window_fn = window.SlidingWindows(
- spec.pop('size'), spec.pop('period'), spec.pop('offset', 0))
+ YamlProviders.WindowInto._parse_duration(spec.pop('size'), 'size'),
+ YamlProviders.WindowInto._parse_duration(
+ spec.pop('period'), 'period'),
+ YamlProviders.WindowInto._parse_duration(
+ spec.pop('offset', 0), 'offset'))
elif window_type == 'sessions':
- window_fn = window.FixedWindows(spec.pop('gap'))
+ window_fn = window.Sessions(
+ YamlProviders.WindowInto._parse_duration(spec.pop('gap'), 'gap'))
if spec:
raise ValueError(f'Unknown parameters {spec.keys()}')
# TODO: Triggering, etc.
return beam.WindowInto(window_fn)
- def LogForTesting():
+ @staticmethod
+ def log_for_testing():
"""Logs each element of its input PCollection.
The output of this transform is a copy of its input for ease of use in
@@ -694,16 +741,18 @@ def create_builtin_provider():
logging.info(x)
return x
- return beam.Map(log_and_return)
+ return "LogForTesting" >> beam.Map(log_and_return)
- return InlineProvider({
- 'Create': create,
- 'LogForTesting': LogForTesting,
- 'PyTransform': fully_qualified_named_transform,
- 'Flatten': Flatten,
- 'WindowInto': WindowInto,
- },
- no_input_transforms=('Create', ))
+ @staticmethod
+ def create_builtin_provider():
+ return InlineProvider({
+ 'Create': YamlProviders.create,
+ 'LogForTesting': YamlProviders.log_for_testing,
+ 'PyTransform': YamlProviders.fully_qualified_named_transform,
+ 'Flatten': YamlProviders.Flatten,
+ 'WindowInto': YamlProviders.WindowInto,
+ },
+ no_input_transforms=('Create', ))
class PypiExpansionService:
@@ -943,7 +992,7 @@ def standard_providers():
standard_providers = yaml.load(fin, Loader=SafeLoader)
return merge_providers(
- create_builtin_provider(),
+ YamlProviders.create_builtin_provider(),
create_mapping_providers(),
create_combine_providers(),
io_providers(),
diff --git a/sdks/python/apache_beam/yaml/yaml_provider_unit_test.py b/sdks/python/apache_beam/yaml/yaml_provider_unit_test.py
new file mode 100644
index 00000000000..ec71422fd16
--- /dev/null
+++ b/sdks/python/apache_beam/yaml/yaml_provider_unit_test.py
@@ -0,0 +1,68 @@
+#
+# 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.
+#
+
+import logging
+import unittest
+
+from apache_beam.yaml.yaml_provider import YamlProviders
+
+
+class WindowIntoTest(unittest.TestCase):
+ def __init__(self, methodName="runWindowIntoTest"):
+ unittest.TestCase.__init__(self, methodName)
+ self.parse_duration = YamlProviders.WindowInto._parse_duration
+
+ def test_parse_duration_ms(self):
+ value = self.parse_duration('1000ms', 'size')
+ self.assertEqual(1, value)
+
+ def test_parse_duration_sec(self):
+ value = self.parse_duration('1s', 'size')
+ self.assertEqual(1, value)
+
+ def test_parse_duration_min(self):
+ value = self.parse_duration('1m', 'size')
+ self.assertEqual(60, value)
+
+ def test_parse_duration_hour(self):
+ value = self.parse_duration('1h', 'size')
+ self.assertEqual(3600, value)
+
+ def test_parse_duration_from_decimal(self):
+ value = self.parse_duration('1.5m', 'size')
+ self.assertEqual(90, value)
+
+ def test_parse_duration_to_decimal(self):
+ value = self.parse_duration('1ms', 'size')
+ self.assertEqual(0.001, value)
+
+ def test_parse_duration_with_missing_suffix(self):
+ value = self.parse_duration('1', 'size')
+ self.assertEqual(1, value)
+
+ def test_parse_duration_with_invalid_suffix(self):
+ with self.assertRaises(ValueError):
+ self.parse_duration('1x', 'size')
+
+ def test_parse_duration_with_missing_value(self):
+ with self.assertRaises(ValueError):
+ self.parse_duration('s', 'size')
+
+
+if __name__ == '__main__':
+ logging.getLogger().setLevel(logging.INFO)
+ unittest.main()
diff --git a/sdks/python/apache_beam/yaml/yaml_transform_test.py b/sdks/python/apache_beam/yaml/yaml_transform_test.py
index 6414f3dbbeb..9641df0896f 100644
--- a/sdks/python/apache_beam/yaml/yaml_transform_test.py
+++ b/sdks/python/apache_beam/yaml/yaml_transform_test.py
@@ -505,7 +505,7 @@ class YamlWindowingTest(unittest.TestCase):
- type: WindowInto
windowing:
type: fixed
- size: 4
+ size: 4s
- type: SumGlobally
''',
providers=TEST_PROVIDERS)
@@ -524,7 +524,7 @@ class YamlWindowingTest(unittest.TestCase):
- type: SumGlobally
windowing:
type: fixed
- size: 4
+ size: 4s
''',
providers=TEST_PROVIDERS)
assert_that(result, equal_to([6, 9]))
@@ -548,7 +548,7 @@ class YamlWindowingTest(unittest.TestCase):
input: [Create1, Create2]
windowing:
type: fixed
- size: 4
+ size: 4s
output: SumGlobally
''',
providers=TEST_PROVIDERS)
@@ -566,7 +566,7 @@ class YamlWindowingTest(unittest.TestCase):
elements: [0, 1, 2, 3, 4, 5]
windowing:
type: fixed
- size: 4
+ size: 4s
- type: SumGlobally
''',
providers=TEST_PROVIDERS)
@@ -585,7 +585,7 @@ class YamlWindowingTest(unittest.TestCase):
- type: SumGlobally
windowing:
type: fixed
- size: 4
+ size: 4s
''',
providers=TEST_PROVIDERS)
assert_that(result, equal_to([6, 9]))
@@ -614,7 +614,7 @@ class YamlWindowingTest(unittest.TestCase):
v: sum
windowing:
type: fixed
- size: 10
+ size: 10s
''',
providers=TEST_PROVIDERS)
assert_that(
diff --git a/sdks/python/apache_beam/yaml/yaml_transform_unit_test.py b/sdks/python/apache_beam/yaml/yaml_transform_unit_test.py
index d1886ba4dcf..00e509310d6 100644
--- a/sdks/python/apache_beam/yaml/yaml_transform_unit_test.py
+++ b/sdks/python/apache_beam/yaml/yaml_transform_unit_test.py
@@ -209,7 +209,7 @@ class MainTest(unittest.TestCase):
inputs={'elements': elements})
self.assertRegex(
str(expand_composite_transform(spec, scope)['output']),
- r"PCollection.*Composite/Map.*")
+ r"PCollection.*Composite/LogForTesting.*")
def test_expand_composite_transform_root(self):
with new_pipeline() as p: