Building a clickstream pipeline: lessons from production

Building a clickstream pipeline: lessons from production is an undertaking that will test every layer of your data stack. We've built these systems for high-traffic e-commerce platforms, and every decision—from Kafka partitioning to Delta table layout—shapes whether you can troubleshoot at 2am or sleep soundly. This walkthrough covers a real end-to-end project, showing you not just the happy path, but the gotchas we hit along the way.

Why clickstream pipelines matter

Clickstream data is the heartbeat of user-centric businesses. It tells you not just what users bought, but how they arrived there: which products they viewed, how long they lingered, where they dropped off. Unlike transactional databases that store the final state, clickstreams capture the journey. This abundance of detail creates both opportunity and complexity.

A production clickstream pipeline must handle thousands of events per second, guarantee no data loss, and serve real-time dashboards to product teams. The architecture we'll walk through handles these constraints by separating concerns: raw ingestion, immutable storage, transformation, and aggregation.

Ingest — Events into Kafka

Your frontend fires click events to an HTTP endpoint. Those events need to land in a message queue that can absorb traffic spikes without dropping data. Kafka is the natural fit: it's distributed, durable, and lets you replay events if downstream processing fails.

The first decision is partitioning strategy. A common mistake is keying events by user ID: this balances load across partitions but breaks temporal ordering within a user's session. For clickstream work, we partition by session ID. This keeps all events for a single user session on the same broker, preserving order and making sessionization (grouping events into coherent visits) trivial downstream.

Here's how events land in Kafka from your web service:

import json
from kafka import KafkaProducer
from datetime import datetime

producer = KafkaProducer(
    bootstrap_servers=['kafka-1:9092', 'kafka-2:9092', 'kafka-3:9092'],
    value_serializer=lambda v: json.dumps(v).encode('utf-8'),
    acks='all',
    retries=3,
    compression_type='snappy'
)

event = {
    'session_id': 'sess_12345abcde',
    'user_id': 'user_789',
    'event_type': 'page_view',
    'page': '/products/widget-pro',
    'referrer': '/search?q=widget',
    'timestamp_ms': int(datetime.utcnow().timestamp() * 1000),
    'device_type': 'mobile',
    'country': 'US'
}

producer.send('clickstream-raw', key=event['session_id'].encode(), value=event)
producer.flush()

Key details: We set acks='all' to ensure the broker has replicated the event before confirming. Compression saves network bandwidth at scale. The session ID as key guarantees ordering of events within a session. We use millisecond timestamps because your frontend clocks won't be perfectly synced, and you'll need fine-grained ordering later.

One production lesson: always add a schema registry (like Confluent Schema Registry or AWS Glue Schema Registry). Events evolve. New teams add fields. Without schema governance, you'll have downstream code breaking because a field changed type or disappeared.

Stream — Structured Streaming into bronze Delta tables

Structured Streaming brings Spark's batch semantics to streaming data. You write code that looks like a static DataFrame operation, and Spark handles the streaming execution. We read from Kafka, deserialize the JSON, and write to Delta in append mode, creating an immutable event log.

Delta Lake matters here. It gives you ACID guarantees, schema enforcement, and time travel—critical when you need to replay processing or debug a bug that shipped last week.

from pyspark.sql import SparkSession
from pyspark.sql.functions import col, from_json, schema_of_json
from pyspark.sql.types import StructType, StructField, StringType, LongType

spark = SparkSession.builder \
    .appName("clickstream-ingest") \
    .config("spark.sql.streaming.schemaInference", "false") \
    .getOrCreate()

# Define the event schema
event_schema = StructType([
    StructField("session_id", StringType()),
    StructField("user_id", StringType()),
    StructField("event_type", StringType()),
    StructField("page", StringType()),
    StructField("referrer", StringType()),
    StructField("timestamp_ms", LongType()),
    StructField("device_type", StringType()),
    StructField("country", StringType()),
])

