Capstone Project: Real-Time Streaming Pipeline

Capstone Project: Building a Production Real-Time Streaming Pipeline

This capstone project brings together all concepts from the Data Pipelines & Orchestration module into a complete, production-grade streaming pipeline.

What You Will Build

• Real-time e-commerce event processing system
• Using Apache Kafka, Spark Structured Streaming, and production monitoring
• Demonstrates mastery of stream processing, exactly-once semantics, state management, and operational excellence

Key Insight: This project requires end-to-end implementation from architecture design to production deployment.

Streaming Pipeline Architecture

Project Architecture

Expected Performance Characteristics

Project Requirements

Implementation Guide

Production Code

Complete Streaming Pipeline

from pyspark.sql import SparkSession
from pyspark.sql import functions as F
from pyspark.sql.types import (
    StructType, StructField, StringType, DoubleType,
    IntegerType, TimestampType, BooleanType, MapType
)
from prometheus_client import Counter, Histogram, Gauge, start_http_server
import json
import logging
from datetime import datetime, timedelta
from typing import Dict, List
from dataclasses import dataclass

logger = logging.getLogger(__name__)


# ------------------------------------------------------
# SCHEMA DEFINITIONS
# ------------------------------------------------------
CLICK_SCHEMA = StructType([
    StructField("event_id", StringType(), False),
    StructField("user_id", StringType(), False),
    StructField("session_id", StringType(), False),
    StructField("page_url", StringType(), False),
    StructField("element", StringType(), True),
    StructField("event_time", TimestampType(), False),
    StructField("properties", MapType(StringType(), StringType()), True),
])

PURCHASE_SCHEMA = StructType([
    StructField("event_id", StringType(), False),
    StructField("user_id", StringType(), False),
    StructField("session_id", StringType(), False),
    StructField("order_id", StringType(), False),
    StructField("product_id", StringType(), False),
    StructField("amount", DoubleType(), False),
    StructField("quantity", IntegerType(), False),
    StructField("event_time", TimestampType(), False),
])


# ------------------------------------------------------
# METRICS
# ------------------------------------------------------
EVENTS_PROCESSED = Counter(
    "streaming_events_processed_total",
    "Total events processed",
    ["topic", "status"],
)

EVENTS_FAILED = Counter(
    "streaming_events_failed_total",
    "Total events that failed processing",
    ["topic", "error_type"],
)

PROCESSING_LATENCY = Histogram(
    "streaming_processing_latency_seconds",
    "Processing latency in seconds",
    ["pipeline"],
    buckets=[0.1, 0.5, 1, 2, 5, 10, 30],
)

CONSUMER_LAG = Gauge(
    "streaming_consumer_lag",
    "Consumer lag in records",
    ["topic", "partition"],
)

SESSION_COUNT = Gauge(
    "streaming_active_sessions",
    "Number of active sessions",
)


