Google Git
Sign in
apache/beam/refs/heads/timestamp-buffer^!/.
commit
09d2a9a9910cda2bd1aa74b82d981c362d8064a9
[log]
author
Shunping Huang <shunping@google.com>
Fri Apr 10 11:34:25 2026 -0400
committer
Shunping Huang <shunping@google.com>
Fri Apr 10 11:34:25 2026 -0400
tree
675bbe2f05e647a66584c2483ed7aac934d4ae36
parent
884ee8a996856c5410171aa50a120aa1d3ffc8d0 [diff]
Rename folder and make it into module so that we can run in dataflow.

diff --git a/sdks/python/apache_beam/examples/timestamp-buffer/run_batchelements.py b/sdks/python/apache_beam/examples/timestamp-buffer/run_batchelements.py
deleted file mode 100644
index 1f8634f..0000000
--- a/sdks/python/apache_beam/examples/timestamp-buffer/run_batchelements.py
+++ /dev/null

@@ -1,17 +0,0 @@
-import logging
-
-import apache_beam as beam
-
-from framework import prism_options, dataflow_options
-from framework import periodic_source
-from framework import dump_to_log
-
-logging.basicConfig(level=logging.INFO)
-
-with beam.Pipeline(options=prism_options) as p:
-  _ = (
-      p
-      | periodic_source
-      | beam.BatchElements(min_batch_size=5, max_batch_size=100, max_batch_duration_secs=4)
-      | dump_to_log
-  )
\ No newline at end of file

diff --git a/sdks/python/apache_beam/examples/timestamp-buffer/run_orderedwindowelements.py b/sdks/python/apache_beam/examples/timestamp-buffer/run_orderedwindowelements.py
deleted file mode 100644
index d59eba2..0000000
--- a/sdks/python/apache_beam/examples/timestamp-buffer/run_orderedwindowelements.py
+++ /dev/null

@@ -1,27 +0,0 @@
-import logging
-
-import apache_beam as beam
-from apache_beam.transforms import DoFn
-
-from framework import prism_options, dataflow_options
-from framework import periodic_source
-from framework import dump_to_log
-from ordered_window_elements import OrderedWindowElementsDoFn
-
-logging.basicConfig(level=logging.INFO)
-
-WINDOW_SIZE = 6
-SLIDE_INTERVAL = 6
-
-with beam.Pipeline(options=prism_options) as p:
-  _ = (
-      p
-      | periodic_source
-      | beam.Map(lambda x, t=DoFn.TimestampParam: (x[0], (t, x[1])))
-      | beam.ParDo(
-          OrderedWindowElementsDoFn(
-              duration=WINDOW_SIZE,
-              slide_interval=SLIDE_INTERVAL,
-              offset=0,
-              allowed_lateness=0))
-      | dump_to_log)

diff --git a/sdks/python/apache_beam/examples/timestamp-buffer/run_timestampbuffer.py b/sdks/python/apache_beam/examples/timestamp-buffer/run_timestampbuffer.py
deleted file mode 100644
index cab6760..0000000
--- a/sdks/python/apache_beam/examples/timestamp-buffer/run_timestampbuffer.py
+++ /dev/null

@@ -1,27 +0,0 @@
-import logging
-
-import apache_beam as beam
-from apache_beam.transforms import DoFn
-
-from framework import prism_options, dataflow_options
-from framework import periodic_source
-from framework import dump_to_log
-from timestamp_buffer import TimestampBufferDoFnBag
-
-logging.basicConfig(level=logging.INFO)
-
-WINDOW_SIZE = 6
-SLIDE_INTERVAL = 6
-
-class MyBufferDoFn(TimestampBufferDoFnBag):
-  def process_element(self, key, element_ts, value, context,
-                      **extra_state):
-    yield [v[1] for v in context] + [value]
-
-with beam.Pipeline(options=prism_options) as p:
-  _ = (
-      p
-      | periodic_source
-      | beam.Map(lambda x, t=DoFn.TimestampParam: (x[0], (t, x[1])))
-      | beam.ParDo(MyBufferDoFn(40))
-      | dump_to_log)

diff --git a/sdks/python/apache_beam/examples/timestamp_buffer/__init__.py b/sdks/python/apache_beam/examples/timestamp_buffer/__init__.py
new file mode 100644
index 0000000..e69de29
--- /dev/null
+++ b/sdks/python/apache_beam/examples/timestamp_buffer/__init__.py

diff --git a/sdks/python/apache_beam/examples/timestamp-buffer/framework.py b/sdks/python/apache_beam/examples/timestamp_buffer/framework.py
similarity index 100%
rename from sdks/python/apache_beam/examples/timestamp-buffer/framework.py
rename to sdks/python/apache_beam/examples/timestamp_buffer/framework.py

diff --git a/sdks/python/apache_beam/examples/timestamp-buffer/ordered_window_elements.py b/sdks/python/apache_beam/examples/timestamp_buffer/ordered_window_elements.py
similarity index 100%
rename from sdks/python/apache_beam/examples/timestamp-buffer/ordered_window_elements.py
rename to sdks/python/apache_beam/examples/timestamp_buffer/ordered_window_elements.py

diff --git a/sdks/python/apache_beam/examples/timestamp_buffer/run_batchelements.py b/sdks/python/apache_beam/examples/timestamp_buffer/run_batchelements.py
new file mode 100644
index 0000000..8147e60
--- /dev/null
+++ b/sdks/python/apache_beam/examples/timestamp_buffer/run_batchelements.py

