apache_beam.io.unbounded_source module

Experimental UnboundedSource support for the Python SDK.

UnboundedSource support is currently experimental in the Python SDK; the API may change in backwards-incompatible ways.

An unbounded source reads an effectively infinite stream of records (message queues, change-data-capture feeds, and similar) with checkpoint-based pause/resume, watermark reporting, and bundle finalization.

To define a source, implement UnboundedSource, an UnboundedReader, and (when the reader has a resumable position) a CheckpointMark:

import apache_beam as beam
from apache_beam.io.unbounded_source import (
    CheckpointMark, UnboundedReader, UnboundedSource)
from apache_beam.utils.timestamp import MAX_TIMESTAMP

class MyCheckpointMark(CheckpointMark):
  def __init__(self, position):
    self.position = position

  def finalize_checkpoint(self):
    # Commit/acknowledge records up to ``position`` upstream, e.g. ack the
    # consumed messages on a queue.
    ...

class MyReader(UnboundedReader):
  def start(self):
    # Position at the first record; return whether one is available.
    ...

  def advance(self):
    # Move to the next record; ``False`` means no data is available now.
    ...

  def get_current(self):
    ...

  def get_current_timestamp(self):
    ...  # event time of the current record

  def get_watermark(self):
    # Lower bound on the timestamps of future records. Return
    # ``MAX_TIMESTAMP`` to signal the reader has permanently finished.
    ...

  def get_checkpoint_mark(self):
    return MyCheckpointMark(...)

class MySource(UnboundedSource):
  def split(self, desired_num_splits, options=None):
    # Return independent sub-sources, or ``[self]`` when not splittable.
    return [self]

  def create_reader(self, options, checkpoint_mark):
    # Build a reader; resume after ``checkpoint_mark`` when it is not None.
    return MyReader(...)

  def get_checkpoint_mark_coder(self):
    return ...  # a Coder for MyCheckpointMark

Read the source in a pipeline with apache_beam.io.Read:

with beam.Pipeline() as p:
  p | beam.io.Read(MySource()) | beam.Map(print)

class apache_beam.io.unbounded_source.CheckpointMark

Bases: object

A durable, serializable position in an UnboundedSource.

Produced by UnboundedReader.get_checkpoint_mark(), encoded with UnboundedSource.get_checkpoint_mark_coder(), and used to resume a reader (see UnboundedSource.create_reader()).

finalize_checkpoint() → None

Called once the runner has durably committed work up to this mark.

Override to acknowledge/commit upstream (for example, ack the consumed messages on a queue). The default is a no-op.

The runner calls this at most once for a committed checkpoint mark. Finalization is best effort; a mark may never be finalized. An exception raised here is logged. On bundle retry an uncommitted mark may be re-cut over an overlapping span, so this method must be idempotent (acknowledge by absolute position).

class apache_beam.io.unbounded_source.UnboundedReader

Bases: object

Reads records from an UnboundedSource.

Lifecycle: exactly one start(), then any number of advance() calls; whenever one returns True the current record is available via get_current() / get_current_timestamp(). A False return means “no data available right now”, which is distinct from end-of-stream: a reader signals a permanent end by reporting a watermark of MAX_TIMESTAMP.

start() → bool

Positions at the first record; returns whether one is available.

advance() → bool

Advances to the next record. False means no data is available now.

Should not block. The wrapper enforces the per-bundle record and time caps only between records, so a blocking start/advance can overrun the time cap and stall the bundle. Return False when no data is currently available instead of waiting.

get_current() → Any

Returns the record claimed by the last successful start/advance.

get_current_timestamp() → Timestamp

Returns the event-time timestamp of the current record.

get_watermark() → Timestamp

An approximate lower bound on timestamps of future records.

Treated as monotonic by the wrapper. Return MAX_TIMESTAMP to signal that this reader has permanently finished.

get_checkpoint_mark() → CheckpointMark

Returns a durable mark to resume from. Call only at a bundle boundary.

close() → None

Releases reader resources. Default no-op.

class apache_beam.io.unbounded_source.UnboundedSource

Bases: SourceBase

A source producing an unbounded stream of records with checkpointing.

Read it in a pipeline with apache_beam.io.Read:

p | beam.io.Read(MyUnboundedSource())

split(desired_num_splits: int, options: Any | None = None) → Iterable[UnboundedSource]

Splits into at most desired_num_splits independent sub-sources.

Each returned sub-source must be independent and must not share mutable state with siblings (the runner may execute them concurrently across workers). Return [self] if the source cannot be split. Splitting is performed once, before any checkpoint exists; once a reader has checkpointed, the restriction is kept intact.

create_reader(options: Any | None, checkpoint_mark: CheckpointMark | None) → UnboundedReader

Creates a reader, optionally resuming from checkpoint_mark.

Contract:

• When checkpoint_mark is None, the returned reader’s start() produces the very first record of the source (or returns False if none yet).

• When checkpoint_mark is not None, the returned reader’s start() produces the first record strictly after the position encoded by checkpoint_mark. The reader must not re-deliver records already covered by the prior bundle.

get_checkpoint_mark_coder() → Coder

Returns the coder for this source’s CheckpointMark instances.

The SDK may call this while encoding or decoding source restrictions. Implementations should be deterministic, side-effect free, and should not perform I/O.

is_bounded() → bool

default_output_coder() → Coder

class apache_beam.io.unbounded_source.ReadFromUnboundedSource(source: UnboundedSource, poll_interval: float = 1.0, max_records_per_bundle: int = 10000, max_read_time_seconds: float = 10.0)

Bases: PTransform

Reads an UnboundedSource.

Most users should prefer apache_beam.io.Read, which dispatches an UnboundedSource here automatically:

p | beam.io.Read(MyUnboundedSource())

Parameters:

• source – the UnboundedSource to read.

• poll_interval – resume delay in seconds applied when the reader has no data, which bounds how often an idle source is polled. Must be >= 0.

• max_records_per_bundle – a busy reader self-checkpoints after emitting this many records in one bundle. Must be >= 1. Defaults to 10000.

• max_read_time_seconds – a busy reader self-checkpoints after this many seconds in one bundle. Must be > 0. Defaults to 10.0. The deadline is checked between records, so a reader that blocks inside advance() may overrun it; max_records_per_bundle is the hard backstop.

The bundle self-checkpoints as soon as either cap is reached.

expand(pbegin)