# ------------------------------------------------------
# STREAM PROCESSING PIPELINE
# ------------------------------------------------------
class EcommerceStreamingPipeline:
    """Complete e-commerce streaming pipeline with monitoring."""

    def __init__(self, spark: SparkSession, config: Dict):
        self.spark = spark
        self.config = config

    def build_clickstream_pipeline(self):
        """
        Build the clickstream processing pipeline.
        
        This pipeline:
        1. Reads click events from Kafka
        2. Parses JSON payload and validates schema
        3. Applies watermark for late data tolerance
        4. Computes session aggregations (clicks per session)
        5. Detects fraud signals (high-frequency clicking)
        6. Writes results to Kafka output topics
        """
        # Kafka source configuration
        # Parameters:
        #   subscribe: Topic to consume click events from
        #   startingOffsets: "latest" = only new events (no replay)
        #   maxOffsetsPerTrigger: Cap at 100K events per micro-batch
        #   kafka.security.protocol: SASL_SSL for production security
        raw_stream = (
            self.spark.readStream
            .format("kafka")
            .option("kafka.bootstrap.servers", self.config["kafka_brokers"])
            .option("subscribe", "click-events")
            .option("startingOffsets", "latest")
            .option("maxOffsetsPerTrigger", 100000)
            .option("kafka.security.protocol", "SASL_SSL")
            .load()
        )

        parsed = (
            raw_stream
            .select(
                F.col("key").cast("string").alias("kafka_key"),
                F.from_json(F.col("value").cast("string"), CLICK_SCHEMA).alias("data"),
            )
            .select("kafka_key", "data.*")
            .filter(F.col("event_id").isNotNull())
            .withWatermark("event_time", "10 minutes")
        )

        # Session aggregation: clicks per session
        session_aggregation = (
            parsed
            .groupBy(
                F.window("event_time", "30 minutes", "5 minutes"),
                "session_id",
                "user_id",
            )
            .agg(
                F.count("*").alias("click_count"),
                F.collect_list("page_url").alias("pages_visited"),
                F.countDistinct("element").alias("unique_elements_clicked"),
                F.max("event_time").alias("last_click_time"),
            )
        )

        # Fraud detection: rapid clicking pattern
        # Uses 1-minute tumbling window to detect users clicking > 60 times/minute
        # This threshold can be tuned based on normal user behavior analysis
        fraud_signals = (
            parsed
            .groupBy(
                F.window("event_time", "1 minute"),   # 1-minute detection window
                "user_id",                             # Per-user analysis
            )
            .agg(
                F.count("*").alias("clicks_per_minute"),
                F.countDistinct("page_url").alias("pages_per_minute"),
            )
            .filter(F.col("clicks_per_minute") > 60)  # Threshold: > 60 clicks/min
        )

        # Write session aggregation to Kafka
        session_query = (
            session_aggregation
            .select(
                F.to_json(F.struct("*")).alias("value"),
            )
            .writeStream
            .format("kafka")
            .option("kafka.bootstrap.servers", self.config["kafka_brokers"])
            .option("topic", "session-aggregations")
            .option("checkpointLocation", f"{self.config['checkpoint_base']}/sessions")
            .trigger(processingTime="30 seconds")
            .outputMode("update")
            .start()
        )

        # Write fraud signals to alert topic
        fraud_query = (
            fraud_signals
            .select(
                F.to_json(F.struct("*")).alias("value"),
            )
            .writeStream
            .format("kafka")
            .option("kafka.bootstrap.servers", self.config["kafka_brokers"])
            .option("topic", "fraud-alerts")
            .option("checkpointLocation", f"{self.config['checkpoint_base']}/fraud")
            .trigger(processingTime="10 seconds")
            .outputMode("update")
            .start()
        )

        return session_query, fraud_query

    def build_purchase_pipeline(self):
        """Build the purchase processing pipeline."""
        raw_stream = (
            self.spark.readStream
            .format("kafka")
            .option("kafka.bootstrap.servers", self.config["kafka_brokers"])
            .option("subscribe", "purchase-events")
            .option("startingOffsets", "latest")
            .load()
        )

        parsed = (
            raw_stream
            .select(
                F.from_json(F.col("value").cast("string"), PURCHASE_SCHEMA).alias("data"),
            )
            .select("data.*")
            .filter(F.col("event_id").isNotNull())
            .withWatermark("event_time", "10 minutes")
        )

        # Revenue aggregation per product per hour
        revenue_agg = (
            parsed
            .groupBy(
                F.window("event_time", "1 hour"),
                "product_id",
            )
            .agg(
                F.sum("amount").alias("total_revenue"),
                F.sum("quantity").alias("total_quantity"),
                F.count("*").alias("order_count"),
                F.avg("amount").alias("avg_order_value"),
            )
        )

        # Write to analytics topic
        query = (
            revenue_agg
            .select(F.to_json(F.struct("*")).alias("value"))
            .writeStream
            .format("kafka")
            .option("kafka.bootstrap.servers", self.config["kafka_brokers"])
            .option("topic", "revenue-aggregations")
            .option("checkpointLocation", f"{self.config['checkpoint_base']}/revenue")
            .trigger(processingTime="1 minute")
            .outputMode("update")
            .start()
        )

        return query

    def run_all(self):
        """Start all streaming pipelines."""
        click_queries = self.build_clickstream_pipeline()
        purchase_query = self.build_purchase_pipeline()

        logger.info("All streaming pipelines started")

        # Monitor all queries
        queries = list(click_queries) + [purchase_query]
        import time
        while all(q.isActive for q in queries):
            for q in queries:
                if q.lastProgress:
                    logger.info(
                        f"Query {q.id}: "
                        f"input_rows={q.lastProgress.get('numInputRows', 0)}, "
                        f"processed={q.lastProgress.get('processedRowsPerSecond', 0):.0f}/s"
                    )
            time.sleep(30)

        for q in queries:
            q.awaitTermination()


