> ## 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.

# KafkaStream

<span id="atoti_kafka.KafkaStream" />

> atoti\_kafka.KafkaStream(<br />
>     *bootstrap\_server*: [str](https://docs.python.org/3/library/stdtypes.html#str),<br />
>     *topic*: [str](https://docs.python.org/3/library/stdtypes.html#str),<br />
>     *group\_id*: [str](https://docs.python.org/3/library/stdtypes.html#str),<br />
>     \*,<br />
>     *batch\_duration*: [timedelta](https://docs.python.org/3/library/datetime.html#datetime.timedelta) = `datetime.timedelta(seconds=1)`,<br />
>     *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({})`,<br />
> )

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.DataStream#atoti.data_stream.DataStream) implementations.
</Callout>

### Attributes

<h4 id="atoti_kafka.KafkaStream.bootstrap_server">
  *bootstrap\_server*
</h4>

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

<h4 id="atoti_kafka.KafkaStream.topic">
  *topic*
</h4>

Topic to subscribe to.

<h4 id="atoti_kafka.KafkaStream.group_id">
  *group\_id*
</h4>

The name of the consumer group to join.

<h4 id="atoti_kafka.KafkaStream.batch_duration">
  *batch\_duration*
</h4>

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

<h4 id="atoti_kafka.KafkaStream.consumer_config">
  *consumer\_config*
</h4>

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).