[Python] Refactor MatchContinuously onto the Watch transform#39461
Open
Eliaaazzz wants to merge 4 commits into
Open
[Python] Refactor MatchContinuously onto the Watch transform#39461Eliaaazzz wants to merge 4 commits into
Eliaaazzz wants to merge 4 commits into
Conversation
Add an experimental Watch transform that watches a growing set of outputs per input element. Watch.growth_of(poll_fn) runs a periodic poll loop as a splittable DoFn and emits an unbounded PCollection of (input, output) pairs. Each process() performs one poll round and self-checkpoints via defer_remainder. New outputs are deduplicated with a stable 128-bit blake2b hash of the encoded output, and a manual watermark estimator advances per poll. Per-input termination supports never() and after_total_of(); polling also stops when a poll returns PollResult.complete(). The DoFn is its own RestrictionProvider, and restriction state serializes through a tagged GrowthState coder. Tests cover termination conditions, coder round-trips, the restriction tracker claim/checkpoint/dedup logic, and DirectRunner end-to-end runs.
- Replace growth_of and the with_* builders with constructor arguments. - Infer the output coder from PollFn.default_output_coder() or a PollResult[V] return annotation via the coder registry; require determinism only of the dedup key coder, converting it GroupByKey-style with as_deterministic_coder, and add output_key_fn/output_key_coder mirroring Java's outputKeyFn/outputKeyCoder. - Rework the restriction tracker to Java's design: try_claim takes the (PollResult, termination_state) round and validates it against the restriction; try_split derives the replay primary and merged residual. - Match Java's watermark handling: seed the estimator from the input timestamp and advance it to the poll watermark or earliest new output when resuming.
The replay round leaves the watermark at the seed and reports no residual. A run that defers twice before completing keeps the watermark where the last poll left it and ends without a residual.
Route MatchContinuously through Watch when deduplication is enabled. The polling loop and the set of already-matched file ids now live in the splittable DoFn restriction, replacing the per-key state DoFns. Because the matched ids are part of the restriction, a runner with checkpointing enabled restores them after a restart and does not reprocess files. The docstring is updated accordingly. Verified on Flink 1.20: after killing the TaskManager mid stream the job restored from a checkpoint and every file was still emitted exactly once. has_deduplication=False keeps the previous PeriodicImpulse behaviour.
Contributor
|
Caution The consumer version of Gemini Code Assist on GitHub has been sunset. All code review activity has officially ceased. |
Contributor
|
Assigning reviewers: R: @tvalentyn for label python. Note: If you would like to opt out of this review, comment Available commands:
The PR bot will only process comments in the main thread (not review comments). |
Contributor
|
R: @Abacn |
Contributor
|
Stopping reviewer notifications for this pull request: review requested by someone other than the bot, ceding control. If you'd like to restart, comment |
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
2 participants
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Routes
fileio.MatchContinuouslythrough theWatchtransform when deduplication is enabled.Stacked on #39023. That PR adds
Watchitself and is still open, so its three commits currently show up here too. Only the last commit, "[Python] Refactor MatchContinuously onto the Watch transform", is new in this PR. Once #39023 merges this reduces to a single commit.What changes
The polling loop and the set of already-matched file ids move into the splittable DoFn restriction. The per-key state DoFns
_RemoveDuplicatesand_RemoveOldDuplicatesare removed, sinceWatchperforms the deduplication.has_deduplication=Falsekeeps the previousPeriodicImpulsebehaviour, so that path is unchanged.Behaviour change worth calling out
Because the matched ids are part of the restriction, a runner with checkpointing enabled restores them after a restart and does not reprocess files. The class docstring previously stated the opposite, that already processed files are reprocessed on restart, which was accurate for the earlier memory-only implementation. The docstring is updated in this PR.
Validation
Fault tolerance on Flink 1.20 with checkpointing enabled: two files present at start, two added while running, then the TaskManager was killed mid stream. The JobManager restored the job from checkpoint 3 and every file was still emitted exactly once, with no reprocessing.
Also exercised on Dataflow Runner v2 reading a real GCS prefix: files present at startup and files added to the bucket mid run were each emitted exactly once.
Unit tests:
watch_test.py28 passed,fileio_test.pyMatchContinuously tests 8 passed. Formatted with yapf 0.43.0 and isort 7.0.0.Thank you for your contribution! Follow this checklist to help us incorporate your contribution quickly and easily:
R: @username).addresses #123), if applicable. This will automatically add a link to the pull request in the issue. If you would like the issue to automatically close on merging the pull request, commentfixes #<ISSUE NUMBER>instead.CHANGES.mdwith noteworthy changes.See the Contributor Guide for more tips on how to make review process smoother.
To check the build health, please visit https://github.com/apache/beam/blob/master/.test-infra/BUILD_STATUS.md