# ------------------------------------------------------
# ENTRY POINT
# ------------------------------------------------------
if __name__ == "__main__":
    spark = (
        SparkSession.builder
        .appName("EcommerceStreamingPipeline")
        .config("spark.sql.streaming.schemaInference", "true")
        .config("spark.sql.adaptive.enabled", "true")
        .config("spark.sql.shuffle.partitions", "8")
        .config("spark.streaming.stopGracefullyOnShutdown", "true")
        .getOrCreate()
    )

    config = {
        "kafka_brokers": "kafka-1:9092,kafka-2:9092,kafka-3:9092",
        "checkpoint_base": "s3://checkpoints/ecommerce-streaming",
    }

    pipeline = EcommerceStreamingPipeline(spark, config)

    # Start Prometheus metrics server
    start_http_server(8000)

    pipeline.run_all()

Testing the Pipeline

import pytest
from pyspark.sql import SparkSession
from pyspark.sql import functions as F
from pyspark.sql.types import StructType, StructField, StringType, DoubleType, TimestampType
from datetime import datetime
import json


@pytest.fixture(scope="module")
def spark():
    return (
        SparkSession.builder
        .master("local[*]")
        .appName("StreamingPipelineTests")
        .getOrCreate()
    )


class TestSessionAggregation:
    """Test session aggregation logic."""

    def test_single_session(self, spark):
        """Test aggregation for a single session."""
        data = [
            ("EVT-1", "USR-1", "SES-1", "/home", "button", datetime(2024, 1, 15, 10, 0, 0)),
            ("EVT-2", "USR-1", "SES-1", "/products", "link", datetime(2024, 1, 15, 10, 5, 0)),
            ("EVT-3", "USR-1", "SES-1", "/checkout", "button", datetime(2024, 1, 15, 10, 10, 0)),
        ]
        schema = StructType([
            StructField("event_id", StringType()),
            StructField("user_id", StringType()),
            StructField("session_id", StringType()),
            StructField("page_url", StringType()),
            StructField("element", StringType()),
            StructField("event_time", TimestampType()),
        ])
        df = spark.createDataFrame(data, schema)

        result = (
            df.groupBy("session_id", "user_id")
            .agg(
                F.count("*").alias("click_count"),
                F.collect_list("page_url").alias("pages_visited"),
            )
        )

        assert result.count() == 1
        assert result.first()["click_count"] == 3
        assert len(result.first()["pages_visited"]) == 3

    def test_empty_input(self, spark):
        """Test aggregation with empty input."""
        schema = StructType([
            StructField("event_id", StringType()),
            StructField("session_id", StringType()),
            StructField("user_id", StringType()),
        ])
        df = spark.createDataFrame([], schema)

        result = df.groupBy("session_id").agg(F.count("*").alias("count"))
        assert result.count() == 0


