Real-time Insurance Claims Streaming Pipeline

import json
import random
import time
import uuid
from datetime import datetime, timezone
from kafka import KafkaProducer

KAFKA_BOOTSTRAP = "localhost:9092"
TOPICS = {
    "purchase":   "claims.submissions",
    "submission": "claims.submissions",
    "decision":   "claims.decisions",
}
PRODUCTS   = ["life_basic", "hospital_cash", "accident", "funeral"]
MARKETS    = ["KE", "UG", "TZ", "GH", "NG"]
DECISIONS  = ["approved", "rejected", "pending_review"]
FRAUD_RATE = 0.003  # 0.3% fraud injection rate

producer = KafkaProducer(
    bootstrap_servers=KAFKA_BOOTSTRAP,
    value_serializer=lambda v: json.dumps(v).encode("utf-8"),
    key_serializer=lambda k: k.encode("utf-8"),
)


def make_policy_event() -> dict:
    policy_id  = f"POL-{uuid.uuid4().hex[:10].upper()}"
    customer_id = f"CUST-{random.randint(10000, 99999)}"
    return {
        "event_id":     str(uuid.uuid4()),
        "event_type":   "policy_purchase",
        "policy_id":    policy_id,
        "customer_id":  customer_id,
        "product":      random.choice(PRODUCTS),
        "market":       random.choice(MARKETS),
        "premium_kes":  round(random.uniform(50, 500), 2),
        "is_fraud":     random.random() < FRAUD_RATE,
        "timestamp":    datetime.now(timezone.utc).isoformat(),
    }


def make_claim_event(policy_id: str, customer_id: str) -> dict:
    is_fraud = random.random() < FRAUD_RATE
    return {
        "event_id":     str(uuid.uuid4()),
        "event_type":   "claim_submission",
        "claim_id":     f"CLM-{uuid.uuid4().hex[:10].upper()}",
        "policy_id":    policy_id,
        "customer_id":  customer_id,
        "claim_amount": round(random.uniform(500, 50000), 2),
        "is_fraud":     is_fraud,
        "fraud_reason": "rapid_successive_claim" if is_fraud else None,
        "timestamp":    datetime.now(timezone.utc).isoformat(),
    }


def make_decision_event(claim_id: str) -> dict:
    return {
        "event_id":   str(uuid.uuid4()),
        "event_type": "claim_decision",
        "claim_id":   claim_id,
        "decision":   random.choice(DECISIONS),
        "decision_by": "auto_rules_engine",
        "payout_kes": round(random.uniform(500, 50000), 2),
        "timestamp":  datetime.now(timezone.utc).isoformat(),
    }


def run(events_per_second: int = 50):
    print(f"Producing {events_per_second} events/sec...")
    active_policies = []

    while True:
        for _ in range(events_per_second):
            # Policy purchase
            policy = make_policy_event()
            active_policies.append(policy)
            producer.send(
                TOPICS["purchase"],
                key=policy["policy_id"],
                value=policy,
            )

            # Randomly submit a claim against an existing policy
            if active_policies and random.random() < 0.3:
                p = random.choice(active_policies)
                claim = make_claim_event(p["policy_id"], p["customer_id"])
                producer.send(
                    TOPICS["submission"],
                    key=claim["claim_id"],
                    value=claim,
                )

                # Immediately produce a decision 40% of the time
                if random.random() < 0.4:
                    decision = make_decision_event(claim["claim_id"])
                    producer.send(
                        TOPICS["decision"],
                        key=decision["claim_id"],
                        value=decision,
                    )

        producer.flush()
        # Cap active policies list to avoid unbounded memory growth
        if len(active_policies) > 10_000:
            active_policies = active_policies[-5_000:]

        time.sleep(1)


if __name__ == "__main__":
    run(events_per_second=100)

from pyspark.sql import SparkSession
from pyspark.sql import functions as F
from pyspark.sql.types import (
    StructType, StructField, StringType,
    DoubleType, BooleanType, TimestampType
)

