From 7aed93c11e436403b4cddc2017cc1138430744ea Mon Sep 17 00:00:00 2001 From: Eliaazzz Date: Thu, 10 Sep 2026 21:30:09 +1000 Subject: [PATCH 1/9] [Website] Add GSoC 2026 Python streaming blog post --- .../en/blog/gsoc-26-python-streaming.md | 91 +++++++++++++++++++ website/www/site/data/authors.yml | 2 + 2 files changed, 93 insertions(+) create mode 100644 website/www/site/content/en/blog/gsoc-26-python-streaming.md diff --git a/website/www/site/content/en/blog/gsoc-26-python-streaming.md b/website/www/site/content/en/blog/gsoc-26-python-streaming.md new file mode 100644 index 000000000000..b8a6d6457513 --- /dev/null +++ b/website/www/site/content/en/blog/gsoc-26-python-streaming.md @@ -0,0 +1,91 @@ +--- +title: "Google Summer of Code 2026: Native Streaming Transforms for the Python SDK" +date: 2026-09-10T00:00:00+10:00 +categories: + - blog + - gsoc +authors: + - eliaaazzz +--- + + +During Google Summer of Code 2026, I added two streaming APIs to the Apache Beam +Python SDK: `UnboundedSource` for custom streaming sources and `Watch` for polling +growing datasets. The project also improved continuous file matching and addressed +runner issues found while testing the new transforms. + + + +## Custom streaming sources + +The new [`UnboundedSource` API](https://github.com/apache/beam/pull/38724) provides +a reader interface for sources such as message queues and change feeds. A source +author implements `UnboundedSource`, `UnboundedReader`, and `CheckpointMark`, then +reads the source with `beam.io.Read(MySource())`. + +The SDK wraps the reader in a splittable DoFn (SDF), which lets runners pause and +resume reads while preserving progress and reporting event-time watermarks. +Bundle finalization lets a source acknowledge records after the runner commits +their output. The wrapper also limits each read invocation so a busy source +periodically yields work to the runner. + +## Watching growing datasets + +The [`Watch` transform](https://github.com/apache/beam/pull/39023) repeatedly calls +a poll function, emits new outputs, and stops when polling completes or a +termination condition is met. This supports use cases such as discovering files +as they arrive. + +A key design question was how much history to retain for duplicate suppression. +By default, `Watch` stores a hash for every distinct output key. That history can +grow throughout a long-running pipeline. The opt-in +[`timestamp_cursor` mode](https://github.com/apache/beam/pull/39090) limits the +retained history by event time. Outputs more than `allowed_lateness` behind the +greatest emitted event time are treated as already seen and dropped. This mode +suits sources whose outputs arrive in roughly non-decreasing event time. + +[`MatchContinuously` now uses `Watch`](https://github.com/apache/beam/pull/39461), +making the same option available for continuous file matching. The cursor design +was also [ported to Java](https://github.com/apache/beam/pull/39746). + +## Lessons from runner validation + +I validated the transforms on DirectRunner, Prism, Flink, and Dataflow. Testing +empty polls, checkpoint recovery, and long-running pipelines exposed runner +behavior that short unit tests could miss: + +- Flink accumulated state entries each time an SDF saved unfinished work. + [Reusing a state entry](https://github.com/apache/beam/pull/39191) addressed the + growth. +- Prism needed to [schedule downstream consumers](https://github.com/apache/beam/pull/39572) + when a source paused without emitting data, and to + [honor requested resume delays](https://github.com/apache/beam/pull/39849). +- Portable Spark batch needed to + [retain and resume unfinished SDF work](https://github.com/apache/beam/pull/39331). + Streaming SDF support remains [open](https://github.com/apache/beam/issues/19468). + +These tests made runner validation part of the API design process. Correct +polling depends on how runners preserve progress, advance watermarks, and +schedule work after a pause. + +## Next steps + +Both Python APIs remain experimental. Follow-up work includes Spark streaming +SDF support and distributed benchmarks. The +[final report](https://github.com/Eliaaazzz/gsoc-2026-beam) includes the complete +contribution list, validation details, and local benchmark results. + +Thank you to my mentor, Yi Hu, and the Apache Beam community for their guidance +and reviews throughout the project. I look forward to continuing this work. diff --git a/website/www/site/data/authors.yml b/website/www/site/data/authors.yml index d98f2281bddc..6c30d4cab05a 100644 --- a/website/www/site/data/authors.yml +++ b/website/www/site/data/authors.yml @@ -60,6 +60,8 @@ dhalperi: name: Dan Halperin email: dhalperi@apache.org twitter: +eliaaazzz: + name: Elia Liu emilymye: name: Emily Ye email: emilyye@apache.org From 1aa164b00d281ffc1965099e8f254aea571b86ec Mon Sep 17 00:00:00 2001 From: Eliaazzz Date: Thu, 10 Sep 2026 22:03:35 +1000 Subject: [PATCH 2/9] [Website] Focus streaming blog on practical API use --- .../en/blog/gsoc-26-python-streaming.md | 91 --------------- .../python-unbounded-sources-and-polling.md | 105 ++++++++++++++++++ 2 files changed, 105 insertions(+), 91 deletions(-) delete mode 100644 website/www/site/content/en/blog/gsoc-26-python-streaming.md create mode 100644 website/www/site/content/en/blog/python-unbounded-sources-and-polling.md diff --git a/website/www/site/content/en/blog/gsoc-26-python-streaming.md b/website/www/site/content/en/blog/gsoc-26-python-streaming.md deleted file mode 100644 index b8a6d6457513..000000000000 --- a/website/www/site/content/en/blog/gsoc-26-python-streaming.md +++ /dev/null @@ -1,91 +0,0 @@ ---- -title: "Google Summer of Code 2026: Native Streaming Transforms for the Python SDK" -date: 2026-09-10T00:00:00+10:00 -categories: - - blog - - gsoc -authors: - - eliaaazzz ---- - - -During Google Summer of Code 2026, I added two streaming APIs to the Apache Beam -Python SDK: `UnboundedSource` for custom streaming sources and `Watch` for polling -growing datasets. The project also improved continuous file matching and addressed -runner issues found while testing the new transforms. - - - -## Custom streaming sources - -The new [`UnboundedSource` API](https://github.com/apache/beam/pull/38724) provides -a reader interface for sources such as message queues and change feeds. A source -author implements `UnboundedSource`, `UnboundedReader`, and `CheckpointMark`, then -reads the source with `beam.io.Read(MySource())`. - -The SDK wraps the reader in a splittable DoFn (SDF), which lets runners pause and -resume reads while preserving progress and reporting event-time watermarks. -Bundle finalization lets a source acknowledge records after the runner commits -their output. The wrapper also limits each read invocation so a busy source -periodically yields work to the runner. - -## Watching growing datasets - -The [`Watch` transform](https://github.com/apache/beam/pull/39023) repeatedly calls -a poll function, emits new outputs, and stops when polling completes or a -termination condition is met. This supports use cases such as discovering files -as they arrive. - -A key design question was how much history to retain for duplicate suppression. -By default, `Watch` stores a hash for every distinct output key. That history can -grow throughout a long-running pipeline. The opt-in -[`timestamp_cursor` mode](https://github.com/apache/beam/pull/39090) limits the -retained history by event time. Outputs more than `allowed_lateness` behind the -greatest emitted event time are treated as already seen and dropped. This mode -suits sources whose outputs arrive in roughly non-decreasing event time. - -[`MatchContinuously` now uses `Watch`](https://github.com/apache/beam/pull/39461), -making the same option available for continuous file matching. The cursor design -was also [ported to Java](https://github.com/apache/beam/pull/39746). - -## Lessons from runner validation - -I validated the transforms on DirectRunner, Prism, Flink, and Dataflow. Testing -empty polls, checkpoint recovery, and long-running pipelines exposed runner -behavior that short unit tests could miss: - -- Flink accumulated state entries each time an SDF saved unfinished work. - [Reusing a state entry](https://github.com/apache/beam/pull/39191) addressed the - growth. -- Prism needed to [schedule downstream consumers](https://github.com/apache/beam/pull/39572) - when a source paused without emitting data, and to - [honor requested resume delays](https://github.com/apache/beam/pull/39849). -- Portable Spark batch needed to - [retain and resume unfinished SDF work](https://github.com/apache/beam/pull/39331). - Streaming SDF support remains [open](https://github.com/apache/beam/issues/19468). - -These tests made runner validation part of the API design process. Correct -polling depends on how runners preserve progress, advance watermarks, and -schedule work after a pause. - -## Next steps - -Both Python APIs remain experimental. Follow-up work includes Spark streaming -SDF support and distributed benchmarks. The -[final report](https://github.com/Eliaaazzz/gsoc-2026-beam) includes the complete -contribution list, validation details, and local benchmark results. - -Thank you to my mentor, Yi Hu, and the Apache Beam community for their guidance -and reviews throughout the project. I look forward to continuing this work. diff --git a/website/www/site/content/en/blog/python-unbounded-sources-and-polling.md b/website/www/site/content/en/blog/python-unbounded-sources-and-polling.md new file mode 100644 index 000000000000..86dabb053d7c --- /dev/null +++ b/website/www/site/content/en/blog/python-unbounded-sources-and-polling.md @@ -0,0 +1,105 @@ +--- +title: "Unbounded Sources and Continuous Polling in the Beam Python SDK" +date: 2026-09-10T00:00:00+10:00 +categories: + - blog + - gsoc +authors: + - eliaaazzz +--- + + +A streaming pipeline may read messages from a queue or discover files that arrive +over time. These sources need to preserve progress, report event-time watermarks, +and wait for more data. Two new APIs in the Beam Python SDK support these patterns: +`UnboundedSource` for reader-based sources and `Watch` for periodic polling. + + + +## Reading from a resumable source + +Use [`UnboundedSource`](https://github.com/apache/beam/pull/38724) when a source +exposes a reader and a position from which it can resume. A source author +implements `UnboundedSource`, `UnboundedReader`, and `CheckpointMark`, then reads +the source with `beam.io.Read(MySource())`. + +The SDK wraps the reader in a splittable DoFn (SDF), allowing runners to pause and +resume reads. The reader supplies timestamps, watermarks, and checkpoint marks; +the wrapper handles the SDF lifecycle. After the runner commits output, bundle +finalization can invoke the checkpoint's acknowledgement hook. Each read +invocation is also limited so a busy source periodically yields to the runner. + +## Polling for new files + +Use [`Watch`](https://github.com/apache/beam/pull/39023) when a source can be +queried periodically. A poll function returns timestamped outputs, and `Watch` +handles duplicate suppression, scheduling, and termination. + +[`MatchContinuously` now uses `Watch`](https://github.com/apache/beam/pull/39461) +for deduplicated file discovery. This local pipeline checks a directory every +five seconds and prints newly discovered file paths: + +{{< highlight py >}} +import os + +import apache_beam as beam +from apache_beam.io import fileio + +with beam.Pipeline() as pipeline: + ( + pipeline + | fileio.MatchContinuously(os.path.join("incoming", "*.json"), interval=5) + | beam.Map(lambda metadata: print(metadata.path)) + ) +{{< /highlight >}} + +The pipeline keeps polling until cancelled. By default, discovery deduplicates +by path, so later changes to an existing file do not emit it again. + +## Limiting deduplication history + +Repeated polls can return the same files. Remembering every output key prevents +duplicates, but that history grows as new files arrive. In `Watch`, setting +[`timestamp_cursor=True`](https://github.com/apache/beam/pull/39090) lets history +expire as event time advances. Outputs more than `allowed_lateness` behind the +greatest emitted event time are also skipped, including previously unseen ones. + +This suits sources with sufficiently ordered timestamps. Retained state still +depends on how many keys fall within that time range. For `MatchContinuously`, +cursor mode uses file modification times and tracks updates to existing paths; +files discovered with older modification times can be skipped. The cursor +design was also [ported to Java](https://github.com/apache/beam/pull/39746). + +## What runner validation revealed + +Validation on DirectRunner, Prism, Flink, and Dataflow exposed issues in how +runners resume work. In Prism, a source could repeatedly emit records and pause +without advancing its watermark. Downstream scheduling depended on watermark +advancement, leaving records waiting while the source continued running. The +[fix](https://github.com/apache/beam/pull/39572) schedules consumers with new data +even when the source watermark stays unchanged, while retaining the readiness +checks needed for side inputs and aggregations. + +Testing also led to fixes for +[Flink checkpoint state growth](https://github.com/apache/beam/pull/39191) and +[Prism resume delays](https://github.com/apache/beam/pull/39849), alongside +[SDF self-checkpointing support in portable Spark batch](https://github.com/apache/beam/pull/39331). +[Spark streaming SDF support](https://github.com/apache/beam/issues/19468) remains +open. Both Python APIs are experimental. + +This work was developed during Google Summer of Code 2026 with guidance from +Yi Hu and the Apache Beam community. The +[full project report](https://github.com/Eliaaazzz/gsoc-2026-beam) includes the +contributions, validation details, and local benchmarks. From 82bfa82fdb7d8921959bdaec6502e183855e3f2d Mon Sep 17 00:00:00 2001 From: Eliaazzz Date: Thu, 10 Sep 2026 22:26:38 +1000 Subject: [PATCH 3/9] [Website] Align streaming blog with project report --- .../python-unbounded-sources-and-polling.md | 105 ------------------ .../en/blog/python-unboundedsource-watch.md | 103 +++++++++++++++++ 2 files changed, 103 insertions(+), 105 deletions(-) delete mode 100644 website/www/site/content/en/blog/python-unbounded-sources-and-polling.md create mode 100644 website/www/site/content/en/blog/python-unboundedsource-watch.md diff --git a/website/www/site/content/en/blog/python-unbounded-sources-and-polling.md b/website/www/site/content/en/blog/python-unbounded-sources-and-polling.md deleted file mode 100644 index 86dabb053d7c..000000000000 --- a/website/www/site/content/en/blog/python-unbounded-sources-and-polling.md +++ /dev/null @@ -1,105 +0,0 @@ ---- -title: "Unbounded Sources and Continuous Polling in the Beam Python SDK" -date: 2026-09-10T00:00:00+10:00 -categories: - - blog - - gsoc -authors: - - eliaaazzz ---- - - -A streaming pipeline may read messages from a queue or discover files that arrive -over time. These sources need to preserve progress, report event-time watermarks, -and wait for more data. Two new APIs in the Beam Python SDK support these patterns: -`UnboundedSource` for reader-based sources and `Watch` for periodic polling. - - - -## Reading from a resumable source - -Use [`UnboundedSource`](https://github.com/apache/beam/pull/38724) when a source -exposes a reader and a position from which it can resume. A source author -implements `UnboundedSource`, `UnboundedReader`, and `CheckpointMark`, then reads -the source with `beam.io.Read(MySource())`. - -The SDK wraps the reader in a splittable DoFn (SDF), allowing runners to pause and -resume reads. The reader supplies timestamps, watermarks, and checkpoint marks; -the wrapper handles the SDF lifecycle. After the runner commits output, bundle -finalization can invoke the checkpoint's acknowledgement hook. Each read -invocation is also limited so a busy source periodically yields to the runner. - -## Polling for new files - -Use [`Watch`](https://github.com/apache/beam/pull/39023) when a source can be -queried periodically. A poll function returns timestamped outputs, and `Watch` -handles duplicate suppression, scheduling, and termination. - -[`MatchContinuously` now uses `Watch`](https://github.com/apache/beam/pull/39461) -for deduplicated file discovery. This local pipeline checks a directory every -five seconds and prints newly discovered file paths: - -{{< highlight py >}} -import os - -import apache_beam as beam -from apache_beam.io import fileio - -with beam.Pipeline() as pipeline: - ( - pipeline - | fileio.MatchContinuously(os.path.join("incoming", "*.json"), interval=5) - | beam.Map(lambda metadata: print(metadata.path)) - ) -{{< /highlight >}} - -The pipeline keeps polling until cancelled. By default, discovery deduplicates -by path, so later changes to an existing file do not emit it again. - -## Limiting deduplication history - -Repeated polls can return the same files. Remembering every output key prevents -duplicates, but that history grows as new files arrive. In `Watch`, setting -[`timestamp_cursor=True`](https://github.com/apache/beam/pull/39090) lets history -expire as event time advances. Outputs more than `allowed_lateness` behind the -greatest emitted event time are also skipped, including previously unseen ones. - -This suits sources with sufficiently ordered timestamps. Retained state still -depends on how many keys fall within that time range. For `MatchContinuously`, -cursor mode uses file modification times and tracks updates to existing paths; -files discovered with older modification times can be skipped. The cursor -design was also [ported to Java](https://github.com/apache/beam/pull/39746). - -## What runner validation revealed - -Validation on DirectRunner, Prism, Flink, and Dataflow exposed issues in how -runners resume work. In Prism, a source could repeatedly emit records and pause -without advancing its watermark. Downstream scheduling depended on watermark -advancement, leaving records waiting while the source continued running. The -[fix](https://github.com/apache/beam/pull/39572) schedules consumers with new data -even when the source watermark stays unchanged, while retaining the readiness -checks needed for side inputs and aggregations. - -Testing also led to fixes for -[Flink checkpoint state growth](https://github.com/apache/beam/pull/39191) and -[Prism resume delays](https://github.com/apache/beam/pull/39849), alongside -[SDF self-checkpointing support in portable Spark batch](https://github.com/apache/beam/pull/39331). -[Spark streaming SDF support](https://github.com/apache/beam/issues/19468) remains -open. Both Python APIs are experimental. - -This work was developed during Google Summer of Code 2026 with guidance from -Yi Hu and the Apache Beam community. The -[full project report](https://github.com/Eliaaazzz/gsoc-2026-beam) includes the -contributions, validation details, and local benchmarks. 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..f882c03d621e --- /dev/null +++ b/website/www/site/content/en/blog/python-unboundedsource-watch.md @@ -0,0 +1,103 @@ +--- +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 includes an `UnboundedSource` API for custom +unbounded sources and a `Watch` transform for repeatedly polling growing inputs. +This project brought both APIs to Python, improved `Watch` deduplication, and +addressed runner issues found while validating the new transforms. + + + +## 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 SDK runs the reader through a splittable DoFn (SDF), which allows a read to +pause and resume while preserving its progress. The wrapper handles +checkpointing and event-time watermarks, and uses bundle finalization to invoke +`CheckpointMark.finalize_checkpoint` after the runner has durably committed the +output. A source can use this hook to acknowledge consumed messages. + +Each invocation is limited by record count and elapsed time so a busy source +periodically yields to the runner. This was an important design refinement from +mentor review. + +## 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. + +Deduplication was a central design challenge. The default mode retains a hash +for every distinct output key, so its history grows throughout a long-running +watch. The opt-in +[`timestamp_cursor` mode](https://github.com/apache/beam/pull/39090) lets history +expire as event time advances. Outputs more than `allowed_lateness` behind the +greatest emitted event time are also skipped, including previously unseen ones. +This suits inputs arriving in roughly non-decreasing event time; retained state +depends on the keys within that time range. + +[Refactoring `MatchContinuously` onto `Watch`](https://github.com/apache/beam/pull/39461) +made cursor mode available for continuous file matching. The same design was +also [ported back to Java](https://github.com/apache/beam/pull/39746). + +## Validation across runners + +Both transforms were exercised on DirectRunner, Prism, Flink, and Dataflow. +Long-running pipelines, checkpoint recovery, and polling 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. + +## Benchmarks and remaining work + +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. On Prism, `UnboundedSource` +processed about 34,000 to 44,000 records per second across checkpoint settings, +excluding startup. In the 200,000-output `Watch` benchmark, cursor mode reduced +total time from 111 to 24 seconds on DirectRunner and from 59 to 15 seconds on +Prism. These were single-machine experiments; distributed benchmarks remain +future 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. From 0965209c5aa22ceff0148451c71533bc0240c9d5 Mon Sep 17 00:00:00 2001 From: Eliaazzz Date: Thu, 10 Sep 2026 22:36:08 +1000 Subject: [PATCH 4/9] [Website] Expand streaming blog design and validation --- .../en/blog/python-unboundedsource-watch.md | 120 +++++++++++++----- 1 file changed, 90 insertions(+), 30 deletions(-) diff --git a/website/www/site/content/en/blog/python-unboundedsource-watch.md b/website/www/site/content/en/blog/python-unboundedsource-watch.md index f882c03d621e..b20d53c502c3 100644 --- a/website/www/site/content/en/blog/python-unboundedsource-watch.md +++ b/website/www/site/content/en/blog/python-unboundedsource-watch.md @@ -23,8 +23,10 @@ limitations under the License. The Apache Beam Python SDK now includes an `UnboundedSource` API for custom unbounded sources and a `Watch` transform for repeatedly polling growing inputs. -This project brought both APIs to Python, improved `Watch` deduplication, and -addressed runner issues found while validating the new transforms. +These APIs give source authors control over reading a continuous stream or +discovering new items through repeated queries. This project brought both APIs +to Python, improved `Watch` deduplication, and addressed runner issues found +while validating the new transforms. @@ -36,15 +38,32 @@ The first public Python and Python SDKs. Source authors implement `UnboundedSource`, `UnboundedReader`, and `CheckpointMark`, then read the source with `beam.io.Read(MySource())`. -The SDK runs the reader through a splittable DoFn (SDF), which allows a read to -pause and resume while preserving its progress. The wrapper handles -checkpointing and event-time watermarks, and uses bundle finalization to invoke -`CheckpointMark.finalize_checkpoint` after the runner has durably committed the -output. A source can use this hook to acknowledge consumed messages. - -Each invocation is limited by record count and elapsed time so a busy source -periodically yields to the runner. This was an important design refinement from -mentor review. +The reader exposes methods such as `start()`, `advance()`, `get_current()`, and +`get_checkpoint_mark()`. Returning `False` from `advance()` means that no record +is available now; 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 a splittable DoFn (SDF), Beam's mechanism for +managing work that can pause and resume. 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. 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. + +The wrapper checks the record count and elapsed time between reads, yielding +when either limit is reached. This refinement came from mentor review: +correctly saving progress also requires giving the runner regular opportunities +to take over. The [Python I/O connector guide](https://beam.apache.org/documentation/io/developing-io-python/) +documents the API and its lifecycle. ## The Watch transform @@ -55,24 +74,50 @@ 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. -Deduplication was a central design challenge. The default mode retains a hash -for every distinct output key, so its history grows throughout a long-running -watch. The opt-in -[`timestamp_cursor` mode](https://github.com/apache/beam/pull/39090) lets history -expire as event time advances. Outputs more than `allowed_lateness` behind the -greatest emitted event time are also skipped, including previously unseen ones. -This suits inputs arriving in roughly non-decreasing event time; retained state -depends on the keys within that time range. +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. [Refactoring `MatchContinuously` onto `Watch`](https://github.com/apache/beam/pull/39461) -made cursor mode available for continuous file matching. The same design was -also [ported back to Java](https://github.com/apache/beam/pull/39746). +made cursor mode available for continuous file matching and saved deduplication +history with pipeline checkpoints. 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 Both transforms were exercised on DirectRunner, Prism, Flink, and Dataflow. -Long-running pipelines, checkpoint recovery, and polling exposed issues beyond -the SDK implementations: +Validation 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. @@ -82,16 +127,31 @@ the SDK implementations: - [Portable Spark batch](https://github.com/apache/beam/pull/39331) gained support for retaining and resuming unfinished SDF work. -## Benchmarks and remaining 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. On Prism, `UnboundedSource` -processed about 34,000 to 44,000 records per second across checkpoint settings, -excluding startup. In the 200,000-output `Watch` benchmark, cursor mode reduced -total time from 111 to 24 seconds on DirectRunner and from 59 to 15 seconds on -Prism. These were single-machine experiments; distributed benchmarks remain -future work. +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 100,000 reduced the self-checkpoint count to 11 +and reached about 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 From 451a38668077de5225b9915402ff251be644a263 Mon Sep 17 00:00:00 2001 From: Eliaazzz Date: Fri, 11 Sep 2026 14:49:55 +1000 Subject: [PATCH 5/9] [Website] Address review on the GSoC streaming blog post Name the Google Summer of Code project in the intro and add a motivation section describing the gaps the two APIs fill. Describe what the MatchContinuously refactor changed about its per-file state. List a contact email on the author entry. --- .../en/blog/python-unboundedsource-watch.md | 33 ++++++++++++++----- website/www/site/data/authors.yml | 1 + 2 files changed, 25 insertions(+), 9 deletions(-) diff --git a/website/www/site/content/en/blog/python-unboundedsource-watch.md b/website/www/site/content/en/blog/python-unboundedsource-watch.md index b20d53c502c3..8d229369a358 100644 --- a/website/www/site/content/en/blog/python-unboundedsource-watch.md +++ b/website/www/site/content/en/blog/python-unboundedsource-watch.md @@ -21,15 +21,29 @@ See the License for the specific language governing permissions and limitations under the License. --> -The Apache Beam Python SDK now includes an `UnboundedSource` API for custom -unbounded sources and a `Watch` transform for repeatedly polling growing inputs. -These APIs give source authors control over reading a continuous stream or -discovering new items through repeated queries. This project brought both APIs -to Python, improved `Watch` deduplication, and addressed runner issues found -while validating the new transforms. +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. +## Motivation + +The Java SDK has had both of these for years. A Python user who wanted to read +from a streaming system with no existing connector had to reach for a +cross-language Java connector or write an unbounded splittable DoFn directly, +wiring up a restriction tracker, a watermark estimator, and their own decision +about when to pause and resume. Polling an input that keeps growing had no +Python answer at all: there was no `Watch` transform, and +`fileio.MatchContinuously` kept one state entry per matched file path for the +life of the pipeline, so its state grew with every file the pattern had ever +seen. + +This project brought both APIs to Python, gave duplicate suppression a way to +forget outputs it has moved past, and fixed the runner bugs that validating +them turned up. + ## The UnboundedSource API The first public Python @@ -102,9 +116,10 @@ a single timestamp; the retained hashes depend on the keys within that time range. [Refactoring `MatchContinuously` onto `Watch`](https://github.com/apache/beam/pull/39461) -made cursor mode available for continuous file matching and saved deduplication -history with pipeline checkpoints. The existing implementation remains for -users who disable duplicate suppression. The cursor design was also +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 diff --git a/website/www/site/data/authors.yml b/website/www/site/data/authors.yml index 6c30d4cab05a..2f03952d6d62 100644 --- a/website/www/site/data/authors.yml +++ b/website/www/site/data/authors.yml @@ -62,6 +62,7 @@ dhalperi: twitter: eliaaazzz: name: Elia Liu + email: elialiulzy@gmail.com emilymye: name: Emily Ye email: emilyye@apache.org From 2eea12333f1ef5ac2ec78828229e37af602753ac Mon Sep 17 00:00:00 2001 From: Eliaazzz Date: Fri, 11 Sep 2026 15:23:13 +1000 Subject: [PATCH 6/9] [Website] Lead the blog motivation with the use cases Say what the two APIs let a Python developer build: a source for their own message queue or database change feed, and reusable polling for an input that keeps growing. Credit the splittable DoFn support that already existed, and note that checkpoint finalization is best effort. --- .../en/blog/python-unboundedsource-watch.md | 37 ++++++++++++------- 1 file changed, 23 insertions(+), 14 deletions(-) diff --git a/website/www/site/content/en/blog/python-unboundedsource-watch.md b/website/www/site/content/en/blog/python-unboundedsource-watch.md index 8d229369a358..8b8413c1373b 100644 --- a/website/www/site/content/en/blog/python-unboundedsource-watch.md +++ b/website/www/site/content/en/blog/python-unboundedsource-watch.md @@ -30,19 +30,28 @@ Apache Beam, mentored by Yi Hu. ## Motivation -The Java SDK has had both of these for years. A Python user who wanted to read -from a streaming system with no existing connector had to reach for a -cross-language Java connector or write an unbounded splittable DoFn directly, -wiring up a restriction tracker, a watermark estimator, and their own decision -about when to pause and resume. Polling an input that keeps growing had no -Python answer at all: there was no `Watch` transform, and -`fileio.MatchContinuously` kept one state entry per matched file path for the -life of the pipeline, so its state grew with every file the pattern had ever -seen. - -This project brought both APIs to Python, gave duplicate suppression a way to -forget outputs it has moved past, and fixed the runner bugs that validating -them turned up. +I wanted a Python developer to be able to read their own message queue or +database change feed without dropping into Java. Java has had +`UnboundedSource` since 2016. Python could read whatever system already had a +connector, and the ones living outside the SDK reach it through cross-language +wrappers that run a Java implementation behind an expansion service, Kafka, +Kinesis, and Debezium change data capture among them. + +Writing such a source in Python was possible before this project. An unbounded +splittable DoFn gets restriction tracking, checkpoint and resume, and watermark +estimators from Beam. Each author still had to work out the part specific to +reading a stream: express the reader's position as a restriction, decide when to +stop and hand progress back, and keep the watermark moving. `UnboundedSource` +settles that once, with a checkpoint mark carrying the resume position and a +hook that fires after the runner commits, which is where a queue source +acknowledges its messages. + +`Watch` is for the other shape of streaming input, the kind that keeps growing. +Java has had it since 2017. Python could poll for new files by composing +`PeriodicImpulse` with `MatchAll`, and with duplicate suppression on, +`fileio.MatchContinuously` held one state entry per matched path for the life of +the pipeline. I wanted polling that any source could reuse and a way to stop +that history from growing forever. ## The UnboundedSource API @@ -69,7 +78,7 @@ 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. Readers can also be reused across resumed bundles on the same worker, +messages. Finalization is best effort, so the hook has to 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. From 15c45cd6b4ac50161b3969e1d498fcbe00703912 Mon Sep 17 00:00:00 2001 From: Eliaazzz Date: Fri, 11 Sep 2026 15:33:23 +1000 Subject: [PATCH 7/9] [Website] State the motivation as the SDF learning curve An unbounded splittable DoFn could already do this. UnboundedSource exists so a source author does not have to learn restrictions, resumption, and watermark estimators first. --- .../en/blog/python-unboundedsource-watch.md | 38 ++++++++----------- 1 file changed, 16 insertions(+), 22 deletions(-) diff --git a/website/www/site/content/en/blog/python-unboundedsource-watch.md b/website/www/site/content/en/blog/python-unboundedsource-watch.md index 8b8413c1373b..e2c099604e2e 100644 --- a/website/www/site/content/en/blog/python-unboundedsource-watch.md +++ b/website/www/site/content/en/blog/python-unboundedsource-watch.md @@ -30,28 +30,22 @@ Apache Beam, mentored by Yi Hu. ## Motivation -I wanted a Python developer to be able to read their own message queue or -database change feed without dropping into Java. Java has had -`UnboundedSource` since 2016. Python could read whatever system already had a -connector, and the ones living outside the SDK reach it through cross-language -wrappers that run a Java implementation behind an expansion service, Kafka, -Kinesis, and Debezium change data capture among them. - -Writing such a source in Python was possible before this project. An unbounded -splittable DoFn gets restriction tracking, checkpoint and resume, and watermark -estimators from Beam. Each author still had to work out the part specific to -reading a stream: express the reader's position as a restriction, decide when to -stop and hand progress back, and keep the watermark moving. `UnboundedSource` -settles that once, with a checkpoint mark carrying the resume position and a -hook that fires after the runner commits, which is where a queue source -acknowledges its messages. - -`Watch` is for the other shape of streaming input, the kind that keeps growing. -Java has had it since 2017. Python could poll for new files by composing -`PeriodicImpulse` with `MatchAll`, and with duplicate suppression on, -`fileio.MatchContinuously` held one state entry per matched path for the life of -the pipeline. I wanted polling that any source could reuse and a way to stop -that history from growing forever. +Reading a custom unbounded source in Python already worked through an unbounded +splittable DoFn. The learning curve is the problem. You have to model the +reader's position as a restriction, decide when to stop and hand progress back, +and drive a watermark estimator, all before writing a line of code that talks +to your queue. `UnboundedSource` asks for `start()`, `advance()`, +`get_watermark()`, and a checkpoint mark, and the wrapper handles the SDF part. +That puts a source for your own message broker or database change feed within +reach in Python, where Kafka, Kinesis, and Debezium change data capture reach it +today through cross-language wrappers that run a Java implementation behind an +expansion service. + +`Watch` is the same kind of convenience for an input that keeps growing. Python +could poll for new files by composing `PeriodicImpulse` with `MatchAll`, and +with duplicate suppression on, `fileio.MatchContinuously` held one state entry +per matched path for the life of the pipeline. `Watch` makes the polling +reusable for any source and bounds that history. ## The UnboundedSource API From d4095713cac72f2eecb5311f5939b51822acdd82 Mon Sep 17 00:00:00 2001 From: Eliaazzz Date: Fri, 11 Sep 2026 15:57:11 +1000 Subject: [PATCH 8/9] [Website] Clarify streaming blog motivation and API behavior --- .../en/blog/python-unboundedsource-watch.md | 82 ++++++++++--------- 1 file changed, 45 insertions(+), 37 deletions(-) diff --git a/website/www/site/content/en/blog/python-unboundedsource-watch.md b/website/www/site/content/en/blog/python-unboundedsource-watch.md index e2c099604e2e..f75472be1695 100644 --- a/website/www/site/content/en/blog/python-unboundedsource-watch.md +++ b/website/www/site/content/en/blog/python-unboundedsource-watch.md @@ -28,24 +28,26 @@ Apache Beam, mentored by Yi Hu. +This post describes the implementation on Beam's `master` branch as of September +2026. The `Watch` `allowed_lateness` option and the `MatchContinuously` +integration described below are newer than Beam 2.76.0. + ## Motivation -Reading a custom unbounded source in Python already worked through an unbounded -splittable DoFn. The learning curve is the problem. You have to model the -reader's position as a restriction, decide when to stop and hand progress back, -and drive a watermark estimator, all before writing a line of code that talks -to your queue. `UnboundedSource` asks for `start()`, `advance()`, -`get_watermark()`, and a checkpoint mark, and the wrapper handles the SDF part. -That puts a source for your own message broker or database change feed within -reach in Python, where Kafka, Kinesis, and Debezium change data capture reach it -today through cross-language wrappers that run a Java implementation behind an -expansion service. - -`Watch` is the same kind of convenience for an input that keeps growing. Python -could poll for new files by composing `PeriodicImpulse` with `MatchAll`, and -with duplicate suppression on, `fileio.MatchContinuously` held one state entry -per matched path for the life of the pipeline. `Watch` makes the polling -reusable for any source and bounds that history. +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 @@ -55,15 +57,15 @@ The first public Python 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()`, and -`get_checkpoint_mark()`. Returning `False` from `advance()` means that no record -is available now; the reader can resume when more data arrives. The reader also +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 a splittable DoFn (SDF), Beam's mechanism for -managing work that can pause and resume. The wrapper saves the reader's +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 @@ -72,15 +74,18 @@ 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, so the hook has to 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. - -The wrapper checks the record count and elapsed time between reads, yielding -when either limit is reached. This refinement came from mentor review: -correctly saving progress also requires giving the runner regular opportunities -to take over. The [Python I/O connector guide](https://beam.apache.org/documentation/io/developing-io-python/) -documents the API and its lifecycle. +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 @@ -116,7 +121,9 @@ 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. +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 @@ -127,9 +134,9 @@ duplicate suppression. The cursor design was also ## Validation across runners -Both transforms were exercised on DirectRunner, Prism, Flink, and Dataflow. -Validation covered pause and resume behavior, acknowledgments, watermarks, and -polling. The `UnboundedSource` wrapper passed five end-to-end tests submitted +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 @@ -158,9 +165,10 @@ 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 100,000 reduced the self-checkpoint count to 11 -and reached about 44,000 records per second. Throughput was measured from the -first record to the last, excluding runner startup. +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 From d270a3e4071d8c2dbf610595d6bb498dfee93115 Mon Sep 17 00:00:00 2001 From: Eliaazzz Date: Sat, 12 Sep 2026 00:19:11 +1000 Subject: [PATCH 9/9] [Website] Scope the blog post by release version Readers track Beam releases, so name 2.77.0 instead of the master branch. --- .../www/site/content/en/blog/python-unboundedsource-watch.md | 4 +--- 1 file changed, 1 insertion(+), 3 deletions(-) diff --git a/website/www/site/content/en/blog/python-unboundedsource-watch.md b/website/www/site/content/en/blog/python-unboundedsource-watch.md index f75472be1695..be7f15af1381 100644 --- a/website/www/site/content/en/blog/python-unboundedsource-watch.md +++ b/website/www/site/content/en/blog/python-unboundedsource-watch.md @@ -28,9 +28,7 @@ Apache Beam, mentored by Yi Hu. -This post describes the implementation on Beam's `master` branch as of September -2026. The `Watch` `allowed_lateness` option and the `MatchContinuously` -integration described below are newer than Beam 2.76.0. +This post describes both APIs as of Beam 2.77.0. ## Motivation