# Read from Kafka
df = spark.readStream \
    .format("kafka") \
    .option("kafka.bootstrap.servers", "kafka-1:9092,kafka-2:9092,kafka-3:9092") \
    .option("subscribe", "clickstream-raw") \
    .option("startingOffsets", "latest") \
    .load()

# Deserialize JSON payload
parsed = df.select(
    from_json(col("value").cast("string"), event_schema).alias("event"),
    col("timestamp").alias("kafka_timestamp")
).select("event.*", "kafka_timestamp")

# Write to Delta (bronze layer)
query = parsed.writeStream \
    .format("delta") \
    .mode("append") \
    .option("checkpointLocation", "/data/checkpoints/clickstream-bronze") \
    .option("mergeSchema", "false") \
    .start("/data/bronze/clickstream_events")

query.awaitTermination()

The checkpoint location is critical. Structured Streaming tracks which offsets it has processed. If the job crashes, it resumes from the last checkpoint, not from the beginning. Without checkpoints, you either replay data (duplicates) or lose data.

We use append mode deliberately: the bronze table is an event log. Nothing is deleted or updated. This immutability is your safety net.

Transform — Sessionize events in silver

Raw events are granular and low-level. A user's session might span hundreds of clicks over 30 minutes. To make events useful, we need to group them into sessions and enrich them with context.

Sessionization is the process of grouping events into logical sessions. A common rule: if more than 30 minutes pass between two events from the same user, start a new session. Spark's window functions and lag function make this tractable.

from pyspark.sql.functions import (
    col, lag, when, sum, max, min, count, 
    window, lit, from_unixtime, unix_timestamp
)
from pyspark.sql.window import Window

# Read bronze events
events = spark.read.format("delta").load("/data/bronze/clickstream_events")

# Define a window ordered by timestamp
w = Window.partitionBy("user_id").orderBy("timestamp_ms")

# Calculate time gap to previous event
events_with_gap = events.withColumn(
    "prev_timestamp_ms",
    lag("timestamp_ms").over(w)
).withColumn(
    "gap_ms",
    col("timestamp_ms") - col("prev_timestamp_ms")
).fillna(0, "gap_ms")

# Mark session boundaries (gap > 30 minutes = 1800000 ms)
session_boundary = events_with_gap.withColumn(
    "is_new_session",
    when(col("gap_ms") > 1800000, 1).otherwise(0)
)

# Assign session IDs by cumulative sum of boundaries
session_assigned = session_boundary.withColumn(
    "session_num",
    sum("is_new_session").over(w)
).withColumn(
    "computed_session_id",
    concat_ws("_", col("user_id"), col("session_num"))
)

# Aggregate to session level
sessions = session_assigned.groupBy("computed_session_id", "user_id").agg(
    min("timestamp_ms").alias("session_start_ms"),
    max("timestamp_ms").alias("session_end_ms"),
    count("*").alias("event_count"),
    collect_list("event_type").alias("event_sequence"),
    max(when(col("event_type") == "purchase", col("page"))).alias("checkout_page"),
    max("device_type").alias("device_type"),
    max("country").alias("country")
).withColumn(
    "session_duration_sec",
    (col("session_end_ms") - col("session_start_ms")) / 1000
)

# Write to silver (deduplicated, enriched)
sessions.write \
    .mode("overwrite") \
    .format("delta") \
    .partitionBy("user_id") \
    .save("/data/silver/sessions")

Partitioning by user ID at the silver layer makes downstream queries faster: product teams often filter by user cohorts. The trade-off is that writing is slightly slower, but the query speedup typically justifies it.

One gotcha: if you rerun this transformation on overlapping data, you'll create duplicate sessions. Use a unique session_id and write with mode("ignore") or merge logic, or ensure your job only processes new data (via checkpoint tracking or explicit date ranges).

Serve — Daily aggregates in gold for dashboards

Product teams don't want raw sessions. They want summaries: how many users came today, conversion rate by traffic source, average session duration by device type. These are the gold tables that power dashboards and metrics.

Gold tables are typically aggregated, small enough to query in milliseconds, and updated on a schedule (daily, hourly). We compute them from silver and write to Parquet or Delta, then expose via a SQL engine (Presto, Athena, or a BI tool).

