┌──────────┐     ┌─────────┐     ┌───────────────────┐     ┌──────────┐
│  Event   │────>│  Kafka  │────>│ Spark Structured  │────>│  Delta   │
│ Sources  │     │ Cluster │     │    Streaming      │     │  Lake    │
└──────────┘     └─────────┘     └───────────────────┘     └──────────┘
  (APIs,           (Buffer,         (Transform,              (Bronze,
   Apps,            decouple)        aggregate,               Silver,
   IoT)                              enrich)                  Gold)

import json
import time
from datetime import datetime, timezone
from confluent_kafka import Producer, KafkaError
from confluent_kafka.serialization import (
    StringSerializer, SerializationContext, MessageField
)


class EventProducer:
    """Production Kafka producer with error handling and metrics."""

    def __init__(self, bootstrap_servers: str, topic: str):
        self.topic = topic
        self.delivered = 0
        self.failed = 0

        self.producer = Producer({
            "bootstrap.servers": bootstrap_servers,
            "client.id": f"producer-{topic}",
            "acks": "all",                    # Wait for all replicas
            "retries": 5,                     # Retry transient errors
            "retry.backoff.ms": 500,
            "enable.idempotence": True,       # Exactly-once producer
            "compression.type": "snappy",     # Compress for throughput
            "linger.ms": 20,                  # Batch messages for 20ms
            "batch.size": 65536,              # 64KB batch size
            "max.in.flight.requests.per.connection": 5,
        })

    def produce(
        self,
        key: str,
        value: dict,
        headers: dict | None = None,
    ):
        """Produce a message to Kafka."""
        # Add event metadata
        value["_event_time"] = datetime.now(timezone.utc).isoformat()
        value["_event_id"] = f"{key}_{int(time.time() * 1000)}"

        kafka_headers = []
        if headers:
            kafka_headers = [
                (k, v.encode("utf-8")) for k, v in headers.items()
            ]

        self.producer.produce(
            topic=self.topic,
            key=key,
            value=json.dumps(value),
            headers=kafka_headers,
            callback=self._delivery_callback,
        )

    def _delivery_callback(self, err, msg):
        if err:
            self.failed += 1
            print(f"Delivery failed: {err}")
        else:
            self.delivered += 1

    def flush(self, timeout: float = 30):
        """Wait for all messages to be delivered."""
        remaining = self.producer.flush(timeout)
        if remaining > 0:
            print(f"Warning: {remaining} messages not delivered")

    def get_stats(self) -> dict:
        return {
            "delivered": self.delivered,
            "failed": self.failed,
            "success_rate": (
                self.delivered / (self.delivered + self.failed)
                if (self.delivered + self.failed) > 0
                else 0
            ),
        }


# Usage
producer = EventProducer(
    bootstrap_servers="kafka-broker-1:9092,kafka-broker-2:9092",
    topic="user-events",
)

# Produce events
events = [
    {"user_id": "u123", "action": "page_view", "page": "/pricing"},
    {"user_id": "u456", "action": "purchase", "amount": 99.99},
    {"user_id": "u123", "action": "click", "element": "cta_button"},
]

for event in events:
    producer.produce(
        key=event["user_id"],    # Partition by user_id for ordering
        value=event,
        headers={"source": "web-app", "version": "1.0"},
    )

producer.flush()
print(f"Stats: {producer.get_stats()}")

import json
from confluent_kafka import Consumer, KafkaError, TopicPartition


class EventConsumer:
    """Production Kafka consumer with offset management."""

    def __init__(
        self,
        bootstrap_servers: str,
        topic: str,
        group_id: str,
        auto_commit: bool = False,
    ):
        self.topic = topic
        self.consumer = Consumer({
            "bootstrap.servers": bootstrap_servers,
            "group.id": group_id,
            "auto.offset.reset": "earliest",
            "enable.auto.commit": auto_commit,
            "max.poll.interval.ms": 300000,     # 5 min max processing
            "session.timeout.ms": 30000,
            "fetch.min.bytes": 1024,             # Wait for 1KB
            "fetch.wait.max.ms": 500,
        })
        self.consumer.subscribe([topic])

    def consume_batch(
        self, batch_size: int = 100, timeout: float = 1.0
    ) -> list[dict]:
        """Consume a batch of messages."""
        messages = []
        while len(messages) < batch_size:
            msg = self.consumer.poll(timeout)
            if msg is None:
                break
            if msg.error():
                if msg.error().code() == KafkaError._PARTITION_EOF:
                    continue
                print(f"Consumer error: {msg.error()}")
                continue

            messages.append({
                "key": msg.key().decode("utf-8") if msg.key() else None,
                "value": json.loads(msg.value().decode("utf-8")),
                "topic": msg.topic(),
                "partition": msg.partition(),
                "offset": msg.offset(),
                "timestamp": msg.timestamp(),
                "headers": {
                    h[0]: h[1].decode("utf-8")
                    for h in (msg.headers() or [])
                },
            })

        return messages

    def commit(self):
        """Manually commit offsets after successful processing."""
        self.consumer.commit()

    def close(self):
        self.consumer.close()


