Spark with Kafka: a Spark & Data Eng guide for developers

Introduction

Real-time data pipelines are no longer a luxury—they're a requirement for teams building fraud detection systems, recommendation engines, and live dashboards. Spark with Kafka: a Spark & Data Eng guide for developers bridges the gap between streaming message brokers and scalable distributed processing, letting us consume events from Kafka topics, transform them reliably, and sink results back to storage or downstream systems. This guide walks through the practical patterns we've used in production, from handling binary payloads to tracking consumer lag across clusters.

Read topics — Parse value bytes into structured columns

Kafka stores messages as byte arrays, so the first step in any Spark–Kafka pipeline is reading those bytes and parsing them into structured columns. Spark's structured streaming API makes this straightforward via the kafka format.

Here's a basic example that reads from a Kafka topic and extracts the key and value:

from pyspark.sql import SparkSession
from pyspark.sql.functions import col, from_json

spark = SparkSession.builder \
    .appName("KafkaReader") \
    .getOrCreate()

kafka_df = spark.readStream \
    .format("kafka") \
    .option("kafka.bootstrap.servers", "localhost:9092") \
    .option("subscribe", "events") \
    .option("startingOffsets", "latest") \
    .load()

# Kafka messages have key (bytes) and value (bytes) columns
raw = kafka_df.select(
    col("key").cast("string"),
    col("value").cast("string"),
    col("timestamp")
)

In this setup, we're subscribing to the events topic and letting Kafka know we want to start consuming from the latest offset. The key and value columns arrive as binary, so we cast them to strings for now. However, real-world events are rarely plain strings—they're JSON, Avro, or Protobuf. Casting and hoping won't scale.

For JSON payloads, we immediately hit the parsing challenge: the value column contains JSON text, but we need structured columns like user_id, event_type, and timestamp. We handle this by defining a schema and using from_json.

from pyspark.sql.types import StructType, StructField, StringType, LongType

event_schema = StructType([
    StructField("user_id", StringType()),
    StructField("event_type", StringType()),
    StructField("event_timestamp", LongType()),
    StructField("properties", StringType())
])

parsed_df = raw.select(
    col("key"),
    from_json(col("value"), event_schema).alias("event")
).select(
    col("key"),
    col("event.user_id"),
    col("event.event_type"),
    col("event.event_timestamp"),
    col("event.properties")
)

This unpacks the JSON and flattens it into individual columns. If a message doesn't match the schema, from_json will return null, which we can handle later with filters or error routes. In production, we've found it useful to capture malformed records separately for debugging:

from pyspark.sql.functions import when

parsed_with_errors = raw.select(
    col("key"),
    from_json(col("value"), event_schema).alias("event")
).select(
    col("key"),
    col("event"),
    when(col("event").isNull(), col("value")).otherwise(None).alias("raw_error")
)

valid_events = parsed_with_errors.filter(col("event").isNotNull())
error_events = parsed_with_errors.filter(col("event").isNull())

Schema — Use from_json or Avro with a schema registry

Hardcoding schemas in your application code is a recipe for painful deployments. When the upstream team adds a new field to the event payload, you're forced to update, test, and redeploy. A schema registry—like Confluent's Schema Registry or AWS Glue—keeps schemas versioned and accessible, so both producers and consumers stay in sync.

If your organization uses Avro with a schema registry, Spark integrates well via the spark-avro package. Here's how to read Avro-encoded Kafka messages:

from pyspark.sql.functions import from_avro, expr

# Fetch the schema from the registry (pseudo-code; use your registry's client)
schema_registry_url = "http://schema-registry:8081"
subject = "events-value"
# In practice, use a library like confluent-kafka or rest API

# For simplicity, assume we've already fetched the schema as an Avro JSON
avro_schema = """{
  "type": "record",
  "name": "Event",
  "fields": [
    {"name": "user_id", "type": "string"},
    {"name": "event_type", "type": "string"},
    {"name": "event_timestamp", "type": "long"}
  ]
}"""

kafka_df = spark.readStream \
    .format("kafka") \
    .option("kafka.bootstrap.servers", "localhost:9092") \
    .option("subscribe", "events") \
    .option("startingOffsets", "latest") \
    .load()

parsed_df = kafka_df.select(
    from_avro(col("value"), avro_schema).alias("event")
).select(
    col("event.user_id"),
    col("event.event_type"),
    col("event.event_timestamp")
)

Avro is more compact than JSON and enforces stricter schema evolution rules, which reduces the likelihood of silent data corruption. We recommend Avro if your team already has a schema registry running, or if bandwidth and storage are concerns. For teams starting out, JSON with from_json and manual schema updates is acceptable—just be disciplined about schema versioning in your repository.

Write results — To Delta tables or back to Kafka

Once we've parsed and transformed the events, we need to write them somewhere. The two most common destinations are Delta Lake (for batch analytics and OLAP) and back to Kafka (for downstream real-time consumers).

