airflow.providers.amazon.aws.triggers.kinesis

Classes

KinesisTrigger

Wait asynchronously for records on an Amazon Kinesis Data Stream.

Module Contents

class airflow.providers.amazon.aws.triggers.kinesis.KinesisTrigger(stream_name, aws_conn_id='aws_default', shard_iterator_type='LATEST', batch_size=100, waiter_delay=10, region_name=None, verify=None, botocore_config=None)[source]

Bases: airflow.triggers.base.BaseEventTrigger

Wait asynchronously for records on an Amazon Kinesis Data Stream.

The trigger is long-running and emits one event for each non-empty shard response. Record data is base64-encoded in the event payload and must be decoded by the consumer. Delivery is best-effort: a triggerer failure can cause records to be repeated or missed around the failure window.

When Airflow provides an asset state store for a single watched asset, the trigger checkpoints the last sequence number read from each shard. The same asset and stream identity share one logical cursor; do not configure multiple watchers that require independent progress for the same stream on one asset.

Parameters:

  • stream_name (str) – Name of the Kinesis Data Stream to watch.

  • aws_conn_id (str | None) – AWS connection id.

  • shard_iterator_type (str) – Position used when a shard has no checkpoint. LATEST only sees records that arrive after the watcher starts; TRIM_HORIZON starts from the oldest retained record. Only these types are supported because AT_TIMESTAMP requires a timestamp and sequence-number types require a starting sequence number for each shard. Checkpoints handle resuming each shard.

  • batch_size (int) – Maximum records per GetRecords call and trigger event. Must be between 1 and 10,000. Record data is base64-encoded before it is stored in the metadata database, so use a conservative value for large records.

  • waiter_delay (int) – Seconds between complete polling sweeps. Must be less than the five-minute shard iterator lifetime. Kinesis permits at most five GetRecords calls per second per shard. When reading a backlog with TRIM_HORIZON, draining N records from one shard takes roughly ceil(N / batch_size) * waiter_delay seconds when calls return full batches; with the defaults, 10,000 records take about 1,000 seconds.

  • region_name (str | None) – AWS region for the Kinesis client.

  • verify (bool | str | None) – Whether to verify SSL certificates, or the path to a CA bundle.

  • botocore_config (dict | None) – Botocore configuration passed to the Kinesis client.

stream_name[source]
aws_conn_id = 'aws_default'[source]
shard_iterator_type = 'LATEST'[source]
batch_size = 100[source]
waiter_delay = 10[source]
region_name = None[source]
verify = None[source]
botocore_config = None[source]
serialize()[source]

Return the information needed to reconstruct this Trigger.

Returns:

Tuple of (class path, keyword arguments needed to re-instantiate).

Return type:

tuple[str, dict[str, Any]]

property hook: airflow.providers.amazon.aws.hooks.kinesis.KinesisHook[source]
async run()[source]

Run the trigger in an asynchronous context.

The trigger should yield an Event whenever it wants to fire off an event, and return None if it is finished. Single-event triggers should thus yield and then immediately return.

If it yields, it is likely that it will be resumed very quickly, but it may not be (e.g. if the workload is being moved to another triggerer process, or a multi-event trigger was being used for a single-event task defer).

In either case, Trigger classes should assume they will be persisted, and then rely on cleanup() being called when they are no longer needed.

Was this entry helpful?