KAFKA_BOOTSTRAP = "localhost:9092"
SNOWFLAKE_OPTIONS = {
    "sfURL":       "{ACCOUNT_NAME}",
    "sfUser":      "{USER}",
    "sfPassword":  "{PASSWORD}",
    "sfDatabase":  "INSURANCE_DB",
    "sfSchema":    "STREAMING",
    "sfWarehouse": "CLAIMS_WH",
}

spark = SparkSession.builder \
    .appName("InsuranceClaimsStreaming") \
    .config("spark.jars.packages",
            "org.apache.spark:spark-sql-kafka-0-10_2.12:3.4.0,"
            "net.snowflake:spark-snowflake_2.12:2.12.0-spark_3.4") \
    .getOrCreate()

spark.sparkContext.setLogLevel("WARN")

# --- Schema for claim submission events ---
claim_schema = StructType([
    StructField("event_id",      StringType(),    False),
    StructField("event_type",    StringType(),    True),
    StructField("claim_id",      StringType(),    True),
    StructField("policy_id",     StringType(),    True),
    StructField("customer_id",   StringType(),    True),
    StructField("claim_amount",  DoubleType(),    True),
    StructField("is_fraud",      BooleanType(),   True),
    StructField("fraud_reason",  StringType(),    True),
    StructField("timestamp",     TimestampType(), True),
])

# --- Read from Kafka ---
raw_stream = spark.readStream \
    .format("kafka") \
    .option("kafka.bootstrap.servers", KAFKA_BOOTSTRAP) \
    .option("subscribe", "claims.submissions,claims.decisions") \
    .option("startingOffsets", "latest") \
    .option("maxOffsetsPerTrigger", 10_000) \
    .load()

# --- Parse JSON payload ---
parsed = raw_stream.select(
    F.from_json(
        F.col("value").cast("string"), claim_schema
    ).alias("data"),
    F.col("timestamp").alias("kafka_timestamp"),
).select("data.*", "kafka_timestamp")

# --- Fraud scoring: flag high-velocity claimants ---
windowed = parsed \
    .withWatermark("timestamp", "2 minutes") \
    .groupBy(
        F.window("timestamp", "1 hour", "10 minutes"),
        F.col("customer_id"),
    ) \
    .agg(
        F.count("claim_id").alias("claims_in_window"),
        F.sum("claim_amount").alias("total_claimed"),
        F.max("is_fraud").alias("any_fraud_flag"),
    ) \
    .withColumn(
        "velocity_fraud_flag",
        F.col("claims_in_window") > 3
    ) \
    .withColumn("window_start", F.col("window.start")) \
    .withColumn("window_end",   F.col("window.end")) \
    .drop("window")

# --- Write micro-batches to Snowflake ---
def write_to_snowflake(batch_df, batch_id):
    if batch_df.isEmpty():
        return
    batch_df.write \
        .format("net.snowflake.spark.snowflake") \
        .options(**SNOWFLAKE_OPTIONS) \
        .option("dbtable", "FACT_CLAIM_VELOCITY") \
        .mode("append") \
        .save()
    print(f"Batch {batch_id}: wrote {batch_df.count()} rows to Snowflake")

query = windowed.writeStream \
    .foreachBatch(write_to_snowflake) \
    .outputMode("update") \
    .trigger(processingTime="30 seconds") \
    .option("checkpointLocation", "/tmp/checkpoints/claims_velocity") \
    .start()

query.awaitTermination()

from datetime import datetime, timedelta
from airflow import DAG
from airflow.operators.python import PythonOperator
from airflow.operators.bash import BashOperator
from airflow.providers.slack.operators.slack_webhook import SlackWebhookOperator
from airflow.models import Variable
import snowflake.connector

default_args = {
    "owner":            "data_engineering",
    "retries":          2,
    "retry_delay":      timedelta(minutes=5),
    "email_on_failure": True,
    "email":            ["data-alerts@bidabi.com"],
}

