Eliaaazzz commented on code in PR #39461:
URL: https://github.com/apache/beam/pull/39461#discussion_r3672435672
##########
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:
One more follow up commit, 4587416, from the same self review pass: the poll
return type is annotated PollResult[FileMetadata], typing.Tuple key inference
is covered next to the native form, the third party test imports are ordered,
and the remaining Java references in test comments are gone.
--
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]
