Apache Kafka¶

Apache Kafka is a distributed event streaming platform. This sheet covers producers, consumers, topics, partitions, consumer groups, exactly-once semantics, and Spark Structured Streaming integration.

Table of Contents¶

Installation / Setup
Core Syntax
Common Operations
Practical Examples
Common Patterns
Best Practices
Common Mistakes
Troubleshooting
Related Cheat Sheets
Official Documentation

Installation / Setup¶

# Install Kafka Python client
pip install kafka-python-ng

# Or use confluent-kafka (recommended)
pip install confluent-kafka

# confluent-kafka (recommended)
from confluent_kafka import Producer, Consumer, KafkaError

# kafka-python
from kafka import KafkaProducer, KafkaConsumer

Core Syntax¶

Kafka Concepts¶

Concept	Description
Topic	Category/feed of messages
Partition	Topic is split into partitions for parallelism
Offset	Position of a message within a partition
Producer	Publishes messages to topics
Consumer	Reads messages from topics
Consumer Group	Group of consumers that divide partition reading
Broker	Kafka server node
Cluster	Collection of brokers

Producer (confluent-kafka)¶

from confluent_kafka import Producer
import json

# Create producer
producer = Producer({
    'bootstrap.servers': 'localhost:9092',
    'acks': 'all',  # wait for all replicas
    'retries': 3,
    'enable.idempotence': True  # exactly-once
})

# Callback for delivery reports
def delivery_callback(err, msg):
    if err:
        print(f"Delivery failed: {err}")
    else:
        print(f"Delivered to {msg.topic()} [{msg.partition()}]")

# Send message
producer.produce(
    topic='events',
    key='user123',
    value=json.dumps({"event": "click", "user": "user123"}).encode('utf-8'),
    callback=delivery_callback
)

# Flush to ensure delivery
producer.flush()

Consumer (confluent-kafka)¶

from confluent_kafka import Consumer, KafkaError

# Create consumer
consumer = Consumer({
    'bootstrap.servers': 'localhost:9092',
    'group.id': 'my-group',
    'auto.offset.reset': 'earliest',  # or 'latest'
    'enable.auto.commit': False,
})

# Subscribe to topic
consumer.subscribe(['events'])

# Poll for messages
while True:
    msg = consumer.poll(1.0)  # timeout in seconds

    if msg is None:
        continue
    if msg.error():
        if msg.error().code() == KafkaError._PARTITION_EOF:
            continue
        print(f"Error: {msg.error()}")
        break

    # Process message
    key = msg.key().decode('utf-8')
    value = json.loads(msg.value().decode('utf-8'))
    print(f"Key: {key}, Value: {value}")

    # Commit offset
    consumer.commit(msg)

consumer.close()

Kafka-python Producer¶

from kafka import KafkaProducer
import json

producer = KafkaProducer(
    bootstrap_servers='localhost:9092',
    value_serializer=lambda v: json.dumps(v).encode('utf-8'),
    key_serializer=lambda k: k.encode('utf-8') if k else None,
    acks='all',
    retries=3
)

producer.send('events', key='user123', value={"event": "click"})
producer.flush()

Kafka-python Consumer¶

from kafka import KafkaConsumer
import json

consumer = KafkaConsumer(
    'events',
    bootstrap_servers=['localhost:9092'],
    group_id='my-group',
    auto_offset_reset='earliest',
    value_deserializer=lambda m: json.loads(m.decode('utf-8')),
    enable_auto_commit=False
)

for message in consumer:
    print(f"Key: {message.key}, Value: {message.value}")
    consumer.commit()

Common Operations¶

Consumer Groups¶

# Consumer group divides partitions among consumers
# Each partition is assigned to exactly one consumer in a group
# More consumers than partitions = idle consumers

# Example: 3 partitions, 2 consumers in group
# Consumer 1: partitions 0, 1
# Consumer 2: partition 2

Topic Management¶

from confluent_kafka.admin import AdminClient, NewTopic

admin = AdminClient({'bootstrap.servers': 'localhost:9092'})

# Create topic
new_topic = NewTopic(
    topic='events',
    num_partitions=3,
    replication_factor=1
)
admin.create_topics([new_topic])

# List topics
topics = admin.list_topics()
print(topics.topics)

# Delete topic
admin.delete_topics(['events'])

Exactly-Once Semantics¶

# Producer: enable idempotence
producer = Producer({
    'bootstrap.servers': 'localhost:9092',
    'enable.idempotence': True,  # exactly-once producer
    'acks': 'all',
    'max.in.flight.requests.per.connection': 5
})

# Consumer: manual commit
consumer = Consumer({
    'bootstrap.servers': 'localhost:9092',
    'group.id': 'my-group',
    'enable.auto.commit': False,
    'isolation.level': 'read_committed'  # read only committed messages
})

Practical Examples¶

Example 1 — Event streaming pipeline¶

from confluent_kafka import Producer, Consumer
import json
from datetime import datetime

# Producer: send events
def send_event(event_type, data):
    producer = Producer({'bootstrap.servers': 'localhost:9092'})
    event = {
        "event_type": event_type,
        "timestamp": datetime.now().isoformat(),
        "data": data
    }
    producer.produce(
        topic='events',
        key=event_type,
        value=json.dumps(event).encode('utf-8')
    )
    producer.flush()

