Add Python example for slowly updating global window side inputs

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]
 

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."""

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 >}}