shunping commented on code in PR #35569:
URL: https://github.com/apache/beam/pull/35569#discussion_r2205565690
##########
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:
I think it will be more complicated than supporting a lambda, because we
will have to handle the `deferred_now` differently. People usually do
`start_timestamp=now` and `stop_timestamp=now+duration`.
If we replace that with `deferred_now`, then we will need to also handle
timestamp arithmetic particularly for it.
--
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: github-unsubscr...@beam.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
