damccorm commented on code in PR #35569:
URL: https://github.com/apache/beam/pull/35569#discussion_r2205704552
##########
sdks/python/apache_beam/transforms/periodicsequence.py:
##########
@@ -301,21 +302,37 @@ def __init__(
sufficient to cover the duration defined by `start_timestamp`,
`stop_timestamp`, and `fire_interval`; otherwise, a `ValueError` is
raised.
+
+ :param rebase_timestamp: Whether to shift the generated timestamps to begin
+ at pipeline execution time. If true, this overrides the start time with
+ the pipeline's actual execution time. The stop time is then adjusted to
+ maintain the original duration between them. This is useful to avoid
+ generating a burst of events at the start of the pipeline due to the
+ implementation of `ImpulseSeqGenDoFn`, which tries to align the generated
+ timestamp with the execution time. Defaults to false.
'''
self.start_ts = start_timestamp
self.stop_ts = stop_timestamp
self.interval = fire_interval
self.apply_windowing = apply_windowing
self.data = data
+ self.rebase_timestamp = rebase_timestamp
if self.data:
self._validate_and_adjust_duration()
def expand(self, pbegin):
+ if self.rebase_timestamp:
Review Comment:
Could we do something in between these options where we still take a new
parameter `rebase`, but it takes an enum with values `REBASE_NONE`,
`REBASE_START`, `REBASE_FINISH`, and `REBASE_ALL`?
--
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]
