KafkaTrigger

classView source ↗

class KafkaTrigger(
    conn: KafkaConn,
    topic: str,
    data_format: KafkaMessageFormatSpec,
    schema: str,
    group_id: str | None = None,
    time_rollover_secs: PositiveInt = 60,
    size_rollover_mb: PositiveInt | None = None,
    messages_rollover: PositiveInt | None = None,
    dlq_topic: str | None = None,
    start: datetime | None = None,
    end: datetime | None = None,
)

Bases: StageTrigger

Categories: trigger

A StageTrigger that consumes messages from a Kafka topic and stages them.

This trigger connects to a Kafka cluster, consumes messages from a specified topic, and uses rollover conditions (time, size, and message count) to buffer and stage the data.

Initialize a KafkaTrigger instance.

Parameters

parameter

conn

The Kafka consumer connection configuration.

parameter

topic

The Kafka topic to consume from.

parameter

data_format

The data format of the messages ("avro", "json", "protobuf").

parameter

schema

The schema string for deserializing messages.

parameter

group_id

Optional Kafka consumer group ID.

parameter

schema

Optional schema name for AWS Glue Schema Registry.

parameter

time_rollover_secs

The timeout in seconds before staging and committing a message buffer. Defaults to 60.

parameter

size_rollover_mb

Optional size-based rollover configuration. Size is specified in megabytes.

parameter

messages_rollover

Optional message count-based rollover configuration.

parameter

dlq_topic

Optional dead-letter queue topic.

parameter

start

Optional datetime for the trigger to start working.

parameter

end

Optional datetime for the trigger to stop working.

Properties

property

connKafkaConn

Kafka consumer connection configuration.

property

data_formatKafkaMessageFormatSpec

Data format of the messages ("avro", "json", "protobuf").

property

dlq_topicstr | None

Dead-letter queue topic.

property

enddatetime | None

property

group_idstr | None

Kafka consumer group ID.

property

messages_rolloverPositiveInt | None

Message count-based rollover configuration.

property

schemastr

Schema string for deserializing messages.

property

size_rollover_mbPositiveInt | None

Size-based rollover configuration. Size is specified in megabytes.

property

startdatetime | None

property

time_rollover_secsint

Timeout in seconds for consuming each message.

property

topicstr

Kafka topic to consume from.

Methods

method

View source ↗

run

def run(context: StageTriggerContext)

Run the Kafka stage trigger to consume messages, buffer them, and stage them based on rollover conditions.

Parameters:

parameter

context

The StageTriggerContext instance.