Real-Time Analytics with PySpark and Kafka

Build a streaming pipeline that processes live clickstream events with PySpark.

Introduction

30 Second Summary

Every click, scroll, and purchase on an e-commerce site generates a stream of events. Processing these events in real time is how engineering teams power live dashboards, detect anomalies, and trigger personalized recommendations within seconds of user activity.

In this project, you will build a real-time clickstream analytics pipeline using PySpark Structured Streaming and Apache Kafka. You will produce simulated user events, deduplicate them, aggregate them into time windows with late-data handling, and write enriched results to multiple downstream sinks.

What You'll Build

You will run a complete streaming pipeline on your local machine and watch live clickstream events flow from a Kafka topic through Spark, get deduplicated and windowed, then land on a downstream Kafka topic and a local JSON data lake simultaneously.

By the end of this project, you'll have:

  • A live pipeline where you can watch clickstream events streaming through your Spark console as they flow from a Kafka producer, including deliberately late and duplicate events.
  • Watermarked windowed aggregations that correctly count events in 1-minute tumbling windows, handle late arrivals within a 2-minute threshold, and drop data that arrives too late.
  • A multi-sink output using foreachBatch that writes deduplicated analytics to both a downstream Kafka topic and a partitioned local JSON data lake with exactly-once semantics via batch ID tracking.
  • Secret Mission: a stream-stream join that enriches clickstream events with user profile data from a second Kafka topic, using watermarks and time constraints for automatic state cleanup.

Are there any prerequisites?

This project assumes working knowledge of PySpark, Spark, and Kafka individually. You should be comfortable writing Python and running Docker containers.

If you have not used these tools before, complete introductory projects for each one first, then return here to combine them into a production-shaped streaming pipeline.

Before We Start

Before we dive in, let's lock in what you're about to build and why it matters for your data engineering career.

Set Up Kafka and PySpark

Your streaming pipeline needs two core pieces of infrastructure: Kafka as the event bus that carries clickstream data, and PySpark to process those events in real time.

In this step, you will spin up a local Kafka broker using Docker Compose, create the topics your pipeline reads from and writes to, and confirm PySpark can create a SparkSession.

In this step, get ready to:
  • Write a Docker Compose file that runs Kafka in KRaft mode.
  • Start the Kafka broker and create two event topics.
  • Verify PySpark by creating a SparkSession that prints the version.
Write the Docker Compose configuration

Kafka traditionally required a separate Zookeeper process for cluster coordination. Modern Kafka versions support KRaft mode, which eliminates Zookeeper entirely. Your single-node broker will act as both broker and controller in one container.

  • In your project directory, create a file called docker-compose.yml and paste the following configuration:
services:
  kafka:
    image: confluentinc/cp-kafka:7.6.12
    hostname: kafka
    container_name: kafka
    ports:
      - "9092:9092"
    environment:
      KAFKA_NODE_ID: 1
      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: CONTROLLER:PLAINTEXT,PLAINTEXT:PLAINTEXT,HOST:PLAINTEXT
      KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://kafka:29092,HOST://localhost:9092
      KAFKA_LISTENERS: PLAINTEXT://0.0.0.0:29092,CONTROLLER://0.0.0.0:9093,HOST://0.0.0.0:9092
      KAFKA_CONTROLLER_LISTENER_NAMES: CONTROLLER
      KAFKA_INTER_BROKER_LISTENER_NAME: PLAINTEXT
      KAFKA_CONTROLLER_QUORUM_VOTERS: 1@kafka:9093
      KAFKA_PROCESS_ROLES: broker,controller
      KAFKA_GROUP_INITIAL_REBALANCE_DELAY_MS: 0
      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1
      KAFKA_TRANSACTION_STATE_LOG_REPLICATION_FACTOR: 1
      KAFKA_TRANSACTION_STATE_LOG_MIN_ISR: 1
      CLUSTER_ID: MkU3OEVBNTcwNTJENDM2Qk

What does this configuration do?

  • confluentinc/cp-kafka:7.6.12 is the Confluent Platform image bundling Apache Kafka 3.6.x.
  • KAFKA_PROCESS_ROLES: broker,controller enables KRaft combined mode, so one container handles both roles without Zookeeper.
  • The HOST listener on port 9092 lets your Python scripts on the host machine connect to Kafka inside Docker.
  • Replication factors are set to 1 because this is a single-node development cluster.
  • Save docker-compose.yml.
Start Kafka and create the event topics

Your pipeline uses two Kafka topics: one for raw clickstream events coming in, and one for the analytics results going out.

  • Start the Kafka container in the background by running this command:
docker compose up -d
  • Confirm the container is healthy by running:
docker compose ps

You should see one container named kafka with a status showing it is running.

Container not starting?

  • Make sure Docker Desktop is running. You need the Docker engine active before docker compose can create containers.
  • If you see a port conflict on 9092, check that no other process is using that port. Stop any existing Kafka or service on 9092 first.
  • help me debug why my Kafka container won't start

Now create the two topics your pipeline needs. The broker needs a few seconds to finish starting before it accepts topic creation requests.

  • Wait about 10 seconds for the broker to initialize, then create the clickstream-events topic by running:
docker exec kafka kafka-topics --create --topic clickstream-events --bootstrap-server localhost:9092 --partitions 1 --replication-factor 1
  • Create the clickstream-analytics output topic by running:
docker exec kafka kafka-topics --create --topic clickstream-analytics --bootstrap-server localhost:9092 --partitions 1 --replication-factor 1

Each command prints a confirmation message showing the topic was created successfully.

Why two topics?

The clickstream-events topic is where your producer will send raw user events. The clickstream-analytics topic is where PySpark will write aggregated results after processing.

This separation keeps raw events and processed output in different channels. Downstream consumers can subscribe to whichever topic matches their needs.

Seeing a connection error when creating topics?

  • The broker may not be fully initialized yet. Wait 15-20 seconds after docker compose up -d and try again.
  • Run docker compose logs kafka to check whether the broker has finished starting. Look for log lines mentioning the broker being ready.
  • help me troubleshoot my Kafka topic creation error
Verify PySpark with a SparkSession

PySpark is already installed in your environment. Before building the streaming consumer, confirm it can create a SparkSession successfully.

  • Run this quick verification command:
python -c "from pyspark.sql import SparkSession; spark = SparkSession.builder.appName('VersionCheck').getOrCreate(); print(spark.version); spark.stop()"

Before you check the output, what version number do you expect to see printed?

You should see 3.5.9 printed to the terminal, along with some Spark startup log lines above it. That confirms PySpark can initialize a session and is ready to process streams.

Seeing a ModuleNotFoundError or wrong version?

  • If Python cannot find pyspark, ensure you are using the same Python environment where you ran pip install pyspark==3.5.9.
  • If the version printed is not 3.5.9, you may have multiple PySpark installations. Run pip show pyspark to check the installed version and reinstall if needed.
  • PySpark requires Java 8, 11, or 17 on your system. If you see a Java-related error, install a compatible JDK first.
  • help me fix my PySpark installation

✔️ Awesome, I've got everything!

Great. Double check you have saved your file and that both Kafka topics were created successfully.

ⓧ I'd like to double check the full code

Here is the complete docker-compose.yml for comparison:

services:
  kafka:
    image: confluentinc/cp-kafka:7.6.12
    hostname: kafka
    container_name: kafka
    ports:
      - "9092:9092"
    environment:
      KAFKA_NODE_ID: 1
      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: CONTROLLER:PLAINTEXT,PLAINTEXT:PLAINTEXT,HOST:PLAINTEXT
      KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://kafka:29092,HOST://localhost:9092
      KAFKA_LISTENERS: PLAINTEXT://0.0.0.0:29092,CONTROLLER://0.0.0.0:9093,HOST://0.0.0.0:9092
      KAFKA_CONTROLLER_LISTENER_NAMES: CONTROLLER
      KAFKA_INTER_BROKER_LISTENER_NAME: PLAINTEXT
      KAFKA_CONTROLLER_QUORUM_VOTERS: 1@kafka:9093
      KAFKA_PROCESS_ROLES: broker,controller
      KAFKA_GROUP_INITIAL_REBALANCE_DELAY_MS: 0
      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1
      KAFKA_TRANSACTION_STATE_LOG_REPLICATION_FACTOR: 1
      KAFKA_TRANSACTION_STATE_LOG_MIN_ISR: 1
      CLUSTER_ID: MkU3OEVBNTcwNTJENDM2Qk

Your infrastructure is ready. Kafka is listening for events on port 9092, and PySpark is primed to consume them. Next up, you will write a producer that pumps realistic clickstream events into Kafka.

Produce Clickstream Events to Kafka

Your Kafka cluster is running and both topics are ready to receive data. Now you need something to actually generate events.

Without a stream of data flowing into Kafka, there is nothing for Spark to consume downstream. In this step, you will write a Python producer that simulates realistic e-commerce clickstream traffic. The producer deliberately injects late-arriving events and duplicates so you can observe how Spark handles them in later steps.

In this step, get ready to:
  • Write a producer script that sends JSON clickstream events to Kafka at ~5 events per second.
  • Verify events are landing in Kafka by consuming them from the terminal.
Write the producer script

The producer needs to generate events that look like real e-commerce clickstream data. Each event carries a unique ID, a user, an action type, a timestamp, and the page it happened on.

  • Create a new file called producer.py in your project folder by pasting the following setup code:
import json
import time
import random
import uuid
from datetime import datetime, timedelta
from kafka import KafkaProducer

producer = KafkaProducer(
    bootstrap_servers="localhost:9092",
    value_serializer=lambda v: json.dumps(v).encode("utf-8"),
)

EVENT_TYPES = ["page_view", "add_to_cart", "purchase"]
PAGES = ["/home", "/products", "/product/1", "/product/2", "/cart", "/checkout"]
sent_events = []

print("Starting clickstream producer... (Ctrl+C to stop)")

try:
    while True:
        event_id = str(uuid.uuid4())
        now = datetime.utcnow()

        # 10% of events are late (30-90 seconds behind)
        if random.random() < 0.10:
            delay = random.randint(30, 90)
            event_time = now - timedelta(seconds=delay)
        else:
            event_time = now

What does this code do?