@@ -0,0 +1,18 @@
+import logging
+
+import apache_beam as beam
+
+from apache_beam.examples.timestamp_buffer.framework import prism_options
+from apache_beam.examples.timestamp_buffer.framework import dataflow_options
+from apache_beam.examples.timestamp_buffer.framework import periodic_source
+from apache_beam.examples.timestamp_buffer.framework import dump_to_log
+
+logging.basicConfig(level=logging.INFO)
+
+with beam.Pipeline(options=prism_options) as p:
+  _ = (
+      p
+      | periodic_source
+      | beam.BatchElements(min_batch_size=5, max_batch_size=100, max_batch_duration_secs=4)
+      | dump_to_log
+  )
\ No newline at end of file

diff --git a/sdks/python/apache_beam/examples/timestamp-buffer/run_combiner1.py b/sdks/python/apache_beam/examples/timestamp_buffer/run_combiner1.py
similarity index 100%
rename from sdks/python/apache_beam/examples/timestamp-buffer/run_combiner1.py
rename to sdks/python/apache_beam/examples/timestamp_buffer/run_combiner1.py

diff --git a/sdks/python/apache_beam/examples/timestamp-buffer/run_combiner2.py b/sdks/python/apache_beam/examples/timestamp_buffer/run_combiner2.py
similarity index 66%
rename from sdks/python/apache_beam/examples/timestamp-buffer/run_combiner2.py
rename to sdks/python/apache_beam/examples/timestamp_buffer/run_combiner2.py
index 8ed55d6..ed8de32 100644
--- a/sdks/python/apache_beam/examples/timestamp-buffer/run_combiner2.py
+++ b/sdks/python/apache_beam/examples/timestamp_buffer/run_combiner2.py

@@ -5,9 +5,10 @@
 from apache_beam.transforms import DoFn
 from apache_beam.transforms.window import FixedWindows
 
-from framework import prism_options, dataflow_options
-from framework import periodic_source
-from framework import dump_to_log
+from apache_beam.examples.timestamp_buffer.framework import prism_options
+from apache_beam.examples.timestamp_buffer.framework import dataflow_options
+from apache_beam.examples.timestamp_buffer.framework import periodic_source
+from apache_beam.examples.timestamp_buffer.framework import dump_to_log
 
 logging.basicConfig(level=logging.INFO)
 

diff --git a/sdks/python/apache_beam/examples/timestamp_buffer/run_orderedwindowelements.py b/sdks/python/apache_beam/examples/timestamp_buffer/run_orderedwindowelements.py
new file mode 100644
index 0000000..91cbeb5
--- /dev/null
+++ b/sdks/python/apache_beam/examples/timestamp_buffer/run_orderedwindowelements.py

@@ -0,0 +1,28 @@
+import logging
+
+import apache_beam as beam
+from apache_beam.transforms import DoFn
+
+from apache_beam.examples.timestamp_buffer.framework import prism_options
+from apache_beam.examples.timestamp_buffer.framework import dataflow_options
+from apache_beam.examples.timestamp_buffer.framework import periodic_source
+from apache_beam.examples.timestamp_buffer.framework import dump_to_log
+from apache_beam.examples.timestamp_buffer.ordered_window_elements import OrderedWindowElementsDoFn
+
+logging.basicConfig(level=logging.INFO)
+
+WINDOW_SIZE = 6
+SLIDE_INTERVAL = 6
+
+with beam.Pipeline(options=prism_options) as p:
+  _ = (
+      p
+      | periodic_source
+      | beam.Map(lambda x, t=DoFn.TimestampParam: (x[0], (t, x[1])))
+      | beam.ParDo(
+          OrderedWindowElementsDoFn(
+              duration=WINDOW_SIZE,
+              slide_interval=SLIDE_INTERVAL,
+              offset=0,
+              allowed_lateness=0))
+      | dump_to_log)

diff --git a/sdks/python/apache_beam/examples/timestamp_buffer/run_timestampbuffer.py b/sdks/python/apache_beam/examples/timestamp_buffer/run_timestampbuffer.py
new file mode 100644
index 0000000..e45caac
--- /dev/null
+++ b/sdks/python/apache_beam/examples/timestamp_buffer/run_timestampbuffer.py

@@ -0,0 +1,28 @@
+import logging
+
+import apache_beam as beam
+from apache_beam.transforms import DoFn
+
+from apache_beam.examples.timestamp_buffer.framework import prism_options
+from apache_beam.examples.timestamp_buffer.framework import dataflow_options
+from apache_beam.examples.timestamp_buffer.framework import periodic_source
+from apache_beam.examples.timestamp_buffer.framework import dump_to_log
+from apache_beam.examples.timestamp_buffer.timestamp_buffer import TimestampBufferDoFnBag
+
+logging.basicConfig(level=logging.INFO)
+
+WINDOW_SIZE = 6
+SLIDE_INTERVAL = 6
+
+class MyBufferDoFn(TimestampBufferDoFnBag):
+  def process_element(self, key, element_ts, value, context,
+                      **extra_state):
+    yield [v[1] for v in context] + [value]
+
+with beam.Pipeline(options=dataflow_options) as p:
+  _ = (
+      p
+      | periodic_source
+      | beam.Map(lambda x, t=DoFn.TimestampParam: (x[0], (t, x[1])))
+      | beam.ParDo(MyBufferDoFn(40))
+      | dump_to_log)

diff --git a/sdks/python/apache_beam/examples/timestamp-buffer/timestamp_buffer.py b/sdks/python/apache_beam/examples/timestamp_buffer/timestamp_buffer.py
similarity index 100%
rename from sdks/python/apache_beam/examples/timestamp-buffer/timestamp_buffer.py
rename to sdks/python/apache_beam/examples/timestamp_buffer/timestamp_buffer.py