> ## Documentation Index
> Fetch the complete documentation index at: https://docs.activeviam.com/llms.txt
> Use this file to discover all available pages before exploring further.

# atoti_kafka.KafkaStream

### *final class* atoti\_kafka.KafkaStream

Consume a Kafka topic and stream its records in the table.

The records’ key deserializer default to [StringDeserializer](https://kafka.apache.org/21/javadoc/org/apache/kafka/common/serialization/StringDeserializer.html).

The records’ message must be a JSON object with columns’ name as keys.

<Callout icon="link">
  **See also**:
  The other [`DataStream`](./atoti.data_stream.data_stream#atoti.data_stream.DataStream) implementations.
</Callout>

#### batch\_duration *: [timedelta](https://docs.python.org/3/library/datetime.html#datetime.timedelta)* *= datetime.timedelta(seconds=1)*

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

#### bootstrap\_server *: [str](https://docs.python.org/3/library/stdtypes.html#str)*

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

#### consumer\_config *: [Mapping](https://docs.python.org/3/library/collections.abc.html#collections.abc.Mapping)\[[str](https://docs.python.org/3/library/stdtypes.html#str), [str](https://docs.python.org/3/library/stdtypes.html#str)]* *= frozendict(\{})*

Mapping containing optional parameters to set up the KafkaConsumer.

The list of available params can be found [here](https://kafka.apache.org/10/javadoc/index.html?org/apache/kafka/clients/consumer/ConsumerConfig.html).

#### group\_id *: [str](https://docs.python.org/3/library/stdtypes.html#str)*

The name of the consumer group to join.

#### topic *: [str](https://docs.python.org/3/library/stdtypes.html#str)*

Topic to subscribe to.