# Usage
consumer = EventConsumer(
    bootstrap_servers="kafka-broker-1:9092",
    topic="user-events",
    group_id="analytics-pipeline",
)

try:
    while True:
        batch = consumer.consume_batch(batch_size=100, timeout=2.0)
        if not batch:
            continue

        # Process batch
        for msg in batch:
            process_event(msg["value"])

        # Commit after successful processing
        consumer.commit()
        print(f"Processed {len(batch)} events")
except KeyboardInterrupt:
    pass
finally:
    consumer.close()

from pyspark.sql import SparkSession
from pyspark.sql.functions import (
    from_json, col, window, count, sum as spark_sum,
    avg, current_timestamp, expr, to_timestamp
)
from pyspark.sql.types import (
    StructType, StructField, StringType, DoubleType, TimestampType
)


spark = SparkSession.builder \
    .appName("UserEventsPipeline") \
    .config("spark.sql.extensions", "io.delta.sql.DeltaSparkSessionExtension") \
    .config("spark.sql.catalog.spark_catalog",
            "org.apache.spark.sql.delta.catalog.DeltaCatalog") \
    .getOrCreate()


# Define the event schema
event_schema = StructType([
    StructField("user_id", StringType(), True),
    StructField("action", StringType(), True),
    StructField("page", StringType(), True),
    StructField("element", StringType(), True),
    StructField("amount", DoubleType(), True),
    StructField("_event_time", StringType(), True),
    StructField("_event_id", StringType(), True),
])


def create_kafka_stream(spark: SparkSession, topic: str):
    """Read from Kafka topic as a streaming DataFrame."""

    raw_stream = spark.readStream \
        .format("kafka") \
        .option("kafka.bootstrap.servers", "kafka-broker-1:9092") \
        .option("subscribe", topic) \
        .option("startingOffsets", "latest") \
        .option("maxOffsetsPerTrigger", 100000) \
        .option("kafka.security.protocol", "SASL_SSL") \
        .load()

    # Parse JSON values
    parsed_stream = raw_stream \
        .select(
            col("key").cast("string").alias("kafka_key"),
            from_json(
                col("value").cast("string"), event_schema
            ).alias("data"),
            col("topic"),
            col("partition"),
            col("offset"),
            col("timestamp").alias("kafka_timestamp"),
        ) \
        .select(
            "kafka_key",
            "data.*",
            "topic",
            "partition",
            "offset",
            "kafka_timestamp",
        ) \
        .withColumn(
            "event_time",
            to_timestamp(col("_event_time"))
        )

    return parsed_stream

def write_bronze_stream(parsed_stream, checkpoint_path: str):
    """Write raw events to Bronze Delta table."""

    bronze_stream = parsed_stream \
        .withColumn("_bronze_ingested_at", current_timestamp()) \
        .writeStream \
        .format("delta") \
        .outputMode("append") \
        .option("checkpointLocation", f"{checkpoint_path}/bronze") \
        .option("mergeSchema", "true") \
        .trigger(processingTime="30 seconds") \
        .toTable("streaming.bronze.user_events")

    return bronze_stream

def create_silver_stream(bronze_table: str):
    """Read from Bronze and apply Silver transformations."""

    silver_stream = spark.readStream \
        .format("delta") \
        .table(bronze_table) \
        .filter(col("user_id").isNotNull()) \
        .filter(col("action").isNotNull()) \
        .dropDuplicates(["_event_id"]) \
        .withColumn("event_date", col("event_time").cast("date")) \
        .withColumn(
            "action_category",
            expr("""
                CASE
                    WHEN action IN ('page_view', 'scroll') THEN 'engagement'
                    WHEN action IN ('click', 'hover') THEN 'interaction'
                    WHEN action IN ('purchase', 'add_to_cart') THEN 'conversion'
                    ELSE 'other'
                END
            """)
        )

    return silver_stream


