Spark Streaming with Kafka

Apache Kafka stores streams of events in named topics. Many companies use it as the central pipe for clicks, payments, and sensor readings. Spark Structured Streaming reads from Kafka, processes the events, and writes results to files, tables, or another Kafka topic.

The Layman View

Kafka works like a post office with labelled mailboxes. Producers drop letters into a mailbox called a topic. Spark picks up the letters, reads them, and files the results. The mailbox keeps every letter for a set time, so Spark can restart and collect missed mail.

The Big Picture

Producers                 Kafka                    Spark                   Sinks
---------                 -----                    -----                   -----
Web app   ---\            topic: clicks            Structured Streaming    Delta table
Mobile app ---->  [P0][P1][P2][P3]  ---------->   parse, filter, count ---> Parquet files
Sensors   ---/            (partitions)                                     Kafka topic

Adding the Connector

Spark needs the Kafka connector package. Match the Scala and Spark versions to your installation.

spark-submit --packages org.apache.spark:spark-sql-kafka-0-10_2.13:YOUR_SPARK_VERSION stream_job.py

Reading from Kafka

raw = (spark.readStream.format("kafka")
         .option("kafka.bootstrap.servers", "broker1:9092,broker2:9092")
         .option("subscribe", "clicks")
         .option("startingOffsets", "latest")
         .load())

raw.printSchema()

root
 |-- key: binary
 |-- value: binary
 |-- topic: string
 |-- partition: integer
 |-- offset: long
 |-- timestamp: timestamp

Kafka delivers the key and value as raw bytes. Cast them to strings before use.

Parsing JSON Messages

from pyspark.sql import functions as F
from pyspark.sql.types import StructType, StructField, StringType, IntegerType

schema = StructType([
    StructField("user",   StringType()),
    StructField("page",   StringType()),
    StructField("amount", IntegerType())
])

events = (raw
    .select(F.col("value").cast("string").alias("json"), "timestamp")
    .select(F.from_json("json", schema).alias("e"), "timestamp")
    .select("e.*", "timestamp"))

bytes  --cast-->  JSON text  --from_json-->  columns: user | page | amount | timestamp

Processing and Writing

page_counts = (events
    .withWatermark("timestamp", "5 minutes")
    .groupBy(F.window("timestamp", "1 minute"), "page")
    .count())

query = (page_counts.writeStream
    .format("parquet")
    .option("path", "/data/page_counts")
    .option("checkpointLocation", "/checkpoints/page_counts")
    .outputMode("append")
    .start())

Writing Back to Kafka

A Kafka sink needs a column named value, and optionally a column named key.

out = events.select(F.col("user").alias("key"),
                    F.to_json(F.struct("user", "page")).alias("value"))

(out.writeStream.format("kafka")
    .option("kafka.bootstrap.servers", "broker1:9092")
    .option("topic", "clicks_clean")
    .option("checkpointLocation", "/checkpoints/clicks_clean")
    .start())

Offsets and Checkpoints

Kafka numbers each message in a partition. Spark records the last processed offset of every partition in the checkpoint folder. A restarted job resumes from those offsets and skips nothing.

Partition 0: ...[40][41][42][43][44]
                            ^
                  checkpoint stores offset 43; restart continues at 44

• startingOffsets applies only the first time a query runs. Later runs follow the checkpoint.
• maxOffsetsPerTrigger limits the batch size and prevents overload after a long pause.

Good Practices

• Use a separate checkpoint folder for every query.
• Define the message schema explicitly.
• Keep the number of Spark tasks close to the number of Kafka partitions.
• Monitor the gap between the newest Kafka offset and the processed offset. A growing gap signals that Spark runs behind.

Starting from Earlier Messages

Kafka keeps messages for a retention period, so a new Spark query can read history. Set startingOffsets to earliest for a first run that processes every stored message. Set it to latest to read only new arrivals.

.option("startingOffsets", "earliest")

Kafka log:  [0][1][2][3][4][5][6][7][8]   now
earliest -> starts at 0
latest   -> starts after 8

Subscribing to Several Topics

Use a comma-separated list or a pattern to read many topics in one query. The topic column in the output tells you the origin of each row.

.option("subscribe", "clicks,payments")
.option("subscribePattern", "events_.*")

Testing Without a Cluster

Run a local Kafka broker in a container, produce a few sample messages, and print the stream to the console sink. Check that the parsed columns look right before you write to permanent storage.

query = events.writeStream.format("console").outputMode("append").start()