diff --git a/website/www/site/content/en/blog/python-unboundedsource-watch.md b/website/www/site/content/en/blog/python-unboundedsource-watch.md new file mode 100644 index 000000000000..be7f15af1381 --- /dev/null +++ b/website/www/site/content/en/blog/python-unboundedsource-watch.md @@ -0,0 +1,187 @@ +--- +title: "UnboundedSource and the Watch Transform in the Apache Beam Python SDK" +date: 2026-09-10T00:00:00+10:00 +categories: + - blog + - gsoc +authors: + - eliaaazzz +--- + + +The Apache Beam Python SDK now has an `UnboundedSource` API for writing custom +unbounded sources and a `Watch` transform for repeatedly polling an input that +keeps growing. I built both during my Google Summer of Code 2026 project with +Apache Beam, mentored by Yi Hu. + + + +This post describes both APIs as of Beam 2.77.0. + +## Motivation + +Writing a connector for a message broker or database change feed means deciding +how to read records, save a position, and resume after a failure. Python already +supported custom streaming reads through a splittable DoFn (SDF). Using one +also meant learning how to represent work as a restriction, hand unfinished +work back to the runner, and report progress through a watermark estimator. +`UnboundedSource` wraps that machinery in a reader API so source authors can +focus on their connector's reading and checkpoint logic. + +Polling a growing input raises a related problem: how to remember which results +have already been emitted. Python's `fileio.MatchContinuously` could poll for +new files, but its deduplication state grew with the number of matched paths. +`Watch` makes this polling logic reusable for other inputs, such as an API that +lists newly available records. Its opt-in `timestamp_cursor` mode lets old +deduplication history expire when the input's event times keep advancing. + +## The UnboundedSource API + +The first public Python +[`UnboundedSource` API](https://github.com/apache/beam/pull/38724) addresses a +[long-standing gap](https://github.com/apache/beam/issues/19137) between the Java +and Python SDKs. Source authors implement `UnboundedSource`, `UnboundedReader`, +and `CheckpointMark`, then read the source with `beam.io.Read(MySource())`. + +The reader exposes methods such as `start()`, `advance()`, `get_current()`, +`get_current_timestamp()`, and `get_checkpoint_mark()`. Reading must not block: +returning `False` from `start()` or `advance()` means that no record is available +now, and the reader can resume when more data arrives. The reader also +reports an event-time watermark through `get_watermark()`, which Beam uses to +track progress and determine when windows can close. A watermark of +`MAX_TIMESTAMP` signals that the source has permanently finished. + +The SDK runs the reader through an SDF. The wrapper saves the reader's +checkpoint with the unfinished work and reports its watermark to the runner. +This lets the same source implementation run on DirectRunner, Prism, Flink, and +Dataflow. Sources can split their work at pipeline startup; an active read is +not subdivided further. + +The wrapper uses bundle finalization to invoke +`CheckpointMark.finalize_checkpoint` after the runner has durably committed +the output. A message-queue source can use this hook to acknowledge consumed +messages. Finalization is best effort: a mark may never be finalized, and +retries can produce marks covering overlapping records. The hook must therefore +be idempotent. Readers can also be reused across resumed bundles on the same +worker, with idle readers evicted from a bounded cache, reducing the need to +reopen connections. + +Mentor review led me to limit how many records a reader can emit and how long +it can run before yielding. The wrapper checks these limits between reads. +A busy source needs to yield regularly so the runner can commit its progress +and finalize checkpoints. The +[Python I/O connector guide](/documentation/io/developing-io-python/#unboundedsource) +includes an example source and explains the API's lifecycle. + +## The Watch transform + +The Python [`Watch` transform](https://github.com/apache/beam/pull/39023) ports +Java's polling transform. For each input element, it calls a user-supplied poll +function, emits newly discovered outputs, and saves progress between rounds. +Polling stops when the poll reports completion or a termination condition fires. +The API includes `PollFn`, `PollResult`, and the `never()` and `after_total_of()` +termination conditions. + +A single SDF manages each input's polling, duplicate suppression, output, +waiting, and termination. For example, a poll can repeatedly list files under +a prefix while `Watch` remembers which results it has already emitted. Keeping +this lifecycle together also lets the transform save its deduplication state +with its progress. + +An output's identity is the hash of its encoded key. The key defaults to the +output itself, and `output_key_fn` can select another identity. `Watch` requires +a deterministic key coder so equal keys produce the same fingerprint across +workers and after a restart. A coder with no deterministic form is rejected +when the pipeline is built. + +The default deduplication mode retains a hash for every distinct output key, +so its history grows throughout a long-running watch. This also allows the +transform to recognize an item seen much earlier. The opt-in +[`timestamp_cursor` mode](https://github.com/apache/beam/pull/39090) addresses +this [state-growth problem](https://github.com/apache/beam/issues/18459) by +letting history expire as event time advances. + +The cursor records the greatest emitted event time. Outputs more than +`allowed_lateness` behind it are skipped, including previously unseen ones, +and hashes older than that threshold can be discarded. This suits inputs +arriving in roughly non-decreasing event time. Increasing `allowed_lateness` +accommodates older arrivals while retaining more history. The cursor itself is +a single timestamp; the retained hashes depend on the keys within that time +range. In cursor mode, an item must keep its original event time across polls; +assigning it a new timestamp on every poll can cause it to be emitted again +after its hash expires. + +[Refactoring `MatchContinuously` onto `Watch`](https://github.com/apache/beam/pull/39461) +replaced its per-file state entries with the `Watch` restriction, so continuous +file matching can use cursor mode and stop accumulating an entry for every file +it has ever matched. The existing implementation remains for users who disable +duplicate suppression. The cursor design was also +[ported back to Java](https://github.com/apache/beam/pull/39746). + +## Validation across runners + +I tested both transforms on DirectRunner, Prism, Flink, and Dataflow. The runs +covered pause and resume behavior, acknowledgments, watermarks, and polling. +The `UnboundedSource` wrapper passed five end-to-end tests submitted +as Dataflow streaming jobs. For `MatchContinuously` on Flink, testing included +killing a worker during a run and restoring from a checkpoint. Prism tests +added files while a watch was running and checked that both deduplication modes +emitted them once and terminated on time. + +These runs exposed issues beyond the SDK implementations: + +- [Flink](https://github.com/apache/beam/pull/39191) accumulated state entries + when an SDF saved unfinished work. Reusing a state entry addressed the growth. +- [Prism](https://github.com/apache/beam/pull/39572) could leave downstream + records unprocessed when a source paused and resumed without advancing its + watermark. Consumers with new data are now scheduled in that case. +- [Portable Spark batch](https://github.com/apache/beam/pull/39331) gained + support for retaining and resuming unfinished SDF work. + +The work also produced a [local Flink contributor guide](https://github.com/apache/beam/pull/39580), +documenting the cluster setup used to reproduce and investigate streaming +behavior. + +## Benchmarks + +The [local benchmarks](https://github.com/Eliaaazzz/gsoc-2026-beam#6-validation-and-benchmarks) +measured `UnboundedSource` throughput and checkpoint cadence, and `Watch` +deduplication overhead as the polled set grew. + +For `UnboundedSource`, an in-memory source supplied one million records to +isolate the wrapper's overhead from external I/O. On Prism, a cap of 1,000 +records per invocation produced 1,001 self-checkpoints and about 34,000 records +per second. Raising the cap to 10,000 reduced the self-checkpoint count to 101 +and reached about 44,000 records per second. A cap of 100,000 reduced the count +to 11, with throughput still around 44,000 records per second. Throughput was +measured from the first record to the last, excluding runner startup. + +The `Watch` benchmark repeatedly listed a set that gained 2,000 items per round +for 100 rounds. Each item retained its original event time. Both modes emitted +all 200,000 items once. Cursor mode reduced total time from 111 to 24 seconds +on DirectRunner and from 59 to 15 seconds on Prism. These single-machine +experiments show how checkpoint frequency and growing deduplication history +affect the transforms; distributed benchmarks remain future work. + +## Remaining work + +Both Python APIs remain experimental, and +[Spark streaming SDF support](https://github.com/apache/beam/issues/19468) is +still open. The [full project report](https://github.com/Eliaaazzz/gsoc-2026-beam) +includes the contribution list, documentation, validation details, and benchmark +methodology. + +Thank you to my mentor, Yi Hu, and the Apache Beam community for their guidance +and reviews throughout Google Summer of Code 2026. diff --git a/website/www/site/data/authors.yml b/website/www/site/data/authors.yml index d98f2281bddc..2f03952d6d62 100644 --- a/website/www/site/data/authors.yml +++ b/website/www/site/data/authors.yml @@ -60,6 +60,9 @@ dhalperi: name: Dan Halperin email: dhalperi@apache.org twitter: +eliaaazzz: + name: Elia Liu + email: elialiulzy@gmail.com emilymye: name: Emily Ye email: emilyye@apache.org