from pyspark.sql.functions import (
    col, date_trunc, count, avg, sum, approx_count_distinct,
    when, hour, date_format
)
from datetime import datetime, timedelta

# Read silver sessions (compute for yesterday to avoid late-arriving data)
run_date = (datetime.utcnow() - timedelta(days=1)).strftime("%Y-%m-%d")
sessions = spark.read.format("delta").load("/data/silver/sessions") \
    .filter(date_format(col("session_start_ms") / 1000, "yyyy-MM-dd") == run_date)

# Aggregate by day and device type
daily_by_device = sessions.groupBy(
    date_format(col("session_start_ms") / 1000, "yyyy-MM-dd").alias("date"),
    col("device_type")
).agg(
    approx_count_distinct("user_id").alias("unique_users"),
    count("*").alias("session_count"),
    avg("session_duration_sec").alias("avg_session_duration_sec"),
    sum(when(col("checkout_page").isNotNull(), 1).otherwise(0)).alias("purchased_sessions")
).withColumn(
    "conversion_rate",
    col("purchased_sessions") / col("session_count")
)

# Aggregate by day and country
daily_by_country = sessions.groupBy(
    date_format(col("session_start_ms") / 1000, "yyyy-MM-dd").alias("date"),
    col("country")
).agg(
    approx_count_distinct("user_id").alias("unique_users"),
    count("*").alias("session_count"),
    avg("session_duration_sec").alias("avg_session_duration_sec")
)

# Write to gold (overwrite yesterday's partition)
daily_by_device.write \
    .mode("overwrite") \
    .format("parquet") \
    .partitionBy("date") \
    .save("/data/gold/daily_device_metrics")

daily_by_country.write \
    .mode("overwrite") \
    .format("parquet") \
    .partitionBy("date") \
    .save("/data/gold/daily_country_metrics")

We backdate aggregates to yesterday because real data is messy. Events arrive late. A session might close after midnight. By computing yesterday's snapshot today, you give late-arriving events a chance to land. Your dashboard shows accurate historical data.

Gold tables are small (often gigabytes) so they can be queried directly by Tableau, Looker, or loaded into a fast SQL database for dashboards. Some teams push gold tables into a data warehouse like Snowflake for cross-pipeline joins.

Common mistakes

• Not handling late data: Clock skew and network delays mean events arrive out of order or late. Don't assume a session is final after 30 minutes; use watermarks in Structured Streaming to tolerate lateness.
• Ignoring schema evolution: Your product team adds a new field to the event payload. Without a schema registry, your Spark job crashes. Use Avro or Protobuf with a registry, or at minimum validate schemas.
• Partition explosion in silver: Partitioning by too many columns (user_id + device_type + country) creates thousands of tiny files. This slows reads and clusters. Partition by the highest cardinality, most-filtered column only.
• Recomputing gold tables from scratch daily: If you have a year of silver data, recomputing all gold aggregates each day is wasteful. Use incremental merge: only recompute the last 2 days (to handle late arrivals) and merge into historical gold tables.
• No data quality checks: A schema change upstream breaks downstream code silently. Add assertions: if event_count for a day is 0, alert. If conversion_rate exceeds 50%, investigate.

Key takeaways

• Kafka with session-ID partitioning preserves event ordering and simplifies sessionization logic downstream.
• Structured Streaming into Delta bronze tables creates an immutable, replayable event log that survives job failures.
• Silver-layer sessionization uses window functions to group events logically, enriching raw data for business logic.
• Gold tables are pre-aggregated, small, and optimized for dashboard queries; compute them incrementally and partition by date to handle late arrivals.
• Backdate aggregates (compute yesterday today) to give late-arriving events a window to land, improving data accuracy.
• Partition strategically: too many partitions create small files and query bloat; too few create hotspots. Prefer date and high-cardinality filter columns.
• Add schema governance (registry), data quality checks (assertions), and checkpoints (streaming state) from day one. These aren't nice-to-haves when you're woken at midnight by a data outage.