KafkaStreamSrc

class

class KafkaStreamSrc(
    data_format: KafkaMessageFormatSpec,
    schema: str,
    topic: KafkaTopicSpec,
    dlq_topic: KafkaTopicSpec | None = None,
    src_cfg: SrcCfgSpec | None = None,
)

Bases: StreamSrc

Categories: source

Kafka streaming source for @stream_publisher -- one batch per rollover.

Consumes a single Kafka topic and emits one tableframe batch each time a rollover threshold fires (time / size / message count). Used as the source of a @stream_publisher; that function receives the batch as a list of frames and must NOT declare a ctx parameter. The emitted table name is the topic.

Every batch this source stages automatically carries a @td.kafka.topic column holding the topic the message was consumed from.

Examples

Consume Avro payloads, staging a batch every 500 messages via a src_cfg rollover threshold:

@stream_publisher(
    source=KafkaStreamSrc(
        topic="users",
        data_format="avro",
        schema=AVRO_SCHEMA,
        src_cfg={"kafka.messages_rollover": 500},
    ),
    flush_interval_mins=1,
    output_tables=["users"],
)
def ingest(users: TableFramesSpec) -> TableFrameSpec:
    return concat(users)

Consume Protobuf payloads; the Protobuf marker carries the message type name:

@stream_publisher(
    source=KafkaStreamSrc(
        topic="persons",
        data_format=Protobuf(message_name="PersonV1"),
        schema=PROTO_SCHEMA,
    ),
    flush_interval_mins=1,
    output_tables=["persons"],
)
def ingest(persons: TableFramesSpec) -> TableFrameSpec:
    return concat(persons)

Parameters

parameter

data_formatKafkaMessageFormatSpec (Literal['avro', 'json'] | Protobuf)

Payload wire format -- "avro", "json" or a Protobuf(message_name=...) marker.

parameter

schemastr

Schema definition string for data_format (an Avro schema, a JSON Schema, or a Protobuf .proto). Required; in schema-registry mode it also cross-checks the registry's resolved schema.

parameter

topicKafkaTopicSpec (str)

Kafka topic to consume; also the emitted table name.

parameter

dlq_topicKafkaTopicSpec | None (str | None)

Optional dead-letter topic that deserialization failures are produced to.

parameter

src_cfgSrcCfgSpec | None (Mapping[Literal['kafka.consume_timeout_seconds', 'kafka.connection_timeout_seconds', 'kafka.logging_level', 'kafka.time_rollover_seconds', 'kafka.size_rollover_mb', 'kafka.messages_rollover'], Any] | None)

Optional connector config over the keys kafka.consume_timeout_seconds, kafka.connection_timeout_seconds, kafka.logging_level, kafka.time_rollover_seconds, kafka.size_rollover_mb and kafka.messages_rollover.

Methods

method

output_table

def output_table() -> str

Return the emitted table name -- the source's topic.