This is an automated email from the ASF dual-hosted git repository.
github-actions[bot] pushed a commit to branch asf-site
in repository https://gitbox.apache.org/repos/asf/beam.git
The following commit(s) were added to refs/heads/asf-site by this push:
new aa960e51e1d Publishing website 2026/08/21 23:45:35 at commit 015e823
aa960e51e1d is described below
commit aa960e51e1d027285c4d09a2dbf6aa6e0fddeb45
Author: runner <runner@main-runner-2404-9kmqn-88jrd>
AuthorDate: Fri Aug 21 23:45:35 2026 +0000
Publishing website 2026/08/21 23:45:35 at commit 015e823
---
website/generated-content/documentation/index.xml | 223 ++++++++++++++++++++-
.../io/developing-io-python/index.html | 159 ++++++++++++++-
website/generated-content/sitemap.xml | 2 +-
3 files changed, 366 insertions(+), 18 deletions(-)
diff --git a/website/generated-content/documentation/index.xml
b/website/generated-content/documentation/index.xml
index 160911548b9..f6c567b5d1b 100644
--- a/website/generated-content/documentation/index.xml
+++ b/website/generated-content/documentation/index.xml
@@ -1674,22 +1674,35 @@ to develop tests for your source.</p>
for Beam&rsquo;s transform style guidance.</p>
<h2 id="implementing-the-source-interface">Implementing the Source
interface</h2>
<p>To create a new data source for your pipeline, you&rsquo;ll need to
provide the format-specific logic that tells the service how to read data from
your input source, and how to split your data source into multiple parts so
that multiple worker instances can read your data in parallel.</p>
+<p>If you&rsquo;re creating a data source that reads unbounded data,
also provide the
+logic for managing your source&rsquo;s watermark and checkpointing.</p>
<p>Supply the logic for your new source by creating the following
classes:</p>
<ul>
-<li>A subclass of <code>BoundedSource</code>.
<code>BoundedSource</code> is a source that reads a
-finite amount of input records. The class describes the data you want to
-read, including the data&rsquo;s location and parameters (such as how much
data to
+<li>A subclass of <code>BoundedSource</code> if you want to read a
finite (batch) data set,
+or a subclass of <code>UnboundedSource</code> if you want to read an
infinite
+(streaming) data set. The class describes the data you want to read,
+including the data&rsquo;s location and parameters (such as how much data
to
read).</li>
-<li>A subclass of <code>RangeTracker</code>.
<code>RangeTracker</code> is a thread-safe object used to
-manage a range for a given position type.</li>
+<li>For a <code>BoundedSource</code>, a subclass of
<code>RangeTracker</code>. <code>RangeTracker</code> is a
+thread-safe object used to manage a range for a given position type.</li>
+<li>For an <code>UnboundedSource</code>, a subclass of
<code>UnboundedReader</code>, which holds the
+state involved in reading the stream, and a subclass of
<code>CheckpointMark</code>,
+which records the position that a reader resumes from.</li>
<li>One or more user-facing wrapper composite transforms
(<code>PTransform</code>) that
wrap read operations. <a href="#ptransform-wrappers">PTransform
wrappers</a> discusses
why you should avoid exposing your sources, and walks through how to create
a wrapper.</li>
</ul>
-<p>You can find these classes in the
-<a
href="https://beam.apache.org/releases/pydoc/2.75.0/apache_beam.io.iobase.html";>apache_beam.io.iobase
module</a>.</p>
-<h3 id="implementing-the-boundedsource-subclass">Implementing the
BoundedSource subclass</h3>
+<p>You can find <code>BoundedSource</code> and
<code>RangeTracker</code> in the
+<a
href="https://beam.apache.org/releases/pydoc/2.75.0/apache_beam.io.iobase.html";>apache_beam.io.iobase
module</a>,
+and the unbounded classes in the
+<a
href="https://beam.apache.org/releases/pydoc/2.75.0/apache_beam.io.unbounded_source.html";>apache_beam.io.unbounded_source
module</a>.</p>
+<h3 id="implementing-the-source-subclass">Implementing the Source
subclass</h3>
+<p>Create a subclass of either <code>BoundedSource</code> or
<code>UnboundedSource</code>, depending on
+whether your data is a finite batch or an infinite stream. In either case, the
+subclass overrides the methods that a runner uses to split the data and to
+create a reader for it.</p>
+<h4 id="boundedsource">BoundedSource</h4>
<p><code>BoundedSource</code> represents a finite data set from which
the service reads, possibly in parallel. <code>BoundedSource</code>
contains a set of methods that the service uses to split the data set for
reading by multiple remote workers.</p>
<p>To implement a <code>BoundedSource</code>, your subclass must
override the following methods:</p>
<ul>
@@ -1706,7 +1719,51 @@ a wrapper.</li>
<p><code>read</code>: This method returns an iterator that reads data
from the source, with respect to the boundaries defined by the given
<code>RangeTracker</code> object.</p>
</li>
</ul>
-<h3 id="implementing-the-rangetracker-subclass">Implementing the
RangeTracker subclass</h3>
+<h4 id="unboundedsource">UnboundedSource</h4>
+<p><code>UnboundedSource</code> represents an infinite data stream
from which the runner may
+read, possibly in parallel. <code>UnboundedSource</code> contains a set
of methods that
+support streaming reads in parallel; these include
<em>checkpointing</em> for failure
+recovery and <em>watermarking</em> for estimating data completeness in
downstream parts
+of your pipeline.</p>
+<p><code>UnboundedSource</code> is experimental, and its API may
change in
+backwards-incompatible ways.</p>
+<p>To implement an <code>UnboundedSource</code>, your subclass must
override the following
+methods:</p>
+<ul>
+<li>
+<p><code>split</code>: The SDK uses this method to generate a list of
<code>UnboundedSource</code>
+objects that represent the sub-streams to read in parallel. Each sub-source
must
+be independent and must not share mutable state with its siblings, because the
+runner may read them concurrently on different workers. Return
<code>[self]</code> if the
+source cannot be split. Splitting happens once, before any checkpoint
exists.</p>
+</li>
+<li>
+<p><code>create_reader</code>: Creates the associated
<code>UnboundedReader</code> for this
+<code>UnboundedSource</code>. When <code>checkpoint_mark</code> is
<code>None</code>, the reader starts at the
+beginning of the stream. Otherwise it resumes strictly after the position that
+the mark encodes and does not re-deliver records that a previous bundle already
+read.</p>
+</li>
+<li>
+<p><code>get_checkpoint_mark_coder</code>: Returns the
<code>Coder</code> for this source&rsquo;s
+<code>CheckpointMark</code> instances. The SDK calls it while encoding
and decoding a
+reader&rsquo;s position, so it should be side-effect free and should not
perform I/O.</p>
+</li>
+</ul>
+<p>Override <code>default_output_coder</code> to return a coder for
your record type. The
+default is a pickle coder, and a tighter coder also gives the output
+<code>PCollection</code> an element type.</p>
+<p><code>UnboundedSource</code> has no per-record deduplication hook.
If your data source can
+deliver the same record more than once, drop the duplicates with the
+<a
href="https://beam.apache.org/releases/pydoc/2.75.0/apache_beam.transforms.deduplicate.html";>Deduplicate
or DeduplicatePerKey</a>
+transform after the read.</p>
+<h3
id="implementing-the-rangetracker-and-unboundedreader-subclasses">Implementing
the RangeTracker and UnboundedReader subclasses</h3>
+<p>A runner uses these classes to do the actual reading of your data set
and to
+track a reader&rsquo;s progress. A <code>BoundedSource</code> reads
through its <code>read</code> method,
+which claims positions from a <code>RangeTracker</code>. An
<code>UnboundedSource</code> reads through
+an <code>UnboundedReader</code>, which also reports a watermark and
produces the checkpoint
+marks that a runner resumes from.</p>
+<h4 id="rangetracker">RangeTracker</h4>
<p>A <code>RangeTracker</code> is a thread-safe object used to manage
the current range and current position of the reader of a
<code>BoundedSource</code> and protect concurrent access to them.</p>
<p>To implement a <code>RangeTracker</code>, you should first
familiarize yourself with the following definitions:</p>
<ul>
@@ -1771,6 +1828,58 @@ a wrapper.</li>
<li><code>fraction_consumed</code>: Returns the approximate fraction
of consumed positions in the source.</li>
</ul>
<p><strong>Note:</strong> Methods of class
<code>iobase.RangeTracker</code> may be invoked by multiple threads,
hence this class must be made thread-safe, for example, by using a single lock
object.</p>
+<h4 id="unboundedreader">UnboundedReader</h4>
+<p>An <code>UnboundedReader</code> holds the state involved in
reading one <code>UnboundedSource</code>,
+such as connections and buffers. <code>start</code> is called exactly
once, then <code>advance</code>
+is called repeatedly; whenever either returns <code>True</code>, the
current record is
+available through <code>get_current</code> and
<code>get_current_timestamp</code>.</p>
+<p>To implement an <code>UnboundedReader</code>, your subclass must
override the following
+methods:</p>
+<ul>
+<li>
+<p><code>start</code>: Initializes the reader, positions it at the
first record, and returns
+whether one is available. This is a good place for expensive
initialization.</p>
+</li>
+<li>
+<p><code>advance</code>: Advances to the next record and returns
whether one is available. A
+<code>False</code> return means that no data is available right now,
which differs from the
+end of the stream: a reader signals a permanent end by reporting a watermark of
+<code>MAX_TIMESTAMP</code>. This method should not block; return
<code>False</code> when no data is
+currently available instead of waiting for more.</p>
+</li>
+<li>
+<p><code>get_current</code>: Returns the record at the current
position, last read by <code>start</code>
+or <code>advance</code>.</p>
+</li>
+<li>
+<p><code>get_current_timestamp</code>: Returns the event-time
timestamp of the current
+record, which becomes the timestamp of the output element.</p>
+</li>
+<li>
+<p><code>get_watermark</code>: Returns a watermark, the approximate
lower bound on the
+timestamps of the records that this reader produces in the future. The runner
+uses the watermark as an estimate of data completeness in windowing and
triggers.
+The watermark is treated as monotonic.</p>
+</li>
+<li>
+<p><code>get_checkpoint_mark</code>: Returns a
<code>CheckpointMark</code> that records how far the
+reader has read. It is called only at a bundle boundary, and the mark is passed
+back to <code>create_reader</code> to resume.</p>
+</li>
+<li>
+<p><code>close</code>: Releases the reader&rsquo;s resources. The
default is a no-op.</p>
+</li>
+</ul>
+<h4 id="checkpoint-marks">Checkpoint marks</h4>
+<p>A <code>CheckpointMark</code> is a durable, serializable position
in the stream. The runner
+persists it with the coder from <code>get_checkpoint_mark_coder</code>
and hands it to
+<code>create_reader</code> when a bundle resumes or when a worker
recovers from a failure.</p>
+<p>Override <code>finalize_checkpoint</code> to acknowledge or commit
the consumed records
+upstream, for example to ack the messages on a queue. It is called after the
+runner has durably committed the work covered by that mark. Finalization is
best
+effort: a mark may never be finalized, and a retried bundle may re-cut a mark
+over an overlapping span, so acknowledge by absolute position and keep the
method
+idempotent.</p>
<h3 id="convenience-source-base-classes">Convenience Source base
classes</h3>
<p>The Beam SDK for Python contains some convenient abstract base classes
to help you easily create new sources.</p>
<h4 id="filebasedsource">FileBasedSource</h4>
@@ -1834,6 +1943,102 @@ recommended that you do not expose the code for the
source itself as
demonstrated in the example above. Use a wrapping
<code>PTransform</code> instead.
<a href="#ptransform-wrappers">PTransform wrappers</a> discusses why you
should avoid
exposing your sources, and walks through how to create a wrapper.</p>
+<h3 id="reading-from-an-unboundedsource">Reading from an
UnboundedSource</h3>
+<p>The following example, <code>QueueSource</code>, reads from a
partitioned message queue.
+<code>my_queue</code> stands in for the client library of the system you
read from.</p>
+<div class="snippet">
+<div class="notebook-skip code-snippet without_switcher">
+<a class="copy" type="button" data-bs-toggle="tooltip"
data-bs-placement="bottom" title="Copy to clipboard">
+<img src="/images/copy-icon.svg"/>
+</a>
+<pre tabindex="0"><code>import apache_beam as beam
+from apache_beam.io.unbounded_source import CheckpointMark
+from apache_beam.io.unbounded_source import UnboundedReader
+from apache_beam.io.unbounded_source import UnboundedSource
+from apache_beam.utils.timestamp import Timestamp
+class QueueCheckpointMark(CheckpointMark):
+def __init__(self, offset):
+self.offset = offset
+def finalize_checkpoint(self):
+# Acknowledging an absolute offset is safe to repeat.
+my_queue.ack_through(self.offset)
+class QueueReader(UnboundedReader):
+def __init__(self, partition, offset):
+self._partition = partition
+self._offset = offset
+self._message = None
+def start(self):
+return self.advance()
+def advance(self):
+message = my_queue.poll(self._partition, self._offset)
+if message is None:
+return False
+self._offset = message.offset
+self._message = message
+return True
+def get_current(self):
+return self._message.body
+def get_current_timestamp(self):
+return Timestamp(micros=self._message.event_time_micros)
+def get_watermark(self):
+return Timestamp(micros=my_queue.oldest_pending_micros(self._partition))
+def get_checkpoint_mark(self):
+return QueueCheckpointMark(self._offset)
+def close(self):
+my_queue.disconnect(self._partition)
+class QueueSource(UnboundedSource):
+def __init__(self, topic, partition=None):
+self._topic = topic
+self._partition = partition
+def split(self, desired_num_splits, options=None):
+if self._partition is not None:
+return [self]
+return [
+QueueSource(self._topic, partition)
+for partition in my_queue.partitions(self._topic)
+]
+def create_reader(self, options, checkpoint_mark):
+offset = None if checkpoint_mark is None else checkpoint_mark.offset
+return QueueReader(self._partition, offset)
+def get_checkpoint_mark_coder(self):
+return beam.coders.PickleCoder()
+def default_output_coder(self):
+return beam.coders.BytesCoder()</code></pre>
+</div>
+</div>
+<p>To read data from the source in your pipeline, use the
<code>Read</code> transform, which
+dispatches an <code>UnboundedSource</code> automatically:</p>
+<div class="snippet">
+<div class="notebook-skip code-snippet without_switcher">
+<a class="copy" type="button" data-bs-toggle="tooltip"
data-bs-placement="bottom" title="Copy to clipboard">
+<img src="/images/copy-icon.svg"/>
+</a>
+<pre tabindex="0"><code>with beam.Pipeline(options=pipeline_options) as
p:
+orders = p |
beam.io.Read(QueueSource(&#39;orders&#39;))</code></pre>
+</div>
+</div>
+<p>A bundle ends when the reader runs out of data, and also once the reader
has
+emitted <code>max_records_per_bundle</code> records or spent
<code>max_read_time_seconds</code> in the
+bundle, so the runner commits the checkpoint and runs finalization before the
+read resumes. An idle reader is polled again after
<code>poll_interval</code> seconds.
+Apply <code>ReadFromUnboundedSource</code> directly to change these
defaults:</p>
+<div class="snippet">
+<div class="notebook-skip code-snippet without_switcher">
+<a class="copy" type="button" data-bs-toggle="tooltip"
data-bs-placement="bottom" title="Copy to clipboard">
+<img src="/images/copy-icon.svg"/>
+</a>
+<pre tabindex="0"><code>from apache_beam.io.unbounded_source import
ReadFromUnboundedSource
+orders = p | ReadFromUnboundedSource(
+QueueSource(&#39;orders&#39;),
+poll_interval=5,
+max_records_per_bundle=1000,
+max_read_time_seconds=30)</code></pre>
+</div>
+</div>
+<p><strong>Note:</strong> As with a bounded source, we recommend that
you do not expose the code
+for the source itself to end-users. Use a wrapping
<code>PTransform</code> instead.
+<a href="#ptransform-wrappers">PTransform wrappers</a> discusses why you
should avoid
+exposing your sources, and walks through how to create a wrapper.</p>
<h2 id="using-the-filebasedsink-abstraction">Using the FileBasedSink
abstraction</h2>
<p>If your data source uses files, you can implement the <a
href="https://beam.apache.org/releases/pydoc/2.75.0/apache_beam.io.filebasedsink.html";>FileBasedSink</a>
abstraction to create a file-based sink. For other sinks, use
<code>ParDo</code>,
diff --git
a/website/generated-content/documentation/io/developing-io-python/index.html
b/website/generated-content/documentation/io/developing-io-python/index.html
index 140aa9d8943..619b6b90635 100644
--- a/website/generated-content/documentation/io/developing-io-python/index.html
+++ b/website/generated-content/documentation/io/developing-io-python/index.html
@@ -35,7 +35,7 @@
<img class=banner-img-mobile
src=/images/banners/tour-of-beam/tour-of-beam-mobile.png alt="Start Tour of
Beam"></a></div><div class=swiper-slide><a
href=https://beam.apache.org/documentation/ml/overview/><img
class=banner-img-desktop
src=/images/banners/machine-learning/machine-learning-desktop.jpg alt="Machine
Learning">
<img class=banner-img-mobile
src=/images/banners/machine-learning/machine-learning-mobile.jpg alt="Machine
Learning"></a></div></div><div class=swiper-pagination></div><div
class=swiper-button-prev></div><div
class=swiper-button-next></div></div><script
src=/js/swiper-bundle.min.min.e0e8f81b0b15728d35ff73c07f42ddbb17a108d6f23df4953cb3e60df7ade675.js></script>
<script
src=/js/sliders/top-banners.min.afa7d0a19acf7a3b28ca369490b3d401a619562a2a4c9612577be2f66a4b9855.js></script>
-<script>function showSearch(){addPlaceholder();var
e,t=document.querySelector(".searchBar");t.classList.remove("disappear"),e=document.querySelector("#iconsBar"),e.classList.add("disappear")}function
addPlaceholder(){$("input:text").attr("placeholder","What are you looking
for?")}function endSearch(){var
e,t=document.querySelector(".searchBar");t.classList.add("disappear"),e=document.querySelector("#iconsBar"),e.classList.remove("disappear")}function
blockScroll(){$("body").toggleClass(" [...]
+<script>function showSearch(){addPlaceholder();var
e,t=document.querySelector(".searchBar");t.classList.remove("disappear"),e=document.querySelector("#iconsBar"),e.classList.add("disappear")}function
addPlaceholder(){$("input:text").attr("placeholder","What are you looking
for?")}function endSearch(){var
e,t=document.querySelector(".searchBar");t.classList.add("disappear"),e=document.querySelector("#iconsBar"),e.classList.remove("disappear")}function
blockScroll(){$("body").toggleClass(" [...]
the <a href=/documentation/io/developing-io-overview/>new I/O connector
overview</a>.</p><p>To connect to a data store that isn’t supported by Beam’s
existing I/O
connectors, you must create a custom I/O connector that usually consist of a
source and a sink. All Beam sources and sinks are composite transforms;
however,
@@ -61,15 +61,72 @@ lead to data corruption or data loss (such as skipping or
duplicating
records) that can be hard to detect. You can use test harnesses and utility
methods available in the <a
href=https://github.com/apache/beam/blob/master/sdks/python/apache_beam/io/source_test_utils.py>source_test_utils
module</a>
to develop tests for your source.</p></li></ol><p>In addition, see the <a
href=/contribute/ptransform-style-guide/>PTransform style guide</a>
-for Beam’s transform style guidance.</p><h2
id=implementing-the-source-interface>Implementing the Source
interface</h2><p>To create a new data source for your pipeline, you’ll
need to provide the format-specific logic that tells the service how to read
data from your input source, and how to split your data source into multiple
parts so that multiple worker instances can read your data in
parallel.</p><p>Supply the logic for your new source by creating the following
classes:< [...]
-finite amount of input records. The class describes the data you want to
-read, including the data’s location and parameters (such as how much
data to
-read).</li><li>A subclass of <code>RangeTracker</code>.
<code>RangeTracker</code> is a thread-safe object used to
-manage a range for a given position type.</li><li>One or more user-facing
wrapper composite transforms (<code>PTransform</code>) that
+for Beam’s transform style guidance.</p><h2
id=implementing-the-source-interface>Implementing the Source
interface</h2><p>To create a new data source for your pipeline, you’ll
need to provide the format-specific logic that tells the service how to read
data from your input source, and how to split your data source into multiple
parts so that multiple worker instances can read your data in
parallel.</p><p>If you’re creating a data source that reads unbounded
data, also p [...]
+logic for managing your source’s watermark and
checkpointing.</p><p>Supply the logic for your new source by creating the
following classes:</p><ul><li>A subclass of <code>BoundedSource</code> if you
want to read a finite (batch) data set,
+or a subclass of <code>UnboundedSource</code> if you want to read an infinite
+(streaming) data set. The class describes the data you want to read,
+including the data’s location and parameters (such as how much data to
+read).</li><li>For a <code>BoundedSource</code>, a subclass of
<code>RangeTracker</code>. <code>RangeTracker</code> is a
+thread-safe object used to manage a range for a given position
type.</li><li>For an <code>UnboundedSource</code>, a subclass of
<code>UnboundedReader</code>, which holds the
+state involved in reading the stream, and a subclass of
<code>CheckpointMark</code>,
+which records the position that a reader resumes from.</li><li>One or more
user-facing wrapper composite transforms (<code>PTransform</code>) that
wrap read operations. <a href=#ptransform-wrappers>PTransform wrappers</a>
discusses
why you should avoid exposing your sources, and walks through how to create
-a wrapper.</li></ul><p>You can find these classes in the
-<a
href=https://beam.apache.org/releases/pydoc/2.75.0/apache_beam.io.iobase.html>apache_beam.io.iobase
module</a>.</p><h3 id=implementing-the-boundedsource-subclass>Implementing the
BoundedSource subclass</h3><p><code>BoundedSource</code> represents a finite
data set from which the service reads, possibly in parallel.
<code>BoundedSource</code> contains a set of methods that the service uses to
split the data set for reading by multiple remote workers.</p><p>To implement a
<code>BoundedS [...]
+a wrapper.</li></ul><p>You can find <code>BoundedSource</code> and
<code>RangeTracker</code> in the
+<a
href=https://beam.apache.org/releases/pydoc/2.75.0/apache_beam.io.iobase.html>apache_beam.io.iobase
module</a>,
+and the unbounded classes in the
+<a
href=https://beam.apache.org/releases/pydoc/2.75.0/apache_beam.io.unbounded_source.html>apache_beam.io.unbounded_source
module</a>.</p><h3 id=implementing-the-source-subclass>Implementing the Source
subclass</h3><p>Create a subclass of either <code>BoundedSource</code> or
<code>UnboundedSource</code>, depending on
+whether your data is a finite batch or an infinite stream. In either case, the
+subclass overrides the methods that a runner uses to split the data and to
+create a reader for it.</p><h4
id=boundedsource>BoundedSource</h4><p><code>BoundedSource</code> represents a
finite data set from which the service reads, possibly in parallel.
<code>BoundedSource</code> contains a set of methods that the service uses to
split the data set for reading by multiple remote workers.</p><p>To implement a
<code>BoundedSource</code>, your subclass must override the following
methods:</p><ul><li><p><code>estimate_size</code>: Services use this method to
estimate [...]
+read, possibly in parallel. <code>UnboundedSource</code> contains a set of
methods that
+support streaming reads in parallel; these include <em>checkpointing</em> for
failure
+recovery and <em>watermarking</em> for estimating data completeness in
downstream parts
+of your pipeline.</p><p><code>UnboundedSource</code> is experimental, and its
API may change in
+backwards-incompatible ways.</p><p>To implement an
<code>UnboundedSource</code>, your subclass must override the following
+methods:</p><ul><li><p><code>split</code>: The SDK uses this method to
generate a list of <code>UnboundedSource</code>
+objects that represent the sub-streams to read in parallel. Each sub-source
must
+be independent and must not share mutable state with its siblings, because the
+runner may read them concurrently on different workers. Return
<code>[self]</code> if the
+source cannot be split. Splitting happens once, before any checkpoint
exists.</p></li><li><p><code>create_reader</code>: Creates the associated
<code>UnboundedReader</code> for this
+<code>UnboundedSource</code>. When <code>checkpoint_mark</code> is
<code>None</code>, the reader starts at the
+beginning of the stream. Otherwise it resumes strictly after the position that
+the mark encodes and does not re-deliver records that a previous bundle already
+read.</p></li><li><p><code>get_checkpoint_mark_coder</code>: Returns the
<code>Coder</code> for this source’s
+<code>CheckpointMark</code> instances. The SDK calls it while encoding and
decoding a
+reader’s position, so it should be side-effect free and should not
perform I/O.</p></li></ul><p>Override <code>default_output_coder</code> to
return a coder for your record type. The
+default is a pickle coder, and a tighter coder also gives the output
+<code>PCollection</code> an element type.</p><p><code>UnboundedSource</code>
has no per-record deduplication hook. If your data source can
+deliver the same record more than once, drop the duplicates with the
+<a
href=https://beam.apache.org/releases/pydoc/2.75.0/apache_beam.transforms.deduplicate.html>Deduplicate
or DeduplicatePerKey</a>
+transform after the read.</p><h3
id=implementing-the-rangetracker-and-unboundedreader-subclasses>Implementing
the RangeTracker and UnboundedReader subclasses</h3><p>A runner uses these
classes to do the actual reading of your data set and to
+track a reader’s progress. A <code>BoundedSource</code> reads through
its <code>read</code> method,
+which claims positions from a <code>RangeTracker</code>. An
<code>UnboundedSource</code> reads through
+an <code>UnboundedReader</code>, which also reports a watermark and produces
the checkpoint
+marks that a runner resumes from.</p><h4 id=rangetracker>RangeTracker</h4><p>A
<code>RangeTracker</code> is a thread-safe object used to manage the current
range and current position of the reader of a <code>BoundedSource</code> and
protect concurrent access to them.</p><p>To implement a
<code>RangeTracker</code>, you should first familiarize yourself with the
following definitions:</p><ul><li><p><strong>Position-based sources</strong> -
A position-based source can be described by a rang [...]
+such as connections and buffers. <code>start</code> is called exactly once,
then <code>advance</code>
+is called repeatedly; whenever either returns <code>True</code>, the current
record is
+available through <code>get_current</code> and
<code>get_current_timestamp</code>.</p><p>To implement an
<code>UnboundedReader</code>, your subclass must override the following
+methods:</p><ul><li><p><code>start</code>: Initializes the reader, positions
it at the first record, and returns
+whether one is available. This is a good place for expensive
initialization.</p></li><li><p><code>advance</code>: Advances to the next
record and returns whether one is available. A
+<code>False</code> return means that no data is available right now, which
differs from the
+end of the stream: a reader signals a permanent end by reporting a watermark of
+<code>MAX_TIMESTAMP</code>. This method should not block; return
<code>False</code> when no data is
+currently available instead of waiting for
more.</p></li><li><p><code>get_current</code>: Returns the record at the
current position, last read by <code>start</code>
+or <code>advance</code>.</p></li><li><p><code>get_current_timestamp</code>:
Returns the event-time timestamp of the current
+record, which becomes the timestamp of the output
element.</p></li><li><p><code>get_watermark</code>: Returns a watermark, the
approximate lower bound on the
+timestamps of the records that this reader produces in the future. The runner
+uses the watermark as an estimate of data completeness in windowing and
triggers.
+The watermark is treated as
monotonic.</p></li><li><p><code>get_checkpoint_mark</code>: Returns a
<code>CheckpointMark</code> that records how far the
+reader has read. It is called only at a bundle boundary, and the mark is passed
+back to <code>create_reader</code> to
resume.</p></li><li><p><code>close</code>: Releases the reader’s
resources. The default is a no-op.</p></li></ul><h4
id=checkpoint-marks>Checkpoint marks</h4><p>A <code>CheckpointMark</code> is a
durable, serializable position in the stream. The runner
+persists it with the coder from <code>get_checkpoint_mark_coder</code> and
hands it to
+<code>create_reader</code> when a bundle resumes or when a worker recovers
from a failure.</p><p>Override <code>finalize_checkpoint</code> to acknowledge
or commit the consumed records
+upstream, for example to ack the messages on a queue. It is called after the
+runner has durably committed the work covered by that mark. Finalization is
best
+effort: a mark may never be finalized, and a retried bundle may re-cut a mark
+over an overlapping span, so acknowledge by absolute position and keep the
method
+idempotent.</p><h3 id=convenience-source-base-classes>Convenience Source base
classes</h3><p>The Beam SDK for Python contains some convenient abstract base
classes to help you easily create new sources.</p><h4
id=filebasedsource>FileBasedSource</h4><p><code>FileBasedSource</code> is a
framework for developing sources for new file types. You can derive your
<code>BoundedSource</code> class from the <a
href=https://github.com/apache/beam/blob/master/sdks/python/apache_beam/io/filebasedsour
[...]
def __init__(self, count):
self.records_read = Metrics.counter(self.__class__, 'recordsRead')
self._count = count
@@ -112,6 +169,92 @@ a wrapper.</li></ul><p>You can find these classes in the
recommended that you do not expose the code for the source itself as
demonstrated in the example above. Use a wrapping <code>PTransform</code>
instead.
<a href=#ptransform-wrappers>PTransform wrappers</a> discusses why you should
avoid
+exposing your sources, and walks through how to create a wrapper.</p><h3
id=reading-from-an-unboundedsource>Reading from an UnboundedSource</h3><p>The
following example, <code>QueueSource</code>, reads from a partitioned message
queue.
+<code>my_queue</code> stands in for the client library of the system you read
from.</p><div class=snippet><div class="notebook-skip code-snippet
without_switcher"><a class=copy type=button data-bs-toggle=tooltip
data-bs-placement=bottom title="Copy to clipboard"><img
src=/images/copy-icon.svg></a><pre tabindex=0><code>import apache_beam as beam
+from apache_beam.io.unbounded_source import CheckpointMark
+from apache_beam.io.unbounded_source import UnboundedReader
+from apache_beam.io.unbounded_source import UnboundedSource
+from apache_beam.utils.timestamp import Timestamp
+
+
+class QueueCheckpointMark(CheckpointMark):
+ def __init__(self, offset):
+ self.offset = offset
+
+ def finalize_checkpoint(self):
+ # Acknowledging an absolute offset is safe to repeat.
+ my_queue.ack_through(self.offset)
+
+
+class QueueReader(UnboundedReader):
+ def __init__(self, partition, offset):
+ self._partition = partition
+ self._offset = offset
+ self._message = None
+
+ def start(self):
+ return self.advance()
+
+ def advance(self):
+ message = my_queue.poll(self._partition, self._offset)
+ if message is None:
+ return False
+ self._offset = message.offset
+ self._message = message
+ return True
+
+ def get_current(self):
+ return self._message.body
+
+ def get_current_timestamp(self):
+ return Timestamp(micros=self._message.event_time_micros)
+
+ def get_watermark(self):
+ return Timestamp(micros=my_queue.oldest_pending_micros(self._partition))
+
+ def get_checkpoint_mark(self):
+ return QueueCheckpointMark(self._offset)
+
+ def close(self):
+ my_queue.disconnect(self._partition)
+
+
+class QueueSource(UnboundedSource):
+ def __init__(self, topic, partition=None):
+ self._topic = topic
+ self._partition = partition
+
+ def split(self, desired_num_splits, options=None):
+ if self._partition is not None:
+ return [self]
+ return [
+ QueueSource(self._topic, partition)
+ for partition in my_queue.partitions(self._topic)
+ ]
+
+ def create_reader(self, options, checkpoint_mark):
+ offset = None if checkpoint_mark is None else checkpoint_mark.offset
+ return QueueReader(self._partition, offset)
+
+ def get_checkpoint_mark_coder(self):
+ return beam.coders.PickleCoder()
+
+ def default_output_coder(self):
+ return beam.coders.BytesCoder()</code></pre></div></div><p>To read data
from the source in your pipeline, use the <code>Read</code> transform, which
+dispatches an <code>UnboundedSource</code> automatically:</p><div
class=snippet><div class="notebook-skip code-snippet without_switcher"><a
class=copy type=button data-bs-toggle=tooltip data-bs-placement=bottom
title="Copy to clipboard"><img src=/images/copy-icon.svg></a><pre
tabindex=0><code>with beam.Pipeline(options=pipeline_options) as p:
+ orders = p |
beam.io.Read(QueueSource('orders'))</code></pre></div></div><p>A bundle
ends when the reader runs out of data, and also once the reader has
+emitted <code>max_records_per_bundle</code> records or spent
<code>max_read_time_seconds</code> in the
+bundle, so the runner commits the checkpoint and runs finalization before the
+read resumes. An idle reader is polled again after <code>poll_interval</code>
seconds.
+Apply <code>ReadFromUnboundedSource</code> directly to change these
defaults:</p><div class=snippet><div class="notebook-skip code-snippet
without_switcher"><a class=copy type=button data-bs-toggle=tooltip
data-bs-placement=bottom title="Copy to clipboard"><img
src=/images/copy-icon.svg></a><pre tabindex=0><code>from
apache_beam.io.unbounded_source import ReadFromUnboundedSource
+
+orders = p | ReadFromUnboundedSource(
+ QueueSource('orders'),
+ poll_interval=5,
+ max_records_per_bundle=1000,
+
max_read_time_seconds=30)</code></pre></div></div><p><strong>Note:</strong> As
with a bounded source, we recommend that you do not expose the code
+for the source itself to end-users. Use a wrapping <code>PTransform</code>
instead.
+<a href=#ptransform-wrappers>PTransform wrappers</a> discusses why you should
avoid
exposing your sources, and walks through how to create a wrapper.</p><h2
id=using-the-filebasedsink-abstraction>Using the FileBasedSink
abstraction</h2><p>If your data source uses files, you can implement the <a
href=https://beam.apache.org/releases/pydoc/2.75.0/apache_beam.io.filebasedsink.html>FileBasedSink</a>
abstraction to create a file-based sink. For other sinks, use
<code>ParDo</code>,
<code>GroupByKey</code>, and other transforms offered by the Beam SDK for
Python. See the
diff --git a/website/generated-content/sitemap.xml
b/website/generated-content/sitemap.xml
index c7f973400f6..c15125ae7e1 100644
--- a/website/generated-content/sitemap.xml
+++ b/website/generated-content/sitemap.xml
@@ -1 +1 @@
-<?xml version="1.0" encoding="utf-8" standalone="yes"?><urlset
xmlns="http://www.sitemaps.org/schemas/sitemap/0.9";
xmlns:xhtml="http://www.w3.org/1999/xhtml";><url><loc>/blog/beam-summit-2026-interview-with-raj-katakam/</loc><lastmod>2026-08-21T12:41:01-04:00</lastmod></url><url><loc>/categories/blog/</loc><lastmod>2026-08-21T12:41:01-04:00</lastmod></url><url><loc>/blog/</loc><lastmod>2026-08-21T12:41:01-04:00</lastmod></url><url><loc>/categories/</loc><lastmod>2026-08-21T12:41:01-04:00<
[...]
\ No newline at end of file
+<?xml version="1.0" encoding="utf-8" standalone="yes"?><urlset
xmlns="http://www.sitemaps.org/schemas/sitemap/0.9";
xmlns:xhtml="http://www.w3.org/1999/xhtml";><url><loc>/blog/beam-summit-2026-interview-with-raj-katakam/</loc><lastmod>2026-08-21T16:59:47-06:00</lastmod></url><url><loc>/categories/blog/</loc><lastmod>2026-08-21T16:59:47-06:00</lastmod></url><url><loc>/blog/</loc><lastmod>2026-08-21T16:59:47-06:00</lastmod></url><url><loc>/categories/</loc><lastmod>2026-08-21T16:59:47-06:00<
[...]
\ No newline at end of file
