apache_beam.io.watch module

Experimental Watch transform for the Python SDK.

Watch continuously watches a growing set of outputs for each input element, calling a user poll function on an interval until a per-input termination condition fires. It is the engine behind periodic file-discovery and any periodic polling source.

For every input element the transform runs an independent loop:

poll -> keep never-seen-before outputs -> emit them (timestamped) ->
update watermark -> check termination -> wait(poll_interval) -> poll -> ...

The output is an unbounded PCollection of (input, output) pairs. Each output carries the event time the poll function first reported it. Dedup hashes each output’s key: the output itself by default, or output_key_fn(output) when one is given. The key coder is inferred when not passed explicitly and converted to its deterministic form, so equal keys hash equally across workers and restarts.

By default, the Watch transform internally stores the hash of all items seen. If the incremental items returned by the poll function guarantee monotonic timestamp growth (new items on the next poll have timestamps larger than the largest of the previous poll), consider setting timestamp_cursor=True for better performance, as it replaces the hash dedup with an O(1) event-time cursor; see Watch.

Example:

from apache_beam.io.watch import Watch, PollResult, after_total_of
from apache_beam.transforms.window import TimestampedValue
from apache_beam.utils.timestamp import Duration, Timestamp

def poll(prefix) -> PollResult[str]:
  now = Timestamp.now()
  outputs = [TimestampedValue(prefix + str(i), now) for i in range(3)]
  return PollResult.complete(outputs)

watched = inputs | Watch(
    poll,
    poll_interval=Duration(seconds=5),
    termination=after_total_of(60))

This API is experimental and may change in backwards-incompatible ways.

class apache_beam.io.watch.Watch(poll_fn: Callable[[Any], PollResult], poll_interval, termination: TerminationCondition | None = None, output_coder: Coder | None = None, output_key_fn: Callable[[Any], Any] | None = None, output_key_coder: Coder | None = None, timestamp_cursor: bool = False, now_fn: Callable[[], float] | None = None)

Bases: PTransform

Watches a growing set of outputs per input via a periodic poll function.

The output is an unbounded PCollection of (input, output) pairs.

Parameters:

• poll_fn – callable input -> PollResult, invoked once per poll round.

• poll_interval – delay between two poll rounds for one input, as a Duration or in seconds.

• termination – per-input stop policy; defaults to never().

• output_coder – coder for the poll outputs, used to keep them in the restriction state. Inferred when omitted: from a PollFn’s default_output_coder(), else from the registered coder for the V of a PollResult[V] return annotation on poll_fn.

• output_key_fn – derives the dedup key from an output; an output is emitted only when its key was never seen before. Defaults to the output itself.

• output_key_coder – coder whose encoding of the key is hashed for dedup; inferred like output_coder when omitted. It is converted with as_deterministic_coder so equal keys always hash equally; a coder with no deterministic form is rejected.

• timestamp_cursor – dedup by event time instead of by key. Each round emits only outputs strictly past the greatest event time already emitted, so the per-input state is a single timestamp. Requires every new output to carry an event time strictly greater than all previously emitted ones; re-listed old outputs at or below the cursor are dropped as already seen. For sources whose new outputs can arrive at or below the cursor, keep the default hash dedup. Incompatible with output_key_fn and output_key_coder.

• now_fn – clock used for termination decisions; tests can inject one.

expand(pcoll)

class apache_beam.io.watch.PollResult(outputs: tuple[TimestampedValue, ...], watermark: Timestamp | None = None)

Bases: Generic[OutputT]

Outputs produced by one poll, plus an optional explicit watermark.

watermark of None lets the transform infer the watermark from the earliest new output. A watermark of MAX_TIMESTAMP (set by complete()) marks the input finished, so polling stops.

The OutputT type parameter can annotate a poll function’s return type, as in -> PollResult[str]; the transform infers the output coder from it.

outputs: tuple[TimestampedValue, ...]

watermark: Timestamp | None = None

property is_complete: bool

static incomplete(outputs: Iterable, timestamp=None) → PollResult

Reports outputs and expects more; the transform infers the watermark.

A raw (non-TimestampedValue) output is stamped with timestamp when given, else with the current processing time. The inferred watermark is safe only for non-decreasing event-time enumerations; out-of-order sources should call with_watermark().

static complete(outputs: Iterable, timestamp=None) → PollResult

Reports the final outputs for an input, after which polling stops.

A raw (non-TimestampedValue) output is stamped with timestamp when given, else with the current processing time. The watermark is released to MAX_TIMESTAMP so downstream event-time windows close.

with_watermark(watermark) → PollResult

Sets an explicit watermark, a promise that no future output for this input will have an event time below watermark.

class apache_beam.io.watch.PollFn

Bases: object

Optional base for a poll function input -> PollResult.

Any callable with that signature works; subclass only to attach an output coder hint via default_output_coder():

from apache_beam import coders

class ListFiles(PollFn):
  def __call__(self, prefix):
    return PollResult.incomplete(list_files(prefix))

  def default_output_coder(self):
    return coders.StrUtf8Coder()

A plain function can instead annotate its return type as PollResult[V] and have the output coder inferred from V.

default_output_coder() → Coder | None

class apache_beam.io.watch.TerminationCondition

Bases: object

Per-input stop policy with immutable, encodable state.

Hooks follow the lifecycle of one input’s polling loop. state flows from for_new_input() through the per-round hooks and is serialized with state_coder().

for_new_input(now: Timestamp, element: Any) → Any

on_seen_new_output(now: Timestamp, state: Any) → Any

on_poll_complete(state: Any) → Any

can_stop_polling(now: Timestamp, state: Any) → bool

state_coder() → Coder

apache_beam.io.watch.never() → TerminationCondition

Polls until PollResult.complete().

apache_beam.io.watch.after_total_of(duration) → TerminationCondition

Stops polling an input after duration (a Duration or seconds) has elapsed since it was first seen.