From 8ba5b979bd76ef17c1a0acaeb502ee74574aae0c Mon Sep 17 00:00:00 2001 From: Ankit Kumar Date: Sat, 4 Oct 2025 11:06:47 +0530 Subject: [PATCH] Example on SiseInput --- .../apache_beam/examples/snippets/snippets.py | 43 +++++++++++++++++++ 1 file changed, 43 insertions(+) diff --git a/sdks/python/apache_beam/examples/snippets/snippets.py b/sdks/python/apache_beam/examples/snippets/snippets.py index f6bf5e5d44ec..ba411507d572 100644 --- a/sdks/python/apache_beam/examples/snippets/snippets.py +++ b/sdks/python/apache_beam/examples/snippets/snippets.py @@ -1772,3 +1772,46 @@ def process(self, element, bundle_finalizer=beam.DoFn.BundleFinalizerParam): bundle_finalizer.register(my_callback_func) # [END BundleFinalize] + + # [START SideInput SlowUpdate] +import apache_beam as beam +from apache_beam.transforms import core +from apache_beam.transforms.window import GlobalWindows, WindowInto, Repeatedly, AfterCount, AccumulationMode +from apache_beam.transforms.periodicsequence import PeriodicImpulse +from apache_beam.transforms.combiners import Latest +import logging +import time + +options = PipelineOptions() +pipeline = beam.Pipeline(options=options) + +start_timestamp = time.time() +main_input_fire_interval = 60 +side_input_fire_interval = 60 + +def add_timestamp(element, timestamp=core.DoFn.TimestampParam): + return (element, timestamp) + +side_input = ( +pipeline +| "SideInputImpulse" >> PeriodicImpulse( +start_timestamp=start_timestamp, +fire_interval=main_input_fire_interval, +) +| "SideInputWindow" >> WindowInto( +GlobalWindows(), +trigger=Repeatedly(AfterCount(1)), +accumulation_mode=AccumulationMode.DISCARDING, +) +| "Add timestamp" >> core.ParDo(add_timestamp) +) + +( +side_input +| "LocalLatest" >> Latest.Globally().without_defaults() +| "Show latest" >> beam.Map(logging.info) +) + +# [END SideInput SlowUpdate] + +