Writing to Delta: Delta tables give us ACID guarantees, schema enforcement, and efficient updates. Streaming writes to Delta use the delta format with a checkpoint directory:

query = parsed_df \
    .writeStream \
    .format("delta") \
    .mode("append") \
    .option("checkpointLocation", "/path/to/checkpoint") \
    .option("path", "/delta/events") \
    .start()

query.awaitTermination()

The checkpoint directory ensures exactly-once semantics: if the Spark job crashes, it resumes from the last committed offset and doesn't duplicate events. This is critical for financial transactions, user activity logs, and any scenario where duplicates skew your analytics.

Writing back to Kafka: Sometimes we need to emit transformed events to another Kafka topic for downstream pipelines. Spark writes to Kafka via the kafka format:

enriched_df = parsed_df.select(
    col("user_id"),
    col("event_type"),
    col("event_timestamp")
).filter(col("event_type").isin(["purchase", "signup"]))

query = enriched_df \
    .select(
        col("user_id").cast("string").alias("key"),
        to_json(struct("*")).alias("value")
    ) \
    .writeStream \
    .format("kafka") \
    .option("kafka.bootstrap.servers", "localhost:9092") \
    .option("topic", "enriched_events") \
    .option("checkpointLocation", "/path/to/kafka/checkpoint") \
    .start()

query.awaitTermination()

Here we're filtering to only important event types, serializing the results back to JSON, and writing to a new topic. The key column determines which Kafka partition receives the message—using user_id ensures all events from the same user go to the same partition, which is useful for stateful downstream processing.

Monitor lag — Track consumer progress and batch duration

A real-time pipeline isn't truly real-time if we don't know how far behind it is. Consumer lag—the difference between the latest offset in Kafka and the offset our Spark job has consumed—is the primary metric we monitor.

Spark's structured streaming API exposes progress information via the StreamingQueryProgress object. We can log this programmatically:

from pyspark.sql.streaming import StreamingQueryListener

class LagMonitor(StreamingQueryListener):
    def onQueryProgress(self, progress):
        sources = progress.sources
        for source in sources:
            if source["description"].startswith("KafkaSource"):
                end_offset = source["endOffset"]
                latest = end_offset.get("0", 0)  # Partition 0
                
                print(f"Lag: {latest - source['currentOffset']}")
                print(f"Batch duration: {progress.batchDuration} ms")

spark.streams.addListener(LagMonitor())

query = parsed_df \
    .writeStream \
    .format("delta") \
    .mode("append") \
    .option("checkpointLocation", "/path/to/checkpoint") \
    .option("path", "/delta/events") \
    .start()

query.awaitTermination()

For production systems, we emit these metrics to a monitoring system like Prometheus or CloudWatch. A typical alerting rule triggers if lag exceeds a threshold (e.g., 1 million messages) for more than 5 minutes, indicating that our pipeline is falling behind.

Batch duration is equally important: if batches take 10 seconds but events arrive every second, lag will grow even if we're processing efficiently. We optimize by tuning maxOffsetsPerTrigger and parallelism:

kafka_df = spark.readStream \
    .format("kafka") \
    .option("kafka.bootstrap.servers", "localhost:9092") \
    .option("subscribe", "events") \
    .option("maxOffsetsPerTrigger", 50000) \
    .load()

# Process with more executors if needed
spark.conf.set("spark.sql.shuffle.partitions", 200)

maxOffsetsPerTrigger limits the number of messages per batch, keeping batch duration predictable. If your cluster can handle it, increase this value and parallelism; if your cluster is resource-constrained, reduce it to avoid out-of-memory errors.

Common mistakes

• Forgetting checkpoint directories: Without checkpoints, Spark cannot resume from where it left off. On failure, you'll either reprocess old data or lose data entirely. Always set checkpointLocation.
• Casting value to string without validation: Not all Kafka values are valid UTF-8. Use from_json or from_avro with proper error handling rather than blindly casting.
• Ignoring batch duration in production: A pipeline that processes 100k events per batch in 30 seconds sounds good until you realize events arrive every second. Monitor both lag and batch duration, not just lag.
• Using stateful operations without careful state management: Operations like groupByKey().mapGroupsWithState() accumulate state in memory. Without proper TTL or cleanup, state can grow unbounded and crash your job.
• Writing schema-less code: Hardcoding field names instead of using a schema registry leads to data pipeline fragility. Invest in a schema registry early.

Key takeaways

• Parse Kafka byte values into structured columns using from_json or from_avro with a well-defined schema.
• Integrate with a schema registry to version schemas and reduce deployment friction as your data model evolves.
• Write transformed data to Delta for durability and exactly-once semantics, or back to Kafka for multi-hop pipelines.
• Monitor consumer lag and batch duration continuously; lag alone doesn't tell the full story if batches are slow.
• Always specify a checkpoint directory to ensure fault tolerance and prevent data loss or duplication.
• Test error paths: malformed messages, schema mismatches, and network failures happen in production—handle them gracefully.