Abacn commented on code in PR #39461:
URL: https://github.com/apache/beam/pull/39461#discussion_r3668081620
##########
sdks/python/apache_beam/io/fileio.py:
##########
@@ -306,37 +408,81 @@ def __init__(
'if possible')
def expand(self, pbegin) -> beam.PCollection[filesystem.FileMetadata]:
- # invoke periodic impulse
- impulse = pbegin | PeriodicImpulse(
- start_timestamp=self.start_ts,
- stop_timestamp=self.stop_ts,
- fire_interval=self.interval)
-
- # match file pattern periodically
- file_pattern = self.file_pattern
- match_files = (
- impulse
- | 'GetFilePattern' >> beam.Map(lambda x: file_pattern)
- | MatchAll(self.empty_match_treatment))
-
- # apply deduplication strategy if required
+ if Duration.of(self.interval).micros <= 0:
+ raise ValueError('MatchContinuously interval must be positive.')
if self.has_deduplication:
- # Making a Key Value so each file has its own state.
- match_files = match_files | 'ToKV' >> beam.Map(lambda x: (x.path, x))
- if self.match_upd:
- match_files = match_files | 'RemoveOldAlreadyRead' >> beam.ParDo(
- _RemoveOldDuplicates())
- else:
- match_files = match_files | 'RemoveAlreadyRead' >> beam.ParDo(
- _RemoveDuplicates())
-
- # apply windowing if required. Apply at last because deduplication relies
on
- # the global window.
+ match_files = self._match_deduplicated(pbegin)
+ else:
+ match_files = self._match_all_each_poll(pbegin)
+
+ # Apply windowing last because dedup relies on the global window.
if self.apply_windowing:
match_files = match_files | beam.WindowInto(FixedWindows(self.interval))
return match_files
+ def _match_deduplicated(self,
+ pbegin) -> beam.PCollection[filesystem.FileMetadata]:
+ # The Watch transform polls the pattern and emits each file once per key:
+ # the path, joined by the last-modified time when matching updated files.
+ # stop_timestamp bounds the watch to the polls that fall in [start, stop).
+ clock = _PollClock()
+ if self.stop_ts == MAX_TIMESTAMP:
+ termination = never()
+ else:
+ start_ts = Timestamp.of(self.start_ts)
+ stop_ts = Timestamp.of(self.stop_ts)
+ if stop_ts < start_ts:
+ raise ValueError(
+ 'MatchContinuously stop_timestamp %s precedes start_timestamp %s' %
+ (stop_ts, start_ts))
+ interval_micros = Duration.of(self.interval).micros
+ span_micros = (stop_ts - start_ts).micros
+ # Ceiling division reproduces PeriodicImpulse's tick count; the window
+ # upper bound is exclusive.
+ max_polls = -(-span_micros // interval_micros)
+ if max_polls == 0:
+ # An empty [start, stop) window never ticks in PeriodicImpulse. Fall
+ # back to the impulse path — with zero ticks dedup is moot — so the
+ # output stays empty and unbounded, rather than let Watch run its
+ # unconditional first poll.
+ return self._match_all_each_poll(pbegin)
+ termination = _WatchWindowTermination(clock, start_ts.micros, max_polls)
+ if self.match_upd:
+ output_key_fn = _file_path_and_mtime_key
+ output_key_coder = TupleCoder([StrUtf8Coder(), FloatCoder()])
Review Comment:
Please clean up the PR against the merged version of watch transform. In
particular, please check
https://github.com/apache/beam/pull/39023#discussion_r3575511681
These explicit coder settings shouldn't be needed, otherwise it suggests the
coder inference in #39023 still has some gaps
--
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]
