advanced
Kafka Streaming
10 min readLast updated: 2026-07-09
Overview
Connect Spark Structured Streaming to Apache Kafka to read and write real-time message streams.
What You Will Learn
In this lesson, you will learn:
- Kafka Integration: Streaming data from Kafka topics.
- Value Casting: Converting binary messages to structured columns.
- Writing to Kafka: Sending structured query outputs back to Kafka topics.
Detailed Concept Explanation
To read streams from Apache Kafka, you use readStream with the kafka format.
The source schema contains key metadata:
key(binary) andvalue(binary)topic(string)partition(int)offset(long)
Code Examples
Python (PySpark) Implementation
python
from pyspark.sql import SparkSession
spark = SparkSession.builder \
.appName("KafkaStreaming") \
.config("spark.jars.packages", "org.apache.spark:spark-sql-kafka-0-10_2.12:3.2.0") \
.getOrCreate()
# Stream from Kafka topic
kafka_stream = spark.readStream.format("kafka") \
.option("kafka.bootstrap.servers", "localhost:9092") \
.option("subscribe", "realtime-topic") \
.load()
# Cast binary value payload to string
messages = kafka_stream.selectExpr("CAST(value AS STRING) AS msg")
# Print to console
query = messages.writeStream.format("console").start()
query.awaitTermination(5)
query.stop()
Expected Output
text
+---------------------+
| msg|
+---------------------+
|{"event": "purchase"}|
+---------------------+
Execution Plan Diagram (Python & Scala)
Execution Plan Diagram
SparkSession.builder
readStream.format(kafka)
load()
selectExpr(CAST value AS STRING)
writeStream.format(console)
start()
awaitTermination(5)
stop()
Scala Implementation
scala
import org.apache.spark.sql.SparkSession
val spark = SparkSession.builder()
.appName("KafkaStreamingScala")
.config("spark.jars.packages", "org.apache.spark:spark-sql-kafka-0-10_2.12:3.2.0")
.getOrCreate()
val kafkaStream = spark.readStream.format("kafka")
.option("kafka.bootstrap.servers", "localhost:9092")
.option("subscribe", "realtime-topic")
.load()
val messages = kafkaStream.selectExpr("CAST(value AS STRING) AS msg")
val query = messages.writeStream.format("console").start()
query.awaitTermination(5000)
query.stop()
Expected Output
text
+---------------------+
| msg|
+---------------------+
|{"event": "purchase"}|
+---------------------+
Common Mistakes
- Missing Kafka dependencies: Running Kafka stream queries without loading the Kafka connector package dependency at startup, which will throw a class-not-found exception.
Best Practices
- Define schemas for JSON payloads: After casting the binary
valueto a string, usefrom_jsonwith an explicit schema to parse JSON messages into structured columns.