def write_silver_stream(silver_stream, checkpoint_path: str):
    """Write Silver stream to Delta table."""

    query = silver_stream \
        .withColumn("_silver_processed_at", current_timestamp()) \
        .writeStream \
        .format("delta") \
        .outputMode("append") \
        .option("checkpointLocation", f"{checkpoint_path}/silver") \
        .trigger(processingTime="30 seconds") \
        .toTable("streaming.silver.user_events")

    return query

def create_realtime_dashboard_stream(silver_table: str):
    """Create real-time aggregations for dashboards."""

    agg_stream = spark.readStream \
        .format("delta") \
        .table(silver_table) \
        .withWatermark("event_time", "5 minutes") \
        .groupBy(
            window(col("event_time"), "1 minute"),
            col("action_category"),
        ) \
        .agg(
            count("*").alias("event_count"),
            spark_sum(
                expr("CASE WHEN amount IS NOT NULL THEN amount ELSE 0 END")
            ).alias("total_revenue"),
            avg("amount").alias("avg_order_value"),
        ) \
        .select(
            col("window.start").alias("window_start"),
            col("window.end").alias("window_end"),
            "action_category",
            "event_count",
            "total_revenue",
            "avg_order_value",
        )

    return agg_stream


def write_dashboard_stream(agg_stream, checkpoint_path: str):
    """Write aggregations to Gold table."""

    query = agg_stream \
        .writeStream \
        .format("delta") \
        .outputMode("complete") \
        .option("checkpointLocation", f"{checkpoint_path}/gold_dashboard") \
        .trigger(processingTime="1 minute") \
        .toTable("streaming.gold.realtime_dashboard")

    return query

Kafka → Spark → Delta Lake (Exactly-Once Path)

1. Kafka: Idempotent producer (enable.idempotence=True)
   - Deduplicates at the broker level using sequence numbers

2. Spark Structured Streaming: Checkpointing
   - Tracks Kafka offsets in checkpoint directory
   - On failure, replays from last committed offset
   - Uses write-ahead log for state management

3. Delta Lake: ACID transactions
   - Each micro-batch is an atomic commit
   - If a batch partially fails, it's fully rolled back
   - Combined with Spark checkpointing = exactly-once writes

# Watermark tells Spark how late data can arrive
stream_with_watermark = parsed_stream \
    .withWatermark("event_time", "10 minutes") \
    .groupBy(
        window(col("event_time"), "5 minutes"),
        col("action"),
    ) \
    .count()

# Events arriving more than 10 minutes late are dropped
# Events within the 10-minute window update the aggregation

# Deduplicate within a watermark window
deduplicated = parsed_stream \
    .withWatermark("event_time", "1 hour") \
    .dropDuplicates(["_event_id", "event_time"])

def monitor_stream(query):
    """Extract monitoring metrics from a streaming query."""
    while query.isActive:
        progress = query.lastProgress
        if progress:
            metrics = {
                "batch_id": progress["batchId"],
                "input_rows": progress["numInputRows"],
                "processing_time_ms": progress["batchDuration"],
                "input_rows_per_sec": progress.get("inputRowsPerSecond", 0),
                "processed_rows_per_sec": progress.get(
                    "processedRowsPerSecond", 0
                ),
            }

            # Alert on processing lag
            if metrics["input_rows_per_sec"] > 0:
                ratio = (
                    metrics["processed_rows_per_sec"]
                    / metrics["input_rows_per_sec"]
                )
                if ratio < 0.8:
                    print(
                        f"WARNING: Processing falling behind. "
                        f"Ratio: {ratio:.2f}"
                    )

            print(f"Batch {metrics['batch_id']}: {metrics}")

        time.sleep(30)

# Check consumer group lag
kafka-consumer-groups.sh \
  --bootstrap-server kafka-broker-1:9092 \
  --describe \
  --group analytics-pipeline

# Key metric: LAG column
# If LAG grows continuously, consumers can't keep up