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 items returned by the poll function arrive in roughly non-decreasing event time, consider setting timestamp_cursor=True for better performance; 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, allowed_lateness=0, 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 – bound the dedup state by event time, for better performance. An output more than allowed_lateness behind the greatest event time emitted so far is taken as already seen and dropped, so this suits sources whose outputs arrive in roughly non-decreasing event time; keep the default for sources that can hand out much older outputs at any time.

• allowed_lateness – how far behind the greatest emitted event time an output is still deduplicated by key, as a Duration or in seconds. Widen it for a source whose outputs arrive out of order, at the cost of a larger state. Ignored unless timestamp_cursor is set; defaults to zero.

• 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.