SNOWFLAKE_CONN = {
    "account":   Variable.get("SNOWFLAKE_ACCOUNT"),
    "user":      Variable.get("SNOWFLAKE_USER"),
    "password":  Variable.get("SNOWFLAKE_PASSWORD"),
    "database":  "INSURANCE_DB",
    "warehouse": "CLAIMS_WH",
    "schema":    "GOLD",
}

with DAG(
    dag_id="insurance_claims_nightly_reconciliation",
    default_args=default_args,
    description="Nightly batch reconciliation for the streaming claims pipeline",
    schedule_interval="0 1 * * *",  # 1 AM daily
    start_date=datetime(2025, 1, 1),
    catchup=False,
    tags=["insurance", "reconciliation", "streaming"],
) as dag:

    # Step 1: Run dbt models to rebuild Gold layer from S3 raw events
    run_dbt_models = BashOperator(
        task_id="run_dbt_gold_models",
        bash_command=(
            "cd /opt/dbt/insurance && "
            "dbt run --select tag:gold --target prod "
            "--vars '{run_date: {{ ds }}}'"
        ),
    )

    # Step 2: Run dbt tests on Gold models
    run_dbt_tests = BashOperator(
        task_id="run_dbt_tests",
        bash_command=(
            "cd /opt/dbt/insurance && "
            "dbt test --select tag:gold --target prod"
        ),
    )

    # Step 3: Reconcile streaming vs batch counts
    def reconcile_counts(**context):
        run_date = context["ds"]
        conn = snowflake.connector.connect(**SNOWFLAKE_CONN)
        cur = conn.cursor()

        # Streaming count from micro-batch writes
        cur.execute(f"""
            SELECT COUNT(*) FROM STREAMING.FACT_CLAIM_VELOCITY
            WHERE DATE(window_start) = '{run_date}'
        """)
        streaming_count = cur.fetchone()[0]

        # Batch count from dbt Gold model
        cur.execute(f"""
            SELECT COUNT(*) FROM GOLD.FACT_CLAIM_EVENTS
            WHERE DATE(event_timestamp) = '{run_date}'
        """)
        batch_count = cur.fetchone()[0]

        if batch_count == 0:
            raise ValueError(f"No batch records found for {run_date}")

        discrepancy_pct = abs(streaming_count - batch_count) / batch_count * 100
        context["ti"].xcom_push(key="discrepancy_pct", value=discrepancy_pct)
        context["ti"].xcom_push(key="streaming_count", value=streaming_count)
        context["ti"].xcom_push(key="batch_count",    value=batch_count)

        print(f"Streaming: {streaming_count} | Batch: {batch_count} | "
              f"Discrepancy: {discrepancy_pct:.4f}%")

        if discrepancy_pct > 0.1:
            raise ValueError(
                f"Discrepancy {discrepancy_pct:.4f}% exceeds 0.1% threshold"
            )
        cur.close()
        conn.close()

    reconcile = PythonOperator(
        task_id="reconcile_streaming_vs_batch",
        python_callable=reconcile_counts,
    )

    # Step 4: Slack alert on success
    notify_success = SlackWebhookOperator(
        task_id="notify_success",
        slack_webhook_conn_id="slack_data_alerts",
        message=(
            ":white_check_mark: *Insurance Claims Reconciliation* passed for "
            "{{ ds }}.\n"
            "Streaming: {{ ti.xcom_pull(task_ids='reconcile_streaming_vs_batch', key='streaming_count') }} | "
            "Batch: {{ ti.xcom_pull(task_ids='reconcile_streaming_vs_batch', key='batch_count') }} | "
            "Discrepancy: {{ ti.xcom_pull(task_ids='reconcile_streaming_vs_batch', key='discrepancy_pct') | round(4) }}%"
        ),
        trigger_rule="all_success",
    )

    run_dbt_models >> run_dbt_tests >> reconcile >> notify_success