---
sidebarDepth: 2
editLink: false
---
# Kafka Tasks
---
This module contains a collection of tasks to produce and consume Kafka events
 ## KafkaBatchConsume
 <div class='class-sig' id='prefect-tasks-kafka-kafka-kafkabatchconsume'><p class="prefect-sig">class </p><p class="prefect-class">prefect.tasks.kafka.kafka.KafkaBatchConsume</p>(bootstrap_servers=None, group_id=None, topics=None, request_timeout=1.0, auto_offset_reset=&quot;earliest&quot;, message_consume_limit=None, kafka_configs=None, **kwargs)<span class="source"><a href="https://github.com/PrefectHQ/prefect/blob/master/src/prefect/tasks/kafka/kafka.py#L8">[source]</a></span></div>

Task for consuming a batch of messages from Kafka topics.

**Args**:     <ul class="args"><li class="args">`bootstrap_servers (str, optional)`: comma separated host and port pairs that are the         addresses of kafka brokers.     </li><li class="args">`group_id (str, required)`: name of the consumer group the consumer will belong to.         Must be specified either at init or runtime.     </li><li class="args">`topics (List[str], required)`: list of topic names to consume messages from. Must         be specified either at init or runtime.     </li><li class="args">`request_timeout (float, optional)`: Maximum time to block waiting for message, event or callback     </li><li class="args">`auto_offset_reset (str, optional)`: configurable offset reset policy     </li><li class="args">`message_consume_limit (int, optional)`: max number of messages to consume before closing         the consumer     </li><li class="args">`kafka_configs (dict, optional)`: a dict of kafka client configuration properties used to         construct the consumer.     </li><li class="args">`**kwargs (Any, optional)`: additional keyword arguments to pass to the standard Task         init method</li></ul>

|methods: &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;|
|:----|
 | <div class='method-sig' id='prefect-tasks-kafka-kafka-kafkabatchconsume-run'><p class="prefect-class">prefect.tasks.kafka.kafka.KafkaBatchConsume.run</p>(bootstrap_servers=None, group_id=None, topics=None, request_timeout=1.0, auto_offset_reset=&quot;earliest&quot;, message_consume_limit=None, kafka_configs=None, **kwargs)<span class="source"><a href="https://github.com/PrefectHQ/prefect/blob/master/src/prefect/tasks/kafka/kafka.py#L50">[source]</a></span></div>
<p class="methods">Run method for this Task. Invoked by calling this Task after initialization within a Flow context, or by using `Task.bind`. **Args**:     <ul class="args"><li class="args">`bootstrap_servers (str, required)`: comma separated host and port pairs that are the         addresses of kafka brokers     </li><li class="args">`group_id (str, required)`: name of the consumer group the consumer will belong to     </li><li class="args">`topics (List[str], required)`: list of topic names to consume messages from     </li><li class="args">`request_timeout (float, optional)`: Maximum time to block waiting for message, event         or callback     </li><li class="args">`auto_offset_reset (str, optional)`: configurable offset reset policy     </li><li class="args">`message_consume_limit (int, optional)`: max number of messages to consume before         closing the consumer     </li><li class="args">`kafka_configs (dict, optional)`: a dict of kafka client configuration properties used         to construct the consumer.     </li><li class="args">`**kwargs (Any, optional)`: additional keyword arguments to pass to the standard Task         init method **Returns**:     - List of consumed messages</li></ul></p>|

---
<br>

 ## KafkaBatchProduce
 <div class='class-sig' id='prefect-tasks-kafka-kafka-kafkabatchproduce'><p class="prefect-sig">class </p><p class="prefect-class">prefect.tasks.kafka.kafka.KafkaBatchProduce</p>(bootstrap_servers=None, topic=None, messages=None, flush_threshold=None, callback=None, kafka_configs=None, **kwargs)<span class="source"><a href="https://github.com/PrefectHQ/prefect/blob/master/src/prefect/tasks/kafka/kafka.py#L141">[source]</a></span></div>

Task for producing a batch of messages to a Kafka topic.

**Args**:     <ul class="args"><li class="args">`bootstrap_servers (str, required)`: comma separated host and port pairs that are the         addresses of kafka brokers     </li><li class="args">`topic (str, required)`: name of topic to produce messages to. Must be specified         either at init or runtime.     </li><li class="args">`messages (List[dict], required)`: list of messages to produce into a topic where         a single message is a dictionary with a key and a value.     </li><li class="args">`flush_threshold (int, optional)`: threshold of messages produced before flushing     </li><li class="args">`callback (Callable, optional)`: callback assigned to a produce call     </li><li class="args">`kafka_configs (dict, optional)`: a dict of kafka client configuration properties used to         construct the producer.     </li><li class="args">`**kwargs (Any, optional)`: additional keyword arguments to pass to the standard Task         init method</li></ul>

|methods: &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;|
|:----|
 | <div class='method-sig' id='prefect-tasks-kafka-kafka-kafkabatchproduce-run'><p class="prefect-class">prefect.tasks.kafka.kafka.KafkaBatchProduce.run</p>(bootstrap_servers=None, topic=None, messages=None, flush_threshold=None, callback=None, kafka_configs=None)<span class="source"><a href="https://github.com/PrefectHQ/prefect/blob/master/src/prefect/tasks/kafka/kafka.py#L178">[source]</a></span></div>
<p class="methods">Run method for this Task. Invoked by calling this Task after initialization within a Flow context, or by using `Task.bind`. **Args**: <ul class="args"><li class="args">`bootstrap_servers (str, required)`: comma separated host and port pairs that are the     addresses of kafka brokers </li><li class="args">`topic (str, required)`: name of topics to produce messages to </li><li class="args">`messages (List[dict], required)`: list of messages to produce into topics where     a single message is a dictionary with a key and a value. </li><li class="args">`flush_threshold (int, optional)`: threshold of messages produced before flushing </li><li class="args">`callback (Callable, optional)`: callback assigned to a produce call </li><li class="args">`kafka_configs (dict, optional)`: a dict of kafka client configuration properties used to     construct the producer.</li></ul></p>|

---
<br>


<p class="auto-gen">This documentation was auto-generated from commit <a href='https://github.com/PrefectHQ/prefect/commit/n/a'>n/a</a> </br>on February 23, 2022 at 19:26 UTC</p>