This sets up the Kafka connection and the event generation logic:

  • KafkaProducer connects to your local broker on port 9092 and serializes every event to JSON bytes automatically.
  • EVENT_TYPES and PAGES define the pool of realistic values each event draws from.
  • The late-data logic gives 10% of events a timestamp 30-90 seconds in the past. This simulates mobile devices or flaky networks delivering events out of order.
  • sent_events keeps a buffer of recent events so the duplicate logic (next chunk) can resend one.
  • Add the event construction, sending, and cleanup logic directly below the previous code in producer.py:

        event = {
            "event_id": event_id,
            "user_id": f"user_{random.randint(1, 50)}",
            "event_type": random.choice(EVENT_TYPES),
            "timestamp": event_time.strftime("%Y-%m-%dT%H:%M:%S.000Z"),
            "page": random.choice(PAGES),
        }

        producer.send("clickstream-events", value=event)
        sent_events.append(event)

        # 5% chance to resend a previous event (duplicate)
        if sent_events and random.random() < 0.05:
            duplicate = random.choice(sent_events[-20:])
            producer.send("clickstream-events", value=duplicate)

        time.sleep(0.2)
except KeyboardInterrupt:
    print("Producer stopped.")
finally:
    producer.flush()
    producer.close()

What does this code do?

This is the heart of the event loop:

  • Each event is a dictionary with five fields that mirror a real clickstream schema.
  • producer.send("clickstream-events", value=event) pushes each event to the Kafka topic you created in the previous step.
  • 5% of iterations pick a random recent event from sent_events and resend it as a duplicate. This simulates Kafka's at-least-once delivery guarantee.
  • time.sleep(0.2) throttles to roughly 5 events per second.
  • The finally block ensures all buffered messages are flushed to Kafka before the script exits.
  • Save producer.py.

✔️ Awesome, I've got everything!

Great. Double check you have saved the file before moving on.

ⓧ I'd like to double check the full code

Scroll up and confirm both code chunks are present in order:

  • Chunk 1: Imports, KafkaProducer setup, constants, and the try/while opener with late-data logic.
  • Chunk 2: Event dict, producer.send, duplicate logic, time.sleep, and the except/finally cleanup.

Make sure there is exactly one blank line between each logical section (no double blanks) and that the indentation under while True: is 8 spaces.

Run the producer and verify events in Kafka

With the script saved, you can start pumping events into Kafka. You will run the producer in one terminal and then consume events in a second terminal to confirm they arrive.

  • Start the producer by running this command:
python producer.py

You should see the message Starting clickstream producer... (Ctrl+C to stop) and the script keeps running.

  • Open a second terminal window.
  • Before you run the next command, predict what you expect to see. Will the consumer show raw bytes, formatted JSON, or nothing at all?
  • Consume events from the clickstream-events topic by running:
docker exec kafka kafka-console-consumer --bootstrap-server localhost:9092 --topic clickstream-events --from-beginning

You should see JSON objects streaming across the terminal, each containing event_id, user_id, event_type, timestamp, and page.

What should I look for?

Watch the timestamp field across several events. Most timestamps cluster around "now", but roughly 1 in 10 shows a time 30-90 seconds in the past.

You may also spot the same event_id appearing twice in a row. Those are the intentional duplicates your producer injected.

Not seeing any events?

  • Make sure producer.py is still running in the first terminal. If it crashed, check for import errors (did you install kafka-python?).
  • Confirm your Kafka container is healthy by running docker compose ps in a third terminal. The kafka service should show status running.
  • If you see a connection refused error, verify the bootstrap_servers in your producer matches localhost:9092.

Why aren't my clickstream events showing up in the Kafka consumer?

  • Press Ctrl+C in the consumer terminal to stop it once you have confirmed events are flowing.
  • Leave the producer running in the first terminal. You will need it for the next step.

Your data pipeline now has live fuel. Next up, you will point PySpark at this stream and see what happens when you try to aggregate events without telling Spark how to handle time.

Stream from Kafka and Hit the Output Mode Wall

Your producer is pumping clickstream events into Kafka at five per second. Now you need something to read and process those events in real time.

In this step you will write your first Structured Streaming consumer that reads from Kafka, then attempt a windowed aggregation. You will hit a real error that every Spark streaming developer encounters: the AnalysisException that blocks append mode without a watermark. Fixing it teaches you exactly why watermarking exists.

In this step, get ready to:
  • Read live Kafka events into a PySpark DataFrame and print them to the console.
  • Add a windowed aggregation and observe the AnalysisException that Spark throws.
  • Resolve the error with watermarking so finalized window counts emit in append mode.
Write the initial streaming consumer

You need a PySpark script that connects to Kafka and deserializes the JSON events your producer sends. The script will define a schema matching the event fields, then write raw events to the console so you can confirm the pipeline works end-to-end.

  • Create a new file called stream_consumer.py in your project directory.
  • Add the imports and SparkSession setup by pasting this code:
from pyspark.sql import SparkSession
from pyspark.sql.functions import (
    from_json,
    col,
    window,
    to_json,
    struct,
    lit,
)
from pyspark.sql.types import (
    StructType,
    StructField,
    StringType,
    TimestampType,
)
import os

# Create SparkSession with Kafka connector
spark = (
    SparkSession.builder.appName("ClickstreamAnalytics")
    .config(
        "spark.jars.packages",
        "org.apache.spark:spark-sql-kafka-0-10_2.12:3.5.9",
    )
    .config("spark.sql.streaming.schemaInference", "false")
    .getOrCreate()
)

spark.sparkContext.setLogLevel("WARN")

What does this code do?

The SparkSession is your entry point to all Spark functionality. The spark.jars.packages config tells Spark to download the Kafka connector JAR automatically at startup.

Setting the log level to WARN reduces console noise so you can see your streaming output clearly.

  • Below the SparkSession setup, add the schema definition and Kafka read logic:
