KafkaStream

atoti_kafka.KafkaStream(
    bootstrap_server: str,
    topic: str,
    group_id: str,
    *,
    batch_duration: timedelta = datetime.timedelta(seconds=1),
    consumer_config: Mapping[str, str] = frozendict({}),
)

Consume a Kafka topic and stream its records in the table. The records’ key deserializer default to StringDeserializer. The records’ message must be a JSON object with columns’ name as keys.

​

Attributes

​

bootstrap_server

host[:port] that the consumer should contact to bootstrap initial cluster metadata.

​

topic

Topic to subscribe to.

​

group_id

The name of the consumer group to join.

​

batch_duration

Time spent batching received events before publishing them to the table in a single transaction.

​

consumer_config

Mapping containing optional parameters to set up the KafkaConsumer. The list of available params can be found here.