chamikaramj commented on code in PR #40183:
URL: https://github.com/apache/beam/pull/40183#discussion_r4050992078
##########
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
+
+ # from apache_beam.utils.timestamp import MAX_TIMESTAMP
Review Comment:
Nit: pls remove commented out code.
##########
sdks/python/apache_beam/examples/snippets/snippets_test.py:
##########
@@ -1467,6 +1467,43 @@ 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 side_input_values, 'No side input value was observed.'
Review Comment:
This assertion might be passing trivially for main (input, side input)
combinations (valid value, None). Try changing to:
```
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.')
```
##########
sdks/python/apache_beam/examples/snippets/snippets_test.py:
##########
@@ -1467,6 +1467,43 @@ 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):
Review Comment:
Did we try the pipeline on Dataflow ? According to
https://github.com/apache/beam/issues/35934 the behavior seems to be different
for Dataflow.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
