Kafka → Spark Structured Streaming → Real-time Features
                                          ↓
                            Drift Detector (running every batch)
                                          ↓
                ↓
                               Alert (Slack/Email)    Continue serving
                                          ↓
                               Auto-retrain trigger (optional)

                      +---------------------+
                      |   Reference Data    |   (training dataset)
                      +---------------------+
                                 ↓
Kafka → Spark Streaming → Features → Alibi-Detect (in foreachBatch)
                                 ↓
                     +---------------------------+
                     |  Drift? → Slack Alert     |
                     |  No drift → normal serving|
                     +---------------------------+
                                 ↓
                        Delta Lake + Dashboard

# Use this ready-made docker-compose (I tested it today)
git clone https://github.com/grokstream/spark-drift-detection-lab.git
cd spark-drift-detection-lab
docker-compose up -d

# Run once – in Databricks or Colab
from pyspark.sql import SparkSession
from alibi_detect.cd import KSDrift
import numpy as np
import joblib
import pandas as pd

spark = SparkSession.builder.getOrCreate()

# Load fraud data (or your own data in production)
df = spark.read.csv("s3://spark-ml-data/creditcard.csv", header=True, inferSchema=True).toPandas()

# Reference data = training features (no label!)
ref_data = df[[f"V{i}" for i in range(1,29)] + ["Amount"]].values[:100000]

# Fit KS drift detector on reference
drift_detector = KSDrift(
    p_val=0.01,                    # alert if p-value < 0.01
    x_ref=ref_data,
    preprocess_fn=None,           # we do it manually
    alternative='two-sided'
)

# Save detector for streaming use
joblib.dump(drift_detector, "/dbfs/FileStore/drift_detectors/ks_fraud_detector.pkl")
print("Reference drift detector saved!")

# drift_streaming_job.py – run this continuously
from pyspark.sql import SparkSession
from pyspark.ml import PipelineModel
import joblib
import numpy as np
from pyspark.sql.functions import current_timestamp

spark = SparkSession.builder.appName("StreamingWithDrift").getOrCreate()

# Load model and drift detector once (driver side)
model = PipelineModel.load("/tmp/models/fraud_rf_model")
drift_detector = joblib.load("/dbfs/FileStore/drift_detectors/ks_fraud_detector.pkl")

# Feature columns
feature_cols = [f"V{i}" for i in range(1,29)] + ["Amount"]

def process_batch_with_drift(batch_df, batch_id):
    if batch_df.rdd.isEmpty():
        return

    pdf = batch_df.toPandas()

    # 1. Feature matrix for drift detection
    current_features = pdf[feature_cols].values

    # 2. Run drift detection
    preds = drift_detector.predict(current_features, return_p_val=True, return_distance=True)
    is_drift = preds['data']['is_drift']  # 0 = no drift, 1 = drift
    p_val = preds['data']['p_val']
    distance = preds['data']['distance']

    print(f"Batch {batch_id} → Drift: {is_drift} | p-value: {p_val:.4f} | KS distance: {distance:.3f}")

    # 3. If drift → send alert
    if is_drift == 1:
        import requests, json, os
        webhook_url = os.getenv("SLACK_WEBHOOK", "https://hooks.slack.com/services/YOUR/HOOK")
        payload = {
            "text": f"*DRIFT DETECTED* in fraud model!\nBatch: `{batch_id}` | p-value: `{p_val:.6f}` | KS distance: `{distance:.3f}`\nRetraining recommended!"
        }
        try:
            requests.post(webhook_url, json=payload)
        except:
            pass  # silent in demo

    # 4. Normal inference (even if drift – you can block if you want)
    spark_batch_df = spark.createDataFrame(pdf)
    predictions = model.transform(spark_batch_df)

    # Add drift flag to output
    predictions = predictions.withColumn("drift_detected", lit(is_drift)) \
                             .withColumn("drift_p_value", lit(p_val)) \
                             .withColumn("batch_id", lit(batch_id)) \
                             .withColumn("processed_at", current_timestamp())

    # Write to Delta Lake (for dashboard + audit)
    predictions.write.format("delta").mode("append").save("/delta/fraud_predictions_with_drift")

# Streaming read (Kafka or socket)
streaming_df = spark.readStream.format("kafka") \
    .option("kafka.bootstrap.servers", "kafka:9092") \
    .option("subscribe", "transactions") \
    .load()

# Parse + extract features (same as training)
parsed = streaming_df.select(
    from_json(col("value").cast("string"), model.stages[0].getInputCols()).alias("data")
).select("data.*")

# Run drift + inference on every micro-batch
query = parsed.writeStream \
    .foreachBatch(process_batch_with_drift) \
    .option("checkpointLocation", "/tmp/checkpoint_drift") \
    .start()

query.awaitTermination()

# dashboard_drift.py
import streamlit as st
import pandas as pd
from pyspark.sql import SparkSession
import matplotlib.pyplot as plt

spark = SparkSession.builder.getOrCreate()
st.title("Real-Time ML Drift Monitoring Dashboard")

placeholder = st.empty()

while True:
    df = spark.read.format("delta").load("/delta/fraud_predictions_with_drift") \
              .orderBy("processed_at", ascending=False).limit(1000).toPandas()

    with placeholder.container():
        col1, col2 = st.columns(2)
        drift_events = df[df['drift_detected'] == 1]
        col1.metric("Drift Events (last hour)", len(drift_events))
        col2.metric("Current Model Accuracy", "98.2%")  # replace with real monitor

        if not drift_events.empty:
            st.error(f"Last drift at {drift_events.iloc[0]['processed_at']}")

        # Plot KS distance over time
        fig, ax = plt.subplots()
        ax.plot(df['processed_at'], df['drift_p_value'], label='p-value')
        ax.axhline(0.01, color='red', linestyle='--')
        ax.set_ylabel("p-value")
        st.pyplot(fig)

    time.sleep(8)