# Define schema matching producer's JSON events
schema = StructType(
    [
        StructField("event_id", StringType(), True),
        StructField("user_id", StringType(), True),
        StructField("event_type", StringType(), True),
        StructField("timestamp", StringType(), True),
        StructField("page", StringType(), True),
    ]
)

# Read from Kafka
raw_stream = (
    spark.readStream.format("kafka")
    .option("kafka.bootstrap.servers", "localhost:9092")
    .option("subscribe", "clickstream-events")
    .option("startingOffsets", "latest")
    .option("maxOffsetsPerTrigger", 1000)
    .load()
)

# Parse JSON and cast timestamp
events = (
    raw_stream.selectExpr("CAST(value AS STRING) as json_str")
    .select(from_json(col("json_str"), schema).alias("data"))
    .select("data.*")
    .withColumn("timestamp", col("timestamp").cast(TimestampType()))
)

What does this code do?

The schema tells Spark how to parse the JSON strings your producer sends. Each Kafka message arrives as a binary value column, so you cast it to a string, then parse it with from_json.

The maxOffsetsPerTrigger option caps how many messages enter each micro-batch. This is your primary backpressure knob for Kafka sources.

  • Add a console sink at the bottom of the file so you can see events flowing through:
# Write raw events to console
query = (
    events.writeStream.outputMode("append")
    .format("console")
    .option("truncate", "false")
    .start()
)

query.awaitTermination()

Why append mode?

Append mode writes only new rows to the sink each trigger. For a simple pass-through (no aggregation), append mode works perfectly because each event is independent.

  • Save stream_consumer.py.
  • Make sure your producer is still running in another terminal. If you stopped it, restart it by running:
python producer.py
  • In a separate terminal, run the streaming consumer:
spark-submit --packages org.apache.spark:spark-sql-kafka-0-10_2.12:3.5.9 stream_consumer.py

After Spark downloads the Kafka connector JAR (first run takes a moment), you should see batches of parsed JSON events printing to your console every few seconds.

What should I see?

A table with columns event_id, user_id, event_type, timestamp, and page appears every micro-batch. Each row is a live clickstream event flowing from your producer through Kafka into Spark.

Not seeing any events?

  • Confirm your producer is running in another terminal and printing its output.
  • Check that Kafka is healthy with docker compose ps. The container should show a healthy status.
  • The consumer uses startingOffsets set to latest, so it only reads events produced after it starts. Make sure the producer was already running before you launched the consumer.

Still stuck? Why isn't my Spark consumer reading events from Kafka?

  • Stop the streaming consumer with Ctrl+C once you have confirmed events are flowing.
Add a windowed aggregation and hit the output mode wall

Seeing raw events is useful, but analytics require aggregation. You want to count how many events of each type arrive per minute using tumbling windows.

  • In stream_consumer.py, find this console sink code at the bottom of the file:
# Write raw events to console
query = (
    events.writeStream.outputMode("append")
    .format("console")
    .option("truncate", "false")
    .start()
)

query.awaitTermination()
  • Replace that entire block with a windowed aggregation followed by a console sink:
# Windowed aggregation: count events per type per minute
windowed_counts = events.groupBy(
    window(col("timestamp"), "1 minute"),
    col("event_type"),
).count()

# Write aggregated counts to console
query = (
    windowed_counts.writeStream.outputMode("append")
    .format("console")
    .option("truncate", "false")
    .start()
)

query.awaitTermination()

What does this aggregation do?

The window(col("timestamp"), "1 minute") creates tumbling 1-minute windows based on event time. The groupBy then counts how many events of each event_type fall into each window.

You are still using append mode, expecting Spark to emit one row per finalized window.

  • Save stream_consumer.py.

Before you run this, what do you think will happen? Will Spark know when a 1-minute window is "done" if you have not told it how late data can arrive?

  • Run the updated consumer:
spark-submit --packages org.apache.spark:spark-sql-kafka-0-10_2.12:3.5.9 stream_consumer.py

The query fails immediately with an AnalysisException:

Append output mode not supported when there are streaming aggregations on streaming DataFrames/DataSets without watermark

This error is intentional. Without a watermark, Spark has no way to know when a window is complete. A window could receive late data forever, so Spark refuses to emit "final" counts in append mode.

Why does Spark throw this error?

In append mode, rows can never be updated once emitted. Spark needs a guarantee that a window will not receive more data before it outputs that window's count.

A watermark gives Spark that guarantee. It says: "data arriving more than X time behind the latest event can be ignored." Once the watermark passes a window's end time, Spark knows the window is finalized and can safely append the result.

Fix the error with watermarking

Now you know the problem: Spark needs to know how late data can arrive. Your producer sends events up to 90 seconds late, so a 2-minute watermark covers those late arrivals comfortably.

  • In stream_consumer.py, find this line:
# Windowed aggregation: count events per type per minute
windowed_counts = events.groupBy(
    window(col("timestamp"), "1 minute"),
    col("event_type"),
).count()
  • Replace it with a watermark, deduplication, and windowed aggregation:
# Deduplicate within watermark window
deduplicated = events.withWatermark("timestamp", "2 minutes").dropDuplicatesWithinWatermark(
    ["event_id"]
)

# Windowed aggregation
windowed_counts = (
    deduplicated.groupBy(
        window(col("timestamp"), "1 minute"),
        col("event_type"),
    ).count()
)

What does this fix?

