Building a Streaming Pipeline with Kafka and Spark

A streaming pipeline has the same stages as a batch one, read, transform, write, but it never finishes: events keep arriving, and results must be produced while they do. The standard open-source pairing is Apache Kafka to carry the events and Spark Structured Streaming to process them. This guide builds that pipeline end to end and runs it on real data. If Kafka's topics, partitions and offsets are new to you, start with Apache Kafka explained.

The design

1. Events land on a Kafka topic as JSON, keyed by pickup zone.
2. Spark reads the topic as a stream and parses each event.
3. Events are aggregated in one-hour windows by the time they happened, with a watermark to allow for late arrivals.
4. Finished windows are written to a sink that a dashboard could read.

Spark's Kafka source comes from a separate package, spark-sql-kafka-0-10, added with --packages when submitting the job (Spark docs: Kafka integration).

The pipeline

To keep the example self-contained, the job first writes one day of January 2024 New York taxi trips (NYC TLC: trip record data) to the topic, standing in for a live producer, then runs the streaming query over it:

import uuid
from pyspark.sql import SparkSession, functions as F
from pyspark.sql.types import DoubleType, IntegerType, StructField, StructType, TimestampType

spark = SparkSession.builder.appName("trips-stream").getOrCreate()
spark.sparkContext.setLogLevel("ERROR")
BROKER = "kafka-demo:19092"
TOPIC = f"trips-{uuid.uuid4().hex[:8]}"                        # a fresh topic for each run

# 1. put one day of real trips on the topic as JSON events, standing in for a live producer
day = (spark.read.parquet("yellow_tripdata_2024-01.parquet")
       .where((F.col("tpep_pickup_datetime") >= "2024-01-15") & (F.col("tpep_pickup_datetime") < "2024-01-16"))
       .select(F.col("tpep_pickup_datetime").cast("timestamp").alias("pickup"),
               F.col("PULocationID").alias("zone"), F.col("fare_amount").alias("fare")))
(day.select(F.col("zone").cast("string").alias("key"), F.to_json(F.struct("pickup", "zone", "fare")).alias("value"))
    .write.format("kafka").option("kafka.bootstrap.servers", BROKER).option("topic", TOPIC).save())
produced = day.count()
print("events produced:", produced)

# 2. read the topic as a stream and parse the JSON payload
schema = StructType([StructField("pickup", TimestampType()), StructField("zone", IntegerType()), StructField("fare", DoubleType())])
events = (spark.readStream.format("kafka")
          .option("kafka.bootstrap.servers", BROKER).option("subscribe", TOPIC).option("startingOffsets", "earliest")
          .load()
          .select(F.from_json(F.col("value").cast("string"), schema).alias("event")).select("event.*"))

# 3. aggregate in one-hour event-time windows, waiting up to 30 minutes for late events
hourly = (events.withWatermark("pickup", "30 minutes")
                .groupBy(F.window("pickup", "1 hour"))
                .agg(F.count("*").alias("trips"), F.round(F.sum("fare"), 2).alias("fares")))

# 4. write finished windows to a sink; availableNow processes everything on the topic, then stops
query = (hourly.writeStream.outputMode("append").format("memory").queryName("hourly_trips")
               .trigger(availableNow=True).start())
query.awaitTermination()

result = spark.sql("SELECT date_format(window.start, 'HH:mm') AS hour, trips, fares FROM hourly_trips ORDER BY window.start")
print("windows emitted:", result.count())
result.show(24, truncate=False)
closed = result.agg(F.sum("trips")).first()[0]
print(f"trips in emitted windows: {closed} of {produced}; {produced - closed} wait in a window the watermark has not passed")
spark.stop()

