Real-time Data Source
         ↓
    Apache Kafka (or socket)
         ↓
Structured Streaming (PySpark / Scala)
         ↓
→ Enrich + Windowed Aggregations + Watermarking
         ↓
→ Delta Lake (ACID table) + PostgreSQL (for dashboard)
         ↓
   Live Dashboard (Streamlit / Grafana)

# Run this in Colab or Databricks notebook
from pyspark.sql import SparkSession
from pyspark.sql.functions import *
from pyspark.sql.types import *

spark = SparkSession.builder \
    .appName("SparkStreamingTutorial") \
    .config("spark.sql.shuffle.partitions", "4") \
    .config("spark.sql.streaming.schemaInference", "true") \
    .getOrCreate()

# Step 1: Simulate real-time data → open terminal and run:
# nc -lk 9999
# Then type lines like:
# {"user":"Alice","action":"click","ts":"2025-11-30T10:00:05Z"}
# {"user":"Bob","action":"purchase","ts":"2025-11-30T10:02:30Z"}

# Step 2: Read streaming data
lines = spark \
    .readStream \
    .format("socket") \
    .option("host", "localhost") \
    .option("port", 9999) \
    .load()

# Parse JSON
schema = StructType([
    StructField("user", StringType()),
    StructField("action", StringType()),
    StructField("ts", TimestampType())
])

events = lines.select(from_json(col("value").cast("string"), schema).alias("data")).select("data.*")

# Step 3: Start streaming query to console
query = events.writeStream \
    .format("console") \
    .outputMode("append") \
    .start()

query.awaitTermination()

# Use free Conduktor Playground or run locally with Docker
docker run -d --name zookeeper -p 2181:2181 confluentinc/cp-zookeeper:latest
docker run -d --name kafka -p 9092:9092 --link zookeeper confluentinc/cp-kafka:latest

# producer.py
from kafka import KafkaProducer
import json, time, random
from datetime import datetime

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

users = ["Alice","Bob","Charlie","Diana"]
actions = ["view","click","add_to_cart","purchase"]

while True:
    event = {
        "user_id": random.choice(users),
        "action": random.choice(actions),
        "product": f"product_{random.randint(100,999)",
        "price": round(random.uniform(10, 1000), 2),
        "event_time": datetime.utcnow().isoformat() + "Z"
    }
    producer.send("ecommerce-events", value=event)
    print("Sent:", event)
    time.sleep(0.5)

# Full production-grade streaming job
from pyspark.sql.functions import *
from pyspark.sql.types import *

# Define schema (always do this in production!)
schema = StructType([
    StructField("user_id", StringType()),
    StructField("action", StringType()),
    StructField("product", StringType()),
    StructField("price", DoubleType()),
    StructField("event_time", TimestampType())
])

# Read from Kafka
kafka_df = spark \
    .readStream \
    .format("kafka") \
    .option("kafka.bootstrap.servers", "localhost:9092") \
    .option("subscribe", "ecommerce-events") \
    .option("startingOffsets", "earliest") \
    .load()

# Parse value from Kafka
events = kafka_df \
    .select(from_json(col("value").cast("string"), schema).alias("data")) \
    .select("data.*")

# Add watermark and windowed aggregation (event-time!)
windowed_counts = events \
    .withWatermark("event_time", "10 minutes") \
    .groupBy(
        window(col("event_time"), "5 minutes", "1 minute"),  # tumbling window
        col("action")
    ) \
    .agg(
        count("*").alias("count"),
        sum("price").alias("revenue")
    ) \
    .select(
        col("window.start"),
        col("window.end"),
        col("action"),
        col("count"),
        col("revenue")
    )

# Write to Delta Lake (supports exactly-once + schema evolution)
query = windowed_counts \
    .writeStream \
    .format("delta") \
    .outputMode("append") \
    .option("checkpointLocation", "/tmp/checkpoint_ecommerce") \
    .partitionBy("start") \
    .table("ecommerce_5min_summary")

query.awaitTermination()

# dashboard.py
import streamlit as st
import pandas as pd
from pyspark.sql import SparkSession

spark = SparkSession.builder.getOrCreate()

st.title("Live E-commerce Dashboard")

# Auto-refresh every 5 seconds
placeholder = st.empty()

while True:
    df = spark.sql("SELECT * FROM ecommerce_5min_summary ORDER BY start DESC LIMIT 20")
    pdf = df.toPandas()

    with placeholder.container():
        st.write("Real-time 5-minute Revenue by Action")
        st.bar_chart(pdf.pivot(index="start", columns="action", values="revenue").fillna(0))
        st.dataframe(pdf)

    time.sleep(5)

.withWatermark("event_time", "30 minutes")  # drop data older than 30 min late

def upsert_to_postgres(microBatchDF, batchId):
    microBatchDF.createOrReplaceTempView("batch")
    spark.sql("""
        MERGE INTO postgres_dashboard.sales_summary t
        USING batch s
        ON t.window_start = s.start AND t.action = s.action
        WHEN MATCHED THEN UPDATE SET *
        WHEN NOT MATCHED THEN INSERT *
    """)

query = windowed_counts.writeStream \
    .foreachBatch(upsert_to_postgres) \
    .start()

# Static user dimension
users_df = spark.read.format("delta").load("/delta/users")

# Streaming join
enriched = events.join(
    users_df,
    events.user_id == users_df.id,
    "left"
)