# Consumer: process events
def process_events():
    consumer = Consumer({
        'bootstrap.servers': 'localhost:9092',
        'group.id': 'analytics',
        'auto.offset.reset': 'earliest'
    })
    consumer.subscribe(['events'])

    while True:
        msg = consumer.poll(1.0)
        if msg and not msg.error():
            event = json.loads(msg.value().decode('utf-8'))
            print(f"Processing: {event}")

Example 2 — Spark Structured Streaming from Kafka¶

# Read stream from Kafka
stream_df = spark.readStream \
    .format("kafka") \
    .option("kafka.bootstrap.servers", "localhost:9092") \
    .option("subscribe", "events") \
    .option("startingOffsets", "earliest") \
    .load()

# Parse JSON messages
from pyspark.sql.functions import from_json, col
from pyspark.sql.types import StructType, StringType, TimestampType

schema = StructType() \
    .add("event_type", StringType()) \
    .add("timestamp", TimestampType()) \
    .add("data", StringType())

parsed_df = stream_df \
    .selectExpr("CAST(key AS STRING)", "CAST(value AS STRING)") \
    .withColumn("parsed", from_json(col("value"), schema)) \
    .select("parsed.*")

# Write to Delta
query = parsed_df.writeStream \
    .format("delta") \
    .outputMode("append") \
    .option("checkpointLocation", "dbfs:/checkpoints/events") \
    .toTable("events_table")

Example 3 — Write to Kafka from Spark¶

# Write DataFrame to Kafka
df.selectExpr(
    "CAST(key AS STRING)",
    "to_json(struct(*)) AS value"
).write \
    .format("kafka") \
    .option("kafka.bootstrap.servers", "localhost:9092") \
    .option("topic", "output_events") \
    .save()

Common Patterns¶

Pattern: dead letter queue¶

def process_with_dlq(consumer, processor):
    while True:
        msg = consumer.poll(1.0)
        if msg and not msg.error():
            try:
                process(msg)
                consumer.commit(msg)
            except Exception as e:
                # Send to DLQ
                producer.produce('events_dlq', msg.value())
                consumer.commit(msg)

Pattern: event-driven architecture¶

# Events trigger downstream actions
# Order created → inventory check → payment → shipping

# Each service is a consumer/producer
# Service A: consumes 'orders', produces 'inventory_checks'
# Service B: consumes 'inventory_checks', produces 'payments'
# Service C: consumes 'payments', produces 'shipments'

Pattern: exactly-once processing¶

# Use transactional producer + consumer offset tracking
producer = Producer({
    'bootstrap.servers': 'localhost:9092',
    'transactional.id': 'my-transactional-producer'
})

producer.init_transactions()

try:
    producer.begin_transaction()
    # produce messages
    producer.produce('topic', key, value)
    producer.commit_transaction()
except:
    producer.abort_transaction()

Best Practices¶

Use consumer groups — for scalability and fault tolerance.
Set appropriate replication factor — minimum 3 for production.
Monitor consumer lag — ensure consumers keep up with producers.
Use keys for ordering — messages with same key go to same partition.
Idempotent consumers — handle duplicate messages gracefully.
Use dead letter queues — capture failed messages for reprocessing.
Partition by key — ensure related messages are processed together.
Tune batch sizes — balance latency vs throughput.

Common Mistakes¶

Mistake	Fix
No consumer group	Use consumer groups for scalability
Too many partitions	Start with moderate count; increase as needed
Not monitoring lag	Set up lag monitoring and alerts
Idempotent consumers not handling duplicates	Design for at-least-once delivery
Hardcoding broker addresses	Use configuration management
Not using dead letter queues	Implement DLQ for failed messages

Troubleshooting¶

Symptom	Cause / Fix
High consumer lag	Add consumers; optimize processing
Messages not delivered	Check topic exists; check broker connectivity
Duplicate messages	Use idempotent consumers; enable exactly-once
Consumer not receiving messages	Check consumer group; check offset reset
Slow producers	Increase batch size; tune acks setting

apache-spark.md — Spark core concepts
pyspark.md — PySpark DataFrame operations
databricks.md — Databricks platform
delta-lake.md — Delta Lake ACID transactions
airflow — batch orchestration

Official Documentation¶

Kafka documentation — https://kafka.apache.org/documentation/
Confluent Kafka Python — https://docs.confluent.io/platform/current/clients/confluent-kafka-python.html
Spark Structured Streaming + Kafka — https://spark.apache.org/docs/latest/structured-streaming-kafka-integration.html

Last Updated¶

2026-09-06 — Initial version.

Apache Kafka¶

Table of Contents¶

Installation / Setup¶

Core Syntax¶

Kafka Concepts¶

Producer (confluent-kafka)¶

Consumer (confluent-kafka)¶

Kafka-python Producer¶

Kafka-python Consumer¶

Common Operations¶

Consumer Groups¶

Topic Management¶

Exactly-Once Semantics¶

Practical Examples¶

Example 1 — Event streaming pipeline¶

Example 2 — Spark Structured Streaming from Kafka¶

Example 3 — Write to Kafka from Spark¶

Common Patterns¶

Pattern: dead letter queue¶

Pattern: event-driven architecture¶

Pattern: exactly-once processing¶

Best Practices¶

Common Mistakes¶

Troubleshooting¶

Related Cheat Sheets¶

Official Documentation¶

Last Updated¶