events produced: 77033
windows emitted: 23
+-----+-----+---------+
|hour |trips|fares    |
+-----+-----+---------+
|00:00|2237 |48799.89 |
|01:00|1085 |20089.81 |
|02:00|754  |13998.68 |
|03:00|443  |8556.17  |
|04:00|358  |9909.33  |
|05:00|483  |14872.73 |
|06:00|906  |24632.68 |
|07:00|1361 |29936.45 |
|08:00|2440 |42761.73 |
|09:00|3098 |53849.1  |
|10:00|3847 |66718.73 |
|11:00|4605 |77844.33 |
|12:00|5077 |86363.22 |
|13:00|5477 |100493.7 |
|14:00|5944 |115933.79|
|15:00|5791 |108218.37|
|16:00|5686 |114265.16|
|17:00|5739 |107304.51|
|18:00|5273 |91778.94 |
|19:00|4540 |91215.43 |
|20:00|3752 |76981.91 |
|21:00|3352 |63926.79 |
|22:00|2756 |60590.43 |
+-----+-----+---------+

trips in emitted windows: 75004 of 77033; 2029 wait in a window the watermark has not passed

Stage by stage

Producing. Each trip becomes one event: a JSON value and a key. Writing to Kafka needs only a value column; the key is optional but matters, because Kafka sends events with the same key to the same partition in order (Kafka docs: introduction).

Reading and parsing. The Kafka source returns key and value as binary, along with topic, partition, offset and timestamp, so the value is cast to a string and parsed with from_json against a declared schema. startingOffsets is set to earliest so the query reads the whole topic; the default for streaming queries is latest, which would skip everything already there, a common first-run surprise. The option applies only when a query starts for the first time; afterwards it resumes from its checkpoint.

Windowing on event time. F.window("pickup", "1 hour") groups events by when the trip happened, not when Spark processed it. The watermark, 30 minutes, tells Spark how long to wait for late events before finalising a window and discarding its state.

Writing. In append mode, a windowed aggregate is emitted only once, after the watermark has passed the end of its window (Spark docs: streaming APIs). availableNow makes the query process everything currently on the topic and stop, which suits a demonstration and scheduled catch-up jobs; a continuous job would use a processing-time trigger and run indefinitely.

Why 23 windows, not 24

The day has 24 hours and 77,033 events, but the output has 23 windows holding 75,004 trips. The 2,029 trips between 23:00 and midnight are not lost. Their window ends at 24:00, and the latest event Spark has seen is just before midnight, so the watermark stands near 23:30 and has not passed the window's end. The window stays open, waiting for either late events or newer events that push the watermark forward. On a live topic, the first events after 00:30 would release it.

This is the behaviour that confuses newcomers most: with append mode and a watermark, results appear after a delay of the window length plus the watermark. It is the price of emitting each window exactly once, final and correct. If a dashboard needs running totals sooner, update mode emits a window's current value on every trigger and revises it as events arrive. Handling late-arriving data explains how to choose the watermark from measured delays.

What production adds

• A durable sink and a checkpoint. The memory sink is for testing. A real job writes to a table or topic with a checkpointLocation, which stores offsets and window state so the query resumes exactly where it stopped.
• Idempotent or transactional writes, because a restart reprocesses the last incomplete batch; idempotency in data pipelines covers the options.
• Schema management: JSON with a hand-written schema works until a producer changes a field. Avro or Protobuf with a schema registry makes changes explicit.
• Malformed events: from_json returns nulls for values it cannot parse. Route those rows to a dead-letter topic instead of dropping them silently.
• Monitoring: input rate, processing rate, batch duration and consumer lag show whether the job is keeping up; a processing rate persistently below the input rate means the job is falling behind and lag will grow.
• Partition count and parallelism sized together, since each Kafka partition is read by one Spark task at a time.

The DataFrame operations themselves are the same as in batch, covered in the PySpark tutorial.

Spark Streaming, Apache Kafka basics, real-time data pipelines and a real-time streaming pipeline project are part of the Data Engineering course.

Sources

• Structured Streaming and Kafka integration guide (Apache Spark documentation) — Apache Spark
• Structured Streaming APIs on DataFrames and Datasets (Apache Spark documentation) — Apache Spark
• Introduction (Apache Kafka documentation) — Apache Kafka
• TLC Trip Record Data (NYC Taxi and Limousine Commission) — NYC Taxi and Limousine Commission