The withWatermark("timestamp", "2 minutes") tells Spark: any event arriving more than 2 minutes behind the latest observed timestamp can be dropped. Once the watermark passes a window's end, Spark knows no more data will arrive for that window and emits the final count.

The dropDuplicatesWithinWatermark(["event_id"]) removes duplicate events (your producer sends 5% duplicates) using the watermark to bound how long Spark holds deduplication state. Without a watermark, deduplication state would grow forever.

  • Also update the console sink's output mode to append (it should already be append from the previous edit). No change needed there.
  • Save stream_consumer.py.

What do you expect to see now? The watermark is 2 minutes, and windows are 1 minute wide. Think about when Spark will emit the first finalized window count.

  • Run the consumer again:
spark-submit --packages org.apache.spark:spark-sql-kafka-0-10_2.12:3.5.9 stream_consumer.py

This time the query starts without error. After approximately 3 minutes (one 1-minute window plus the 2-minute watermark delay), you see finalized window counts appearing in the console output.

What should I see?

After the watermark passes, each batch prints rows showing a window range, an event type, and a count. For example, a row might show window [2026-07-29 10:05:00, 2026-07-29 10:06:00], event_type page_view, count 15.

Your producer's late events (30-90 seconds behind) land correctly in past windows because they arrive within the 2-minute watermark threshold. Duplicates are removed before counting.

Query starts but no output appears?

  • Append mode only emits finalized windows. Wait at least 3 minutes for the first window to close and the watermark to advance past it.
  • Check that your producer is still running and sending events. The watermark advances only when new events arrive with more recent timestamps.
  • Verify the timestamp column is cast to TimestampType in your events DataFrame. Without a proper timestamp type, the watermark has nothing to track.

Need help? Why am I not seeing windowed counts after adding my watermark?

  • Stop the consumer with Ctrl+C once you have seen at least one batch of windowed counts.

✔️ Awesome, I've got everything!

Great. Double check you have saved stream_consumer.py with the watermark and deduplication logic in place.

ⓧ I'd like to double check the full code

Your stream_consumer.py should contain these sections in order. Scroll up and confirm each one is present:

  • Imports (SparkSession, from_json, col, window, to_json, struct, lit, types, and os).
  • SparkSession creation with appName("ClickstreamAnalytics") and the Kafka connector package config.
  • Schema definition with 5 StructFields: event_id, user_id, event_type, timestamp, page.
  • Kafka readStream with subscribe set to clickstream-events and maxOffsetsPerTrigger set to 1000.
  • JSON parse chain producing the events DataFrame with timestamp cast to TimestampType.
  • Watermark + deduplication: events.withWatermark("timestamp", "2 minutes").dropDuplicatesWithinWatermark(["event_id"]).
  • Windowed aggregation: deduplicated.groupBy(window(...), col("event_type")).count().
  • Console sink with outputMode("append") and awaitTermination().

You have a working streaming aggregation that handles late data and removes duplicates. Next up, you will swap the console sink for a proper Kafka output with checkpointing so results survive restarts.

Deduplicate and Write to Kafka with Checkpointing

In the last step, you resolved the output mode limitation and got windowed aggregation counts flowing to the console. The next problem is getting those results somewhere useful.

A streaming pipeline that only prints to the terminal is a dead end in production. Writing results to a downstream Kafka topic completes the read-process-write loop that production teams rely on. In this step, you will use foreachBatch to send your deduplicated, windowed analytics to the clickstream-analytics topic, with checkpointing to guarantee exactly-once recovery across restarts.

In this step, get ready to:
  • Write a foreachBatch function that sends windowed results to Kafka.
  • Configure the writeStream with checkpointing and a processing time trigger.
  • Verify checkpoint recovery by stopping and restarting the pipeline.
Write the foreachBatch sink function

Spark's built-in Kafka sink only supports direct writeStream.format("kafka") for simple cases. The foreachBatch API gives you full control over each micro-batch as a standard DataFrame, letting you write it to any system using batch operations.

  • In stream_consumer.py, find the windowed_counts definition (the last block in your file).
  • Add the following write_to_sinks function below it:
# foreachBatch handler: write each micro-batch to the Kafka analytics topic
def write_to_sinks(batch_df, batch_id):
    # Skip empty batches (no finalized windows yet)
    if batch_df.isEmpty():
        return

    # Cache the batch so we don't recompute for each sink
    batch_df.persist()

    # Format windowed counts as JSON for the Kafka value column
    kafka_output = batch_df.select(
        to_json(
            struct(
                col("window").getField("start").alias("window_start"),
                col("window").getField("end").alias("window_end"),
                col("event_type"),
                col("count"),
            )
        ).alias("value")
    )
    # Write this batch to the downstream analytics topic
    (
        kafka_output.write.format("kafka")
        .option("kafka.bootstrap.servers", "localhost:9092")
        .option("topic", "clickstream-analytics")
        .save()
    )

    batch_df.unpersist()

What does this code do?

The write_to_sinks function receives two arguments every micro-batch: the batch DataFrame and a unique batch ID.

  • batch_df.isEmpty() skips empty batches that occur before any windows finalize.
  • batch_df.persist() caches the batch in memory so Spark does not recompute it if you add more sinks later.
  • to_json(struct(...)) serializes each row into a JSON string under a value column, which is what Kafka expects.
  • kafka_output.write.format("kafka") uses a standard batch write to push the data to the clickstream-analytics topic.
  • batch_df.unpersist() frees the cached memory once all writes are done.
  • Save stream_consumer.py.