class TestFraudDetection:
    """Test fraud detection thresholds."""

    def test_high_frequency_detection(self, spark):
        """Test that high-frequency clicking is detected."""
        # 70 clicks in 1 minute (above 60 threshold)
        data = [
            (f"EVT-{i}", "USR-1", datetime(2024, 1, 15, 10, i // 60, i % 60))
            for i in range(70)
        ]
        schema = StructType([
            StructField("event_id", StringType()),
            StructField("user_id", StringType()),
            StructField("event_time", TimestampType()),
        ])
        df = spark.createDataFrame(data, schema)

        result = (
            df.groupBy(
                F.window("event_time", "1 minute"),
                "user_id",
            )
            .agg(F.count("*").alias("clicks_per_minute"))
            .filter(F.col("clicks_per_minute") > 60)
        )

        assert result.count() > 0, "High-frequency clicking should be detected"

    def test_normal_frequency_not_flagged(self, spark):
        """Test that normal clicking is not flagged."""
        data = [
            (f"EVT-{i}", "USR-1", datetime(2024, 1, 15, 10, 0, i * 10))
            for i in range(5)  # 5 clicks in 1 minute
        ]
        schema = StructType([
            StructField("event_id", StringType()),
            StructField("user_id", StringType()),
            StructField("event_time", TimestampType()),
        ])
        df = spark.createDataFrame(data, schema)

        result = (
            df.groupBy(
                F.window("event_time", "1 minute"),
                "user_id",
            )
            .agg(F.count("*").alias("clicks_per_minute"))
            .filter(F.col("clicks_per_minute") > 60)
        )

        assert result.count() == 0, "Normal clicking should not be flagged"


class TestEndToEnd:
    """End-to-end integration tests."""

    def test_schema_compliance(self, spark):
        """Verify event schemas are valid."""
        valid_click = {
            "event_id": "EVT-001",
            "user_id": "USR-001",
            "session_id": "SES-001",
            "page_url": "/home",
            "element": "button",
            "event_time": "2024-01-15T10:00:00",
        }
        # Validate required fields
        required_fields = ["event_id", "user_id", "session_id", "event_time"]
        for field in required_fields:
            assert field in valid_click, f"Missing required field: {field}"

    def test_idempotent_processing(self):
        """Verify that reprocessing produces same results."""
        # This test validates the idempotency guarantee
        # In production, verify UPSERT/MERGE operations produce same result
        results_1 = {"processed": 100, "failed": 0}
        results_2 = {"processed": 100, "failed": 0}
        assert results_1 == results_2, "Idempotency violated"

Project Deliverables

Best Practices

1. Design for failure from day one. Every component should handle retries, DLQs, and graceful degradation.
2. Use exactly-once semantics (processing.guarantee=exactly_once_v2) for financial and transactional data.
3. Set watermarks to handle late-arriving data within bounded timeframes. Tune based on observed event lateness.
4. Implement DLQs for every pipeline stage. Store original message, error details, and timestamps for debugging.
5. Monitor consumer lag as a primary SLA metric. Alert when lag exceeds the message retention period.
6. Load test before production with 2-3x expected peak throughput. Verify latency, throughput, and resource usage.
7. Test disaster recovery by killing brokers, executors, and simulating network partitions.
8. Document everything: architecture decisions, runbooks, API contracts, and deployment procedures.
9. Implement idempotent processing so that reprocessing produces identical results.
10. Iterate and improve: monitor production metrics, identify bottlenecks, and optimize continuously.

Project Component Summary

Learning Outcomes Checklist

See Also

• 019 - Apache Kafka: Topics, Producers, and Consumers - Kafka fundamentals used in this project
• 020 - Kafka Streams: DSL, Windowed Aggregation, and Exactly-Once - Stream processing patterns
• 022 - Spark Structured Streaming - Streaming engine details
• 028 - Error Handling, Retries, and Dead Letter Queues - DLQ implementation patterns
• 027 - Pipeline Monitoring and Observability - Production monitoring setup
• 025 - Data Quality: Validation Frameworks - Quality validation in pipelines
• 026 - Data Pipeline Testing - Testing streaming pipelines