diff --git a/sdks/python/apache_beam/examples/snippets/snippets.py b/sdks/python/apache_beam/examples/snippets/snippets.py index 49136d4e3779..ead0ed32c220 100644 --- a/sdks/python/apache_beam/examples/snippets/snippets.py +++ b/sdks/python/apache_beam/examples/snippets/snippets.py @@ -1575,6 +1575,66 @@ def cross_join(left, rights): return pipeline, result +def side_input_slow_update_global_window( + first_timestamp, last_timestamp, side_input_interval, main_input_interval): + # [START SideInputPatternSlowUpdateGlobalWindowSnip1] + from apache_beam.transforms import combiners + from apache_beam.transforms import trigger + from apache_beam.transforms import window + from apache_beam.transforms.periodicsequence import PeriodicImpulse + + # To run indefinitely, pass MAX_TIMESTAMP from apache_beam.utils.timestamp + # as last_timestamp. + + # Placeholder that represents an external service, such as a database or a + # configuration endpoint. Replace it with the external read of your choice. + def read_from_placeholder_external_service(refresh_timestamp): + return {'Key_A': str(refresh_timestamp)} + + def enrich_with_side_input(element, config): + # The side input is read as an iterable rather than as a singleton, + # because the global window side input can hold more than one element if + # it fires again before the value is consumed. + latest_config = next(iter(config), {}) + return element, latest_config.get('Key_A') + + # Create pipeline. + pipeline = beam.Pipeline() + + # Periodically read the external data into the global window. + # Repeatedly(AfterCount(1)) emits a new pane for every impulse, and the + # DISCARDING accumulation mode drops the previous value, so that + # Latest.Globally only considers the most recent element. + # Use Latest.Globally().without_defaults(): the variant with defaults adds + # its own side input, which stops the transform from emitting more than once. + side_input = ( + pipeline + | 'SideInputImpulse' >> PeriodicImpulse( + first_timestamp, last_timestamp, side_input_interval) + | 'ReadExternalData' >> beam.Map(read_from_placeholder_external_service) + | 'WindowSideInput' >> beam.WindowInto( + window.GlobalWindows(), + trigger=trigger.Repeatedly(trigger.AfterCount(1)), + accumulation_mode=trigger.AccumulationMode.DISCARDING) + | 'GetLatest' >> combiners.Latest.Globally().without_defaults()) + + # Consume the side input from a main input that uses non-global windows. + # PeriodicImpulse generates test data. Use a real streaming source, such as + # PubSubIO or KafkaIO, in production. + result = ( + pipeline + | 'MainInputImpulse' >> PeriodicImpulse( + first_timestamp, + last_timestamp, + main_input_interval, + apply_windowing=True) + | 'ApplySideInput' >> beam.Map( + enrich_with_side_input, config=beam.pvalue.AsIter(side_input))) + # [END SideInputPatternSlowUpdateGlobalWindowSnip1] + + return pipeline, result + + def bigqueryio_deadletter(): # [START BigQueryIODeadLetter] diff --git a/sdks/python/apache_beam/examples/snippets/snippets_test.py b/sdks/python/apache_beam/examples/snippets/snippets_test.py index e4f2d16ac144..5a47d9289012 100644 --- a/sdks/python/apache_beam/examples/snippets/snippets_test.py +++ b/sdks/python/apache_beam/examples/snippets/snippets_test.py @@ -1467,6 +1467,44 @@ def test_side_input_slow_update(self): for i in range(-1, 10, 1): os.unlink(src_file_pattern + str(first_ts + interval * i)) + def test_side_input_slow_update_global_window(self): + side_input_interval = 5 + main_input_interval = 5 + duration = 30 + + first_ts = math.floor(time.time()) - duration + # Aligning the timestamp to get persistent results. + first_ts = first_ts - ( + first_ts % (side_input_interval * main_input_interval)) + last_ts = first_ts + duration + + expected_main_input_elements = sorted( + first_ts + main_input_interval * i + for i in range(duration // main_input_interval)) + expected_side_input_values = set( + str(first_ts + side_input_interval * i) + for i in range(duration // side_input_interval)) + + def check_enriched_elements(actual): + main_input_elements = sorted(element for element, _ in actual) + assert main_input_elements == expected_main_input_elements, ( + 'Expected main input elements %s, got %s' % + (expected_main_input_elements, main_input_elements)) + + side_input_values = set(value for _, value in actual) + assert None not in side_input_values, ( + 'Some elements were processed before the side input was ready.') + unexpected_values = side_input_values - expected_side_input_values + assert not unexpected_values, ( + 'Observed side input values that the side input never produced: %s' % + unexpected_values) + + pipeline, pipeline_result = snippets.side_input_slow_update_global_window( + first_ts, last_ts, side_input_interval, main_input_interval) + + with pipeline: + assert_that(pipeline_result, check_enriched_elements) + class ValueProviderInfoTest(unittest.TestCase): """Tests for accessing value provider info after run.""" diff --git a/website/www/site/content/en/documentation/patterns/side-inputs.md b/website/www/site/content/en/documentation/patterns/side-inputs.md index 136eeef29ada..3a14fdff314c 100644 --- a/website/www/site/content/en/documentation/patterns/side-inputs.md +++ b/website/www/site/content/en/documentation/patterns/side-inputs.md @@ -47,8 +47,12 @@ For instance, the following code sample uses a `Map` to create a `DoFn`. The `Ma {{< code_sample "examples/java/src/main/java/org/apache/beam/examples/snippets/Snippets.java" SideInputPatternSlowUpdateGlobalWindowSnip1 >}} {{< /highlight >}} +{{< paragraph class="language-py" >}} +The Python sample uses `PeriodicImpulse` to re-read the placeholder external service on a fixed interval, and re-publishes the result into the global window on every firing. Use `Latest.Globally().without_defaults()` rather than `Latest.Globally()`: the variant with defaults adds its own side input, which stops the transform from emitting more than once. For more information, see [Issue 35934](https://github.com/apache/beam/issues/35934). +{{< /paragraph >}} + {{< highlight py >}} -No sample present. +{{< code_sample "sdks/python/apache_beam/examples/snippets/snippets.py" SideInputPatternSlowUpdateGlobalWindowSnip1 >}} {{< /highlight >}}