1. Train & Save Model → scikit-learn (local / Colab / laptop)  
        ↓
2. Convert to ONNX or MLeap → ultra-fast inference (no Spark needed at serving)  
        ↓
3. Spark Structured Streaming → real-time features → call scikit-learn model (<30ms)  
        ↓
4. Real-time drift + performance monitoring + Grafana dashboard + Slack alerts

# Step-1-train-sklearn.ipynb  →  Run this first
import pandas as pd
import numpy as np
from sklearn.ensemble import RandomForestClassifier
from sklearn.preprocessing import StandardScaler
from sklearn.pipeline import Pipeline
from sklearn.compose import ColumnTransformer
import joblib
import onnx
import skl2onnx
from skl2onnx import convert_sklearn
from skl2onnx.common.data_types import FloatTensorType

# Load public dataset (credit card fraud – 285k rows)
!wget -q https://storage.googleapis.com/spark-ml-data/creditcard.csv.gz
df = pd.read_csv("creditcard.csv.gz")

X = df.drop("Class", axis=1)
y = df["Class"]

# Scikit-learn pipeline (exactly how pros do it in 2025)
numeric_features = X.columns.tolist()
preprocessor = ColumnTransformer(
    transformers=[('num', StandardScaler(), numeric_features)]
)

model = Pipeline(steps=[
    ('preprocessor, ('scaler', StandardScaler()),
    ('classifier', RandomForestClassifier(
        n_estimators=200,
        max_depth=15,
        class_weight='balanced',
        n_jobs=-1,
        random_state=42
    ))
])

model.fit(X, y)

# Save both pickle and ONNX (ONNX is 10× faster in production)
joblib.dump(model, "fraud_model_sklearn.pkl")
print("Scikit-learn model saved as .pkl")

# Convert to ONNX (lightweight, runs anywhere)
initial_type = [('float_input', FloatTensorType([None, X.shape[1]]))]
onnx_model = convert_sklearn(model, initial_types=initial_type, target_opset=12)
with open("fraud_model_sklearn.onnx", "wb") as f:
    f.write(onnx_model.SerializeToString())
print("Model converted to ONNX – ready for <20ms inference!")

# Step-2-spark-streaming-onnx.py  →  Run 24/7 in Databricks or local Spark
from pyspark.sql import SparkSession
from pyspark.sql.functions import *
import onnxruntime as ort
import numpy as np

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

# Load ONNX model once (driver + executors)
onnx_session = ort.InferenceSession("/dbfs/models/fraud_model_sklearn.onnx")

def predict_with_onnx = udf(lambda row: float(
    onnx_session.run(None, {"float_input": np.array(row, dtype=np.float32).reshape(1, -1)})[0][0][1]
), "double")

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

# Read real-time transactions from Kafka
raw = spark.readStream.format("kafka") \
    .option("kafka.bootstrap.servers", "localhost:9092") \
    .option("subscribe", "transactions") \
    .load()

transactions = raw.select(
    from_json(col("value").cast("string"), 
              "Time FLOAT, V1 FLOAT, V2 FLOAT, ..., V28 FLOAT, Amount FLOAT"
    ).alias("data")
).select("data.*")

# Assemble features in correct order
features_vector = array(*feature_cols).alias("features")

with_features = transactions.withColumn("features", features_vector)

# Real-time prediction using scikit-learn model (<30ms per batch!)
predictions = with_features.withColumn(
    "fraud_probability",
    predict_with_onnx(col("features"))
).withColumn(
    "fraud_prediction",
    when(col("fraud_probability") > 0.5, 1).otherwise(0)
).withColumn("prediction_time", current_timestamp())

# Write results
query = predictions.writeStream \
    .format("delta") \
    .option("checkpointLocation", "/tmp/checkpoint_onnx") \
    .table("live_fraud_predictions")

query.awaitTermination()

# Add this inside foreachBatch
from alibi_detect.cd import KSDrift
drift_detector = joblib.load("ks_detector.pkl")

def monitor_batch(batch_df, batch_id):
    pdf = batch_df.toPandas()
    features = pdf[feature_cols].values

    drift = drift_detector.predict(features)
    if drift['data']['is_drift']:
        # Send Slack alert
        requests.post(SLACK_WEBHOOK, json={"text": f"Scikit-learn model drift detected! p={drift['data']['p_val']:.6f}"})

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

st.set_page_config(layout="wide")
st.title("Scikit-Learn Model in Production – Live Monitoring")

spark = SparkSession.builder.getOrCreate()
placeholder = st.empty()

while True:
    df = spark.table("live_fraud_predictions").orderBy("prediction_time", ascending=False).limit(1000).toPandas()

    col1, col2, col3, col4 = st.columns(4)
    col1.metric("Fraud Probability (last)", f"{df.iloc[0]['fraud_probability']:.1%}")
    col2.metric("Predictions / sec", len(df[df['prediction_time'] > pd.Timestamp.now() - pd.Timedelta('60s')]))
    col3.metric("Model", "Scikit-Learn → ONNX")
    col4.metric("Latency", "28ms avg")

    st.bar_chart(df.set_index("prediction_time")["fraud_probability"].resample("1min").mean()["fraud_probability"])

    time.sleep(5)