Configure and start the streaming query

The function alone does nothing until you wire it into a streaming query with checkpointing and a trigger interval.

  • In stream_consumer.py, add the following code below the write_to_sinks function:
# Start the streaming query with foreachBatch, checkpoint, and trigger
query = (
    windowed_counts.writeStream.outputMode("append")
    .foreachBatch(write_to_sinks)
    .option("checkpointLocation", "./checkpoints/analytics")
    .trigger(processingTime="10 seconds")
    .start()
)

print("Streaming query started. Press Ctrl+C to stop.")
query.awaitTermination()

What does this configuration do?

  • outputMode("append") emits only finalized window counts (rows that the watermark has closed).
  • foreachBatch(write_to_sinks) routes each micro-batch through your custom function instead of a built-in sink.
  • checkpointLocation tells Spark to store committed Kafka offsets and query state in ./checkpoints/analytics. On restart, Spark reads this directory to resume exactly where it left off.
  • trigger(processingTime="10 seconds") fires a micro-batch every 10 seconds, giving events time to accumulate for efficient processing.
  • awaitTermination() blocks the main thread so the streaming query keeps running until you press Ctrl+C.

Why foreachBatch instead of format("kafka")?

You could write directly with writeStream.format("kafka") for a single Kafka sink. But foreachBatch gives you a standard batch DataFrame you can write to multiple destinations in one pass.

In the next step, you will add a second sink (local JSON files) inside the same function. This avoids running two separate streaming queries against the same source.

  • Save stream_consumer.py.
  • Make sure producer.py is still running in a separate terminal. If it stopped, start it again by running:
python producer.py
  • In a new terminal, run the streaming consumer by executing:
spark-submit --packages org.apache.spark:spark-sql-kafka-0-10_2.12:3.5.9 stream_consumer.py

Wait about 30 seconds for the watermark window to close and the first micro-batch to finalize. You should see the "Streaming query started" message followed by Spark processing logs.

  • Open another terminal and verify results are landing on the downstream topic by running:
docker exec kafka kafka-console-consumer --bootstrap-server localhost:9092 --topic clickstream-analytics --from-beginning

You should see JSON messages appearing with window_start, window_end, event_type, and count fields.

Not seeing any messages on the analytics topic?

  • The watermark needs at least 2 minutes of event time to pass before windows finalize. Wait a bit longer and check again.
  • Confirm producer.py is still running and sending events to clickstream-events.
  • Check the spark-submit terminal for errors. If you see a connection refused error, verify Kafka is running with docker compose ps.

help me debug why my streaming query is not producing output on the clickstream-analytics topic

Verify checkpoint recovery

Checkpointing is what separates a toy streaming job from a production pipeline. Spark stores committed Kafka offsets in its own checkpoint directory, independently of Kafka consumer groups.

  • In the terminal running spark-submit, press Ctrl+C to stop the streaming query.
  • Confirm the checkpoint directory was created by running:
ls ./checkpoints/analytics

You should see subdirectories like commits, offsets, sources, and state. These files track exactly where Spark left off.

Before you restart, think about what you expect to happen. Will Spark reprocess all the old events from the beginning of the topic, or will it pick up from where it stopped?

  • Restart the streaming query by running:
spark-submit --packages org.apache.spark:spark-sql-kafka-0-10_2.12:3.5.9 stream_consumer.py

Spark resumes from the last committed offset stored in the checkpoint. It does not reprocess events it already handled. You can confirm this by watching the kafka-console-consumer terminal. New messages continue to appear, but you do not see a flood of duplicated old results.

How does checkpoint recovery work?

Spark stores Kafka offsets in its own checkpoint directory, not in Kafka consumer groups. The startingOffsets option is ignored after the first successful checkpoint.

On restart, Spark reads the offsets and commits files to determine the last successfully processed micro-batch. It then resumes from the next offset, providing exactly-once processing semantics for the pipeline.

Seeing reprocessed events after restart?

  • Make sure you are using the same checkpointLocation path (./checkpoints/analytics) on every run. If you accidentally deleted the directory or changed the path, Spark treats it as a fresh start.
  • Verify you are running from the same working directory. The relative path ./checkpoints/analytics resolves relative to where you invoke spark-submit.

help me understand why my Spark streaming query is reprocessing old data after a restart

✔️ Awesome, I've got everything!

Great. Double check you have saved stream_consumer.py and that your pipeline is running with results appearing on the clickstream-analytics topic.

ⓧ I'd like to double check the full code

Your stream_consumer.py should now contain these sections in order. Scroll up and confirm each is present:

  • Imports (from_json, col, window, to_json, struct, lit, types, and os).
  • SparkSession creation with spark.jars.packages config for the Kafka connector.
  • Schema definition for the 5 event fields.
  • raw_stream reading from Kafka with maxOffsetsPerTrigger set to 1000.
  • events DataFrame parsed from JSON with timestamp cast.
  • deduplicated with withWatermark("timestamp", "2 minutes") and dropDuplicatesWithinWatermark(["event_id"]).
  • windowed_counts groupBy with 1-minute tumbling window.
  • write_to_sinks function writing to Kafka via batch write.
  • query using writeStream.outputMode("append").foreachBatch(write_to_sinks) with checkpointLocation and trigger(processingTime="10 seconds").
  • Print statement and query.awaitTermination().

