Real-Time Sales Streaming Pipeline
Build a low-latency Structured Streaming pipeline that continuously ingests real-time sales transactions, manages event-time watermarking, isolates malformed payloads, computes tumbling and sliding window metrics, and maintains fault-tolerant checkpoint recovery.
Project Brief
Understand the real-time operational requirements and streaming architecture objectives.
Business Scenario
A global omnichannel retail enterprise processes hundreds of thousands of retail transactions every minute across e-commerce portals, mobile apps, and physical point-of-sale terminals. Operations, fraud detection, and executive leadership need real-time operational visibility into sales velocity, high-value orders, payment failures, and store-level transaction spikes.
The existing batch architecture only refreshes dashboards every few hours, leaving store managers and marketing teams blind to midday flash sales anomalies and checkout gateway outages. The engineering team must implement a robust, continuous streaming pipeline that ingests raw sales events, cleanses and deduplicates records in micro-batches, calculates low-latency operational KPIs, and publishes updates to Snowflake and live monitoring dashboards.
Development Objective
Design and develop a production-ready Structured Streaming pipeline in PySpark and Delta Lake. The pipeline must continuously ingest event streams, handle schema evolution and corrupt records, apply watermarking for late-arriving data, deduplicate transactions within streaming windows, compute windowed revenue aggregates, and synchronize curated datasets to Snowflake with fault-tolerant checkpointing.
Expected Outcome
Unlike batch pipelines, streaming systems require explicit state management, watermarks, trigger intervals, and memory safeguards. Your implementation must demonstrate that backpressure, burst traffic, and stream restarts do not corrupt analytical results.
Source System
Inspect streaming payload contracts, event formats, timestamp semantics, and ingestion parameters.
Source Stream Overview
The source system is an active event stream (e.g., Kafka topic, Event Hub, or simulated streaming JSON directory) emitting point-of-sale and online checkout transactions continuously.
Example Streaming JSON Event
{
"event_id": "EVT-STREAM-98210",
"transaction_id": "TX-990124",
"store_id": "STR-402",
"customer_id": "C-10492",
"event_timestamp": "2026-08-30T10:14:32.450Z",
"channel": "POS_TERMINAL",
"currency": "USD",
"items": [
{
"item_id": "SKU-102",
"quantity": 2,
"unit_price": 45.00,
"discount": 5.00
}
],
"gross_amount": 90.00,
"discount_amount": 10.00,
"net_amount": 80.00,
"payment_method": "CREDIT_CARD",
"payment_status": "SUCCESS"
}Event Schema Fields
Known Streaming Source Constraints
Your pipeline must never crash due to dirty input data. Corrupted events must be isolated to bad-records storage while valid transactions continue streaming uninterrupted.
Expected Architecture
Structured Streaming lakehouse architecture: Bronze changelog, Silver validation, and Gold streaming aggregations.
Streaming Flow Architecture
The architecture implements a real-time medallion pipeline utilizing PySpark Structured Streaming and Delta Lake.
Layer Responsibilities
Architectural Expectations
Never use uncontrolled streaming aggregations without watermarking. Unbounded state stores will inevitably exhaust JVM heap memory and crash the Spark cluster under production workloads.
Development Requirements
Implement the functional components required for streaming ingestion, watermarking, curation, and publishing.
Developer Responsibilities
The implementation must address the following 15 engineering requirements across streaming ingestion, data quality, windowing, and target publishing.
Structured Streaming Ingestion
Configure resilient streaming read streams.
Explicit Schema Definition & Parsing
Parse streaming payloads against strict schemas.
Malformed Event Quarantine
Isolate corrupted JSON payloads without breaking stream execution.
Event-Time Watermarking
Handle late-arriving events deterministically.
Streaming Micro-Batch Deduplication
Eliminate duplicate transactions within streaming windows.
Silver Transaction Processing
Standardize and explode transactions into query-ready Silver Delta tables.
Tumbling Window Aggregations (5-Min)
Calculate 5-minute operational KPIs for live store monitoring.
Sliding Window Aggregations (1-Hour)
Compute 1-hour rolling trends with 10-minute slide intervals.
Fault-Tolerant Checkpointing
Guarantee stateful recovery across driver restarts.
Snowflake Target Synchronization
Publish streaming transactions and metrics to Snowflake.
Streaming Operational Telemetry
Provide full observability into streaming latency and throughput.
Backpressure & Rate-Limiting Configuration
Protect Spark clusters from unexpected traffic surges.
High-Value Transaction Flagging
Identify large or anomalous transactions in real time.
Configuration Externalization
Externalize all streaming parameters across environments.
Disaster Recovery & Stream Reset Runbook
Document operational recovery procedures.
You may choose between append and update output modes for streaming aggregations. Justify your selection based on downstream consumer requirements and watermark behavior.
Testing Requirements
Validate stream resilience against burst traffic, corrupt events, duplicates, restarts, and downstream failures.
Required Test Scenarios
Validate the streaming pipeline against the following 12 test scenarios:
Demonstrating state stability and recovery after unexpected cluster termination is the defining test of a production-grade streaming engineer.
Acceptance Criteria
Verify that the streaming pipeline meets all production Definition of Done criteria.
Definition of Done
The implementation is complete when all 10 criteria are met:
Sign-off requires test logs proving that a mid-stream crash resumes cleanly from checkpoint without missing a single transaction.
Developer Deliverables
Submit all streaming modules, checkpoint configurations, test harnesses, and telemetry evidence.
Required Deliverables
The submission must include the following 11 artifacts:
Ensure your mock event generator can be executed locally by another engineer to replay all 12 test scenarios.
Engineering Constraints
Adhere to operational and technical boundaries for enterprise streaming systems.
Required Boundaries
The solution must strictly comply with the following 10 constraints:
Deploying streaming queries without watermarks or using shared checkpoint directories will result in immediate code review failure.
Suggested Project Structure
Recommended repository layout for production streaming pipelines.
Recommended Project Layout
Structure your repository to separate stream ingestion, watermarking, window aggregations, sinks, and tests:
DEV-006-real-time-streaming-pipeline/
│
├── README.md
│
├── config/
│ ├── dev.yaml
│ ├── staging.yaml
│ └── prod.yaml
│
├── src/
│ ├── streaming/
│ │ ├── stream_reader.py # Structured Streaming reader & rate-limits
│ │ ├── schema.py # Explicit PySpark StructType definitions
│ │ └── listener.py # StreamingQueryListener for latency/telemetry
│ ├── processing/
│ │ ├── bronze_raw.py # Append-only raw event logger
│ │ ├── quarantine.py # Malformed payload dead-letter handler
│ │ ├── silver_curation.py # Unnesting, watermarking & deduplication
│ │ └── gold_aggregations.py # 5-min tumbling & 1-hr sliding aggregations
│ └── sinks/
│ ├── delta_sink.py # foreachBatch Delta Lake writer
│ └── snowflake_sink.py # foreachBatch Snowflake staging & merge loader
│
├── tests/
│ ├── mock_event_generator.py # Burst, malformed, duplicate event generator
│ ├── test_stream_ingestion.py
│ ├── test_watermarking.py
│ ├── test_deduplication.py
│ ├── test_window_aggregates.py
│ └── test_checkpoint_recovery.py
│
└── docs/
├── streaming_architecture.md # Architecture & watermark sizing guide
└── recovery_runbook.md # Stream restart & offset rewind playbookModule Responsibilities
Separating streaming ingestion from sink writers allows unit testing window aggregation logic using batch DataFrames before deploying long-running streaming clusters.
Submission Checklist
Final engineering quality checklist before submitting DEV-006.
Final Review Checklist
Verify every checklist item before submitting your streaming pipeline:
Submit DEV-006 only after the streaming ingestion, watermarking, Gold window aggregations, Snowflake foreachBatch sync, checkpoint recovery, and test suites have been verified.