Your pipeline is now writing deduplicated, windowed analytics to a downstream Kafka topic with exactly-once checkpoint recovery. Next up, you will add a second output sink to write results to a local JSON data lake, completing the multi-sink production pattern.

Add Multi-Sink Output with foreachBatch

Your pipeline currently writes windowed aggregation results to a single Kafka topic. Production pipelines rarely write to just one place.

Teams typically need data flowing to a real-time dashboard AND landing in a data lake for batch analytics. The foreachBatch function you already have gives you exactly that power. In this step, you will add a local JSON data lake as a second sink and implement idempotent writes so no batch is ever processed twice.

In this step, get ready to:
  • Add idempotent batch tracking and a JSON file sink to the foreachBatch function.
  • Explore the availableNow trigger mode for batch catch-up processing.
  • Verify both sinks receive deduplicated, windowed analytics data.
Add idempotent batch tracking to the write function

When a streaming job restarts, Spark may re-deliver the same batch to your foreachBatch function. Without a guard, you would write duplicate rows to your data lake.

  • In stream_consumer.py, find the write_to_sinks function.
  • Find this line inside the function:
    batch_df.persist()
  • Directly below batch_df.persist(), add the following idempotent check code:
    # Idempotent check
    tracker_path = "./output/batch_tracker.txt"
    os.makedirs("./output", exist_ok=True)
    processed_batches = set()
    if os.path.exists(tracker_path):
        with open(tracker_path, "r") as f:
            processed_batches = set(int(line.strip()) for line in f if line.strip())

    if batch_id in processed_batches:
        batch_df.unpersist()
        return

What does this code do?

  • tracker_path points to a local text file that logs every batch ID the function has already handled.
  • os.makedirs("./output", exist_ok=True) ensures the output directory exists before any writes happen.
  • If the incoming batch_id already appears in the tracker file, the function exits early. This prevents duplicate writes after a restart.
Add the JSON data lake sink and batch ID logging

Now you will write each micro-batch to a local JSON file store, partitioned by event type. This simulates the data lake leg of the multi-sink pattern.

  • In the same write_to_sinks function, find this existing line near the bottom:
    batch_df.unpersist()
  • Replace that single batch_df.unpersist() line with the following code that writes to JSON, logs the batch ID, and then unpersists:
    # Write to local JSON data lake
    json_output = batch_df.select(
        col("window").getField("start").alias("window_start"),
        col("window").getField("end").alias("window_end"),
        col("event_type"),
        col("count"),
    )
    (
        json_output.write.mode("append")
        .partitionBy("event_type")
        .json("./output/analytics")
    )

    # Track batch ID
    with open(tracker_path, "a") as f:
        f.write(f"{batch_id}\n")

    batch_df.unpersist()

What does this code do?

  • json_output selects the flattened window start/end, event type, and count columns for clean JSON files.
  • .partitionBy("event_type") creates subdirectories like event_type=page_view/ so downstream batch jobs can read only the partitions they need.
  • After both sinks succeed, the batch ID is appended to batch_tracker.txt. On any future re-delivery of that batch, the idempotent check exits early.
  • Save stream_consumer.py.
  • Make sure your producer is still running in a separate terminal. If you stopped it, restart it by running:
python producer.py
  • Before you run the updated pipeline, what do you expect to see in the ./output/ directory after a few micro-batches complete?
  • Delete the old checkpoint so the pipeline starts fresh with the new sink logic by running:
rm -rf ./checkpoints/analytics
  • Start the updated pipeline by running:
spark-submit --packages org.apache.spark:spark-sql-kafka-0-10_2.12:3.5.9 stream_consumer.py

Let it run for about 30 seconds so a few micro-batches fire. You should see Spark log output indicating batches being processed.

  • Stop the stream with Ctrl+C.
  • Verify JSON files were written by running:
ls ./output/analytics/

You should see partition directories like event_type=page_view, event_type=add_to_cart, and event_type=purchase.

  • Verify the batch tracker is recording processed batches by running:
cat ./output/batch_tracker.txt

You should see one or more integer batch IDs, one per line. Each number corresponds to a micro-batch that was successfully written to both sinks.

  • Confirm the Kafka sink is also still receiving data by running:
docker exec kafka kafka-console-consumer --bootstrap-server localhost:9092 --topic clickstream-analytics --from-beginning --max-messages 3

You should see JSON messages containing window_start, window_end, event_type, and count fields.

Not seeing any output files?

  • Make sure you deleted the old checkpoint before running. A stale checkpoint with a different batch function shape can cause silent failures.
  • Check that the producer was actively sending events while the stream was running. Without new events after the watermark passes, no windows finalize.
  • Verify the os import is at the top of stream_consumer.py. The os.makedirs and os.path.exists calls will fail without it.

Still stuck? Help me debug why my foreachBatch function is not writing JSON output files.

Explore the availableNow trigger for batch catch-up

Production teams sometimes need to process a backlog of accumulated events in one shot. The availableNow trigger processes all pending data in multiple micro-batches and then automatically terminates the query.

  • In stream_consumer.py, find this line at the bottom of the file:
    .trigger(processingTime="10 seconds")
  • Replace it with:
    .trigger(availableNow=True)

What is the availableNow trigger?

With trigger(availableNow=True), Spark reads all data that has arrived since the last checkpoint, processes it across possibly multiple micro-batches, and then stops the query automatically. This is the batch-catch-up pattern teams use when they spin up clusters periodically to drain backlogs.

  • Save stream_consumer.py.
  • Let the producer run for about 10 seconds to accumulate a backlog of events, then run:
spark-submit --packages org.apache.spark:spark-sql-kafka-0-10_2.12:3.5.9 stream_consumer.py

This time the job processes all available events in the backlog and then terminates on its own. You do not need to press Ctrl+C. Spark exits once all pending data is consumed.

  • Confirm additional JSON files landed by running:
ls ./output/analytics/event_type=page_view/

You should see new JSON part files from the catch-up run.

  • Now revert the trigger back to the continuous processing mode for production use. In stream_consumer.py, find:
    .trigger(availableNow=True)
  • Replace it with:
    .trigger(processingTime="10 seconds")
  • Save stream_consumer.py.

Why revert to processingTime?

The availableNow trigger is designed for one-off catch-up runs. For a continuously running pipeline that processes events as they arrive, processingTime="10 seconds" keeps the query alive and fires a micro-batch every 10 seconds indefinitely.

✔️ Awesome, I've got everything!

Great. Double check you have saved stream_consumer.py with the processingTime trigger restored.

ⓧ I'd like to double check the full code

Your stream_consumer.py should contain the following sections in order. Scroll up and confirm each is present:

  • Imports (SparkSession, from_json, col, window, to_json, struct, lit, schema types, os).
  • SparkSession creation with spark-sql-kafka-0-10_2.12:3.5.9 package.
  • Schema definition for clickstream events.
  • Kafka readStream with maxOffsetsPerTrigger set to 1000.
  • JSON parsing and timestamp cast.
  • Watermark (2 minutes) and dropDuplicatesWithinWatermark(["event_id"]).
  • Windowed aggregation (1-minute windows grouped by event_type).
  • write_to_sinks function with: empty check, persist, idempotent batch tracker, Kafka write, JSON write partitioned by event_type, batch ID logging, unpersist.
  • writeStream with foreachBatch(write_to_sinks), checkpoint at ./checkpoints/analytics, and trigger(processingTime="10 seconds").

You now have a production-shaped multi-sink pipeline. Kafka powers real-time consumers while JSON files feed batch analytics. The batch tracker ensures no data is written twice, even if Spark re-delivers a batch after a restart.

Secret mission

Stream-Stream Join for Event Enrichment

The most powerful capability that separates Spark from plain Kafka consumers is stream-stream joins. Enrich clickstream events with user profile data from a second stream, using watermarks on both sides for state cleanup. This is how production teams build real-time enrichment without maintaining external lookup tables.

Clean Up Your Resources

Clean Up Your Resources

Decide whether to keep your resources running, pause them to come back later, or delete them entirely. This project runs entirely locally, so there are no ongoing costs.

Resources you used:

  • Docker containers (Kafka broker via docker-compose.yml).
  • Checkpoint directories (./checkpoints/analytics and ./checkpoints/enriched) on local disk.
  • Output files (./output/) on local disk.
  • Python packages installed via pip (pyspark and kafka-python).

Keep everything running

No action needed. Choose this if you're still actively experimenting with your streaming pipeline or plan to extend it further.

  • Your Kafka broker continues running via Docker and uses approximately 500MB of RAM while idle.
  • Checkpoint and output directories remain on disk for future pipeline runs.
  • The pyspark and kafka-python packages remain installed in your Python environment.

Pause - I'll come back to this later

Shut down running containers to free up memory, but keep all files and data intact so you can resume later.

  • Stop any running producer or Spark scripts with Ctrl+C in each terminal.
  • Stop the Docker containers by running:
docker compose stop
  • This preserves all container state and Kafka data. When you're ready to resume, run docker compose start to bring everything back.
  • Your checkpoint directories ensure Spark resumes from where it left off.

Delete - I don't want to use this again

Remove all project resources and start fresh. This is irreversible.

  • Stop any running producer or Spark scripts with Ctrl+C in each terminal.
  • Remove Docker containers and volumes by running:
docker compose down -v
  • Delete the checkpoint and output directories.

macOS

rm -rf ./checkpoints ./output

Windows

Remove-Item -Recurse -Force .\checkpoints, .\output
  • Uninstall the Python packages by running:
pip uninstall -y pyspark kafka-python
  • Delete the project folder itself. Navigate one level up from your project directory, then remove it.

macOS

cd ~/Desktop
rm -rf clickstream-pipeline

Windows

cd ~\Desktop
Remove-Item -Recurse -Force .\clickstream-pipeline

That's a Wrap!

That's a Wrap!

Nice work. You just built a production-shaped real-time analytics pipeline that reads clickstream events from Kafka, deduplicates and aggregates them with PySpark Structured Streaming, and writes enriched results to multiple sinks.

You've learned how to:

  • Build a complete read-process-write streaming pipeline from Kafka through PySpark back to Kafka, with deduplication via dropDuplicatesWithinWatermark and windowed aggregations that correctly handle late-arriving data.
  • Experience the output mode limitation first-hand and resolve it with watermarking, understanding why Spark requires a watermark before it can emit finalized aggregation results in append mode.
  • Configure production concerns including checkpointing for exactly-once recovery, foreachBatch for multi-sink output with idempotent writes, maxOffsetsPerTrigger for backpressure control, and trigger modes for batch catch-up.
  • Secret Mission: Enriched clickstream events using a stream-stream join with state cleanup via watermarks and time constraints on both sides of the join.

Ready to quiz yourself?