Streaming vs Batch Processing.

# architectures/lambda_architecture.py

"""
Lambda Architecture implementation.

Combines batch accuracy with streaming speed.
"""

from dataclasses import dataclass
from datetime import datetime, timedelta, timezone
from typing import Dict, Any, Optional
from abc import ABC, abstractmethod
from pyspark.sql import functions as F


@dataclass
class MetricResult:
    """Result from either batch or speed layer."""
    value: float
    timestamp: datetime
    is_batch: bool
    coverage_start: datetime
    coverage_end: datetime


class BatchLayer:
    """
    Batch processing layer.
    
    Processes all historical data with Spark.
    Runs periodically (e.g., hourly).
    """
    
    def __init__(self, spark_session, data_lake_path: str):
        self.spark = spark_session
        self.data_path = data_lake_path
        self.last_batch_time: Optional[datetime] = None
    
    def run_batch_job(self, end_time: datetime) -> Dict[str, MetricResult]:
        """
        Run batch aggregation up to end_time.
        
        This processes ALL data from the beginning,
        producing the authoritative "batch views".
        """
        
        # Read all raw events from data lake
        events_df = self.spark.read.parquet(
            f"{self.data_path}/events/"
        ).filter(
            f"event_time < '{end_time.isoformat()}'"
        )
        
        # Compute aggregations
        metrics = {}
        
        # Daily revenue
        daily_revenue = events_df.filter(
            "event_type = 'order.placed'"
        ).groupBy(
            F.to_date("event_time").alias("date")
        ).agg(
            F.sum("payload.total_amount").alias("revenue"),
            F.count("*").alias("order_count")
        ).collect()
        
        for row in daily_revenue:
            metrics[f"revenue_{row.date}"] = MetricResult(
                value=row.revenue,
                timestamp=end_time,
                is_batch=True,
                coverage_start=datetime.combine(row.date, datetime.min.time(), tzinfo=timezone.utc),
                # Today's batch only covers data up to end_time
                coverage_end=min(
                    datetime.combine(row.date, datetime.max.time(), tzinfo=timezone.utc), end_time
                )
            )
        
        # DAU (Daily Active Users)
        dau = events_df.groupBy(
            F.to_date("event_time").alias("date")
        ).agg(
            F.countDistinct("user_id").alias("dau")
        ).collect()
        
        for row in dau:
            metrics[f"dau_{row.date}"] = MetricResult(
                value=row.dau,
                timestamp=end_time,
                is_batch=True,
                coverage_start=datetime.combine(row.date, datetime.min.time(), tzinfo=timezone.utc),
                coverage_end=min(
                    datetime.combine(row.date, datetime.max.time(), tzinfo=timezone.utc), end_time
                )
            )
        
        self.last_batch_time = end_time
        
        return metrics
    
    def get_batch_view(self, metric_name: str) -> Optional[MetricResult]:
        """Get the latest batch view for a metric."""
        # In practice, read from batch view storage (e.g., Cassandra, Redis)
        pass


class SpeedLayer:
    """
    Speed/Streaming layer.
    
    Processes only events since last batch.
    Provides real-time updates.
    """
    
    def __init__(self, flink_env, kafka_source: str):
        self.env = flink_env
        self.kafka_source = kafka_source
        self.realtime_state: Dict[str, MetricResult] = {}
    
    def start_streaming(self, batch_watermark: datetime):
        """
        Start streaming from batch watermark.
        
        Only processes events AFTER the last batch completed.
        """
        
        # In Flink, this would be a DataStream job:
        # 
        # stream = env.from_source(KafkaSource(...), watermark_strategy, "events")
        #     .filter(lambda e: e.event_time > batch_watermark)
        #     .key_by(lambda e: e.event_type)
        #     .window(TumblingEventTimeWindows.of(Time.minutes(1)))
        #     .aggregate(MetricAggregator())
        #     .add_sink(RealtimeViewSink())
        
        pass
    
    def get_realtime_view(
        self,
        metric_name: str,
        since: datetime
    ) -> Optional[MetricResult]:
        """
        Get real-time aggregation since a timestamp.
        
        This fills the gap between last batch and now.
        """
        
        # In practice, query from real-time store (e.g., Druid, Redis)
        return self.realtime_state.get(metric_name)


class ServingLayer:
    """
    Serving layer that merges batch and speed results.
    """
    
    def __init__(self, batch_layer: BatchLayer, speed_layer: SpeedLayer):
        self.batch = batch_layer
        self.speed = speed_layer
    
    def query_metric(self, metric_name: str, date: datetime.date) -> float:
        """
        Query a metric by merging batch and real-time.
        
        Logic:
        1. Get batch view (covers up to last batch time)
        2. Get speed view (covers batch time to now)
        3. Combine appropriately
        """
        
        today = datetime.now(timezone.utc).date()
        
        if date < today:
            # Historical date: batch only (complete)
            batch_result = self.batch.get_batch_view(f"{metric_name}_{date}")
            return batch_result.value if batch_result else 0
        
        elif date == today:
            # Today: might need batch + realtime
            batch_result = self.batch.get_batch_view(f"{metric_name}_{date}")
            
            if batch_result:
                # Batch has partial day, add realtime since batch
                realtime_result = self.speed.get_realtime_view(
                    f"{metric_name}_{date}",
                    since=batch_result.coverage_end
                )
                
                batch_value = batch_result.value
                realtime_value = realtime_result.value if realtime_result else 0
                
                return batch_value + realtime_value
            
            else:
                # No batch yet today, realtime only
                realtime_result = self.speed.get_realtime_view(
                    f"{metric_name}_{date}",
                    since=datetime.combine(date, datetime.min.time(), tzinfo=timezone.utc)
                )
                return realtime_result.value if realtime_result else 0
        
        else:
            # Future date: no data
            return 0

# architectures/kappa_architecture.py

"""
Kappa Architecture implementation.

Single streaming codebase for all processing.
Reprocessing via replay from event log.
"""

from dataclasses import dataclass
from datetime import datetime
from typing import Dict, Any, Optional, Callable
from enum import Enum
import asyncio


class ProcessorVersion(Enum):
    """Processor version for blue-green deployments."""
    V1 = "v1"
    V2 = "v2"


@dataclass
class ProcessorConfig:
    """Configuration for stream processor."""
    version: ProcessorVersion
    kafka_brokers: list
    input_topic: str
    output_table: str
    consumer_group: str
    start_offset: str = "earliest"  # or "latest" or timestamp


class KappaProcessor:
    """
    Unified stream processor for Kappa architecture.
    
    Key principle: One codebase, used for both
    real-time processing and historical reprocessing.
    """
    
    def __init__(self, config: ProcessorConfig):
        self.config = config
        self.aggregators: Dict[str, Callable] = {}
        self.state: Dict[str, Any] = {}
    
    def register_aggregation(
        self,
        name: str,
        aggregator: Callable
    ):
        """Register an aggregation function."""
        self.aggregators[name] = aggregator
    
    async def process_event(self, event: Dict) -> Dict[str, Any]:
        """
        Process a single event through all aggregators.
        
        This same code handles:
        1. Live events (normal operation)
        2. Historical events (reprocessing)
        """
        
        results = {}
        
        for name, aggregator in self.aggregators.items():
            result = await aggregator(event, self.state)
            if result:
                results[name] = result
        
        return results
    
    async def run(self):
        """
        Main processing loop.
        
        Consumes from Kafka, processes events, outputs to serving layer.
        """
        
        # In practice, use Flink/Kafka Streams
        # This is simplified pseudocode
        
        consumer = self._create_consumer()
        
        async for event in consumer:
            try:
                results = await self.process_event(event)
                
                for metric_name, value in results.items():
                    await self._write_to_serving_layer(
                        metric_name,
                        value,
                        event.get("event_time")
                    )
                    
            except Exception as e:
                await self._handle_error(event, e)
    
    def _create_consumer(self):
        """Create Kafka consumer with appropriate offset."""
        # Start from earliest for reprocessing
        # Start from latest for caught-up live processing
        pass
    
    async def _write_to_serving_layer(
        self,
        metric: str,
        value: Any,
        timestamp: datetime
    ):
        """Write result to serving layer (e.g., Druid, ClickHouse)."""
        pass
    
    async def _handle_error(self, event: Dict, error: Exception):
        """Handle processing error."""
        # Log error, send to DLQ, continue
        pass


class KappaOrchestrator:
    """
    Manages Kappa processor deployments and reprocessing.
    """
    
    def __init__(self):
        self.active_version: Optional[ProcessorVersion] = None
        self.processors: Dict[ProcessorVersion, KappaProcessor] = {}
    
    async def deploy_new_version(
        self,
        new_version: ProcessorVersion,
        processor: KappaProcessor
    ):
        """
        Deploy new processor version with reprocessing.
        
        Steps:
        1. Start new processor from beginning of log
        2. Wait for it to catch up to live
        3. Switch traffic to new version
        4. Stop old version
        """
        
        # Start new processor (reads from earliest)
        self.processors[new_version] = processor
        asyncio.create_task(processor.run())
        
        # Wait for catch-up
        await self._wait_for_catchup(processor)
        
        # Atomic switch
        old_version = self.active_version
        self.active_version = new_version
        
        # Stop old processor
        if old_version:
            await self._stop_processor(old_version)
    
    async def _wait_for_catchup(self, processor: KappaProcessor):
        """Wait for processor to catch up to live data."""
        
        while True:
            lag = await self._get_consumer_lag(processor)
            
            if lag < 1000:  # Less than 1000 events behind
                break
            
            await asyncio.sleep(10)
    
    async def _get_consumer_lag(self, processor: KappaProcessor) -> int:
        """Get consumer lag (how far behind live)."""
        # Query Kafka for consumer group lag
        pass
    
    async def _stop_processor(self, version: ProcessorVersion):
        """Gracefully stop a processor version."""
        pass


# =============================================================================
# Aggregation Functions
# =============================================================================

async def revenue_aggregator(
    event: Dict,
    state: Dict
) -> Optional[Dict]:
    """
    Aggregate revenue by day.
    
    Same code works for:
    - Live events: Updates today's revenue
    - Reprocessing: Recomputes historical revenue
    """
    
    if event.get("event_type") != "order.placed":
        return None
    
    event_time = datetime.fromisoformat(event["event_time"])
    date_key = event_time.strftime("%Y-%m-%d")
    amount = event.get("payload", {}).get("total_amount", 0)
    
    # Update state
    if date_key not in state:
        state[date_key] = {"revenue": 0, "orders": 0}
    
    state[date_key]["revenue"] += amount
    state[date_key]["orders"] += 1
    
    return {
        "date": date_key,
        "revenue": state[date_key]["revenue"],
        "orders": state[date_key]["orders"]
    }


async def active_users_aggregator(
    event: Dict,
    state: Dict
) -> Optional[Dict]:
    """Aggregate unique active users by day."""
    
    event_time = datetime.fromisoformat(event["event_time"])
    date_key = event_time.strftime("%Y-%m-%d")
    user_id = event.get("user_id")
    
    if not user_id:
        return None
    
    # Track unique users
    users_key = f"users_{date_key}"
    if users_key not in state:
        state[users_key] = set()
    
    state[users_key].add(user_id)
    
    return {
        "date": date_key,
        "dau": len(state[users_key])
    }

# streaming/flink_processor.py

"""
Apache Flink stream processing implementation.

Flink concepts:
- DataStream: Unbounded stream of records
- Operators: Transformations (map, filter, keyBy, window)
- State: Per-key state for aggregations
- Checkpoints: Fault tolerance via periodic snapshots
- Watermarks: Tracking event time progress
"""

from dataclasses import dataclass
from datetime import datetime, timedelta
from typing import Dict, Any, Optional, Iterator
from abc import ABC, abstractmethod


# =============================================================================
# Core Abstractions (Simplified Flink-like API)
# =============================================================================

@dataclass
class Event:
    """An event in the stream."""
    event_id: str
    event_type: str
    event_time: datetime
    user_id: Optional[str]
    payload: Dict[str, Any]


class DataStream:
    """
    Represents an unbounded stream of events.
    """
    
    def __init__(self, source: Iterator[Event]):
        self._source = source
        self._transformations = []
    
    def filter(self, predicate) -> 'DataStream':
        """Filter events based on predicate."""
        self._transformations.append(('filter', predicate))
        return self
    
    def map(self, mapper) -> 'DataStream':
        """Transform each event."""
        self._transformations.append(('map', mapper))
        return self
    
    def key_by(self, key_selector) -> 'KeyedStream':
        """Partition stream by key for stateful processing."""
        return KeyedStream(self, key_selector)


class KeyedStream:
    """
    Stream partitioned by key.
    
    Enables:
    - Stateful processing per key
    - Windowed aggregations per key
    - Exactly-once per-key guarantees
    """
    
    def __init__(self, stream: DataStream, key_selector):
        self.stream = stream
        self.key_selector = key_selector
    
    def window(self, window_assigner: 'WindowAssigner') -> 'WindowedStream':
        """Apply windowing to the keyed stream."""
        return WindowedStream(self, window_assigner)
    
    def process(self, processor: 'KeyedProcessFunction') -> DataStream:
        """Apply stateful processing function."""
        # Execute processor with per-key state
        pass


class WindowAssigner(ABC):
    """Assigns events to windows."""
    
    @abstractmethod
    def assign_windows(self, event: Event) -> list:
        pass


class TumblingEventTimeWindows(WindowAssigner):
    """
    Fixed-size, non-overlapping windows based on event time.
    
    Example: 1-hour windows
    [00:00-01:00), [01:00-02:00), [02:00-03:00), ...
    """
    
    def __init__(self, size: timedelta):
        self.size = size
    
    def assign_windows(self, event: Event) -> list:
        window_start = self._get_window_start(event.event_time)
        window_end = window_start + self.size
        return [(window_start, window_end)]
    
    def _get_window_start(self, timestamp: datetime) -> datetime:
        """Get the start of the window containing timestamp."""
        epoch = datetime(1970, 1, 1)
        elapsed = (timestamp - epoch).total_seconds()
        window_seconds = self.size.total_seconds()
        window_start_seconds = (elapsed // window_seconds) * window_seconds
        return epoch + timedelta(seconds=window_start_seconds)


class SlidingEventTimeWindows(WindowAssigner):
    """
    Overlapping windows that slide.
    
    Example: 1-hour windows, sliding every 15 minutes
    [00:00-01:00), [00:15-01:15), [00:30-01:30), ...
    """
    
    def __init__(self, size: timedelta, slide: timedelta):
        self.size = size
        self.slide = slide
    
    def assign_windows(self, event: Event) -> list:
        windows = []
        timestamp = event.event_time
        
        # Event belongs to multiple overlapping windows
        earliest_start = self._get_earliest_window_start(timestamp)
        
        start = earliest_start
        while start + self.size > timestamp:
            windows.append((start, start + self.size))
            start = start - self.slide
        
        return windows
    
    def _get_earliest_window_start(self, timestamp: datetime) -> datetime:
        """Get earliest window that contains this timestamp."""
        epoch = datetime(1970, 1, 1)
        elapsed = (timestamp - epoch).total_seconds()
        slide_seconds = self.slide.total_seconds()
        
        # Align to slide boundaries
        aligned = (elapsed // slide_seconds) * slide_seconds
        return epoch + timedelta(seconds=aligned)


class SessionWindows(WindowAssigner):
    """
    Windows based on activity sessions.
    
    A session ends when there's a gap > threshold.
    """
    
    def __init__(self, gap: timedelta):
        self.gap = gap
    
    def assign_windows(self, event: Event) -> list:
        # Sessions are merged dynamically as events arrive
        # Initial window is [event_time, event_time + gap)
        return [(event.event_time, event.event_time + self.gap)]


# =============================================================================
# Windowed Aggregation
# =============================================================================

class WindowedStream:
    """
    Keyed stream with windowing applied.
    """
    
    def __init__(self, keyed_stream: KeyedStream, assigner: WindowAssigner):
        self.keyed_stream = keyed_stream
        self.assigner = assigner
    
    def aggregate(self, aggregator: 'AggregateFunction') -> DataStream:
        """
        Apply aggregation function to each window.
        
        Flink incrementally aggregates as events arrive,
        not waiting for window to complete.
        """
        pass
    
    def reduce(self, reducer) -> DataStream:
        """Reduce events in window to single value."""
        pass


class AggregateFunction(ABC):
    """
    Incremental aggregation function.
    
    Maintains an accumulator that's updated per event.
    """
    
    @abstractmethod
    def create_accumulator(self) -> Any:
        """Create initial accumulator."""
        pass
    
    @abstractmethod
    def add(self, value: Any, accumulator: Any) -> Any:
        """Add value to accumulator."""
        pass
    
    @abstractmethod
    def get_result(self, accumulator: Any) -> Any:
        """Get result from accumulator."""
        pass
    
    @abstractmethod
    def merge(self, acc1: Any, acc2: Any) -> Any:
        """Merge two accumulators (for session windows)."""
        pass


class RevenueAggregator(AggregateFunction):
    """Aggregate revenue in a window."""
    
    def create_accumulator(self) -> Dict:
        return {"total": 0, "count": 0}
    
    def add(self, order: Dict, acc: Dict) -> Dict:
        acc["total"] += order.get("amount", 0)
        acc["count"] += 1
        return acc
    
    def get_result(self, acc: Dict) -> Dict:
        return {
            "revenue": acc["total"],
            "orders": acc["count"],
            "avg_order": acc["total"] / acc["count"] if acc["count"] > 0 else 0
        }
    
    def merge(self, acc1: Dict, acc2: Dict) -> Dict:
        return {
            "total": acc1["total"] + acc2["total"],
            "count": acc1["count"] + acc2["count"]
        }

# streaming/watermarks.py

"""
Watermarks for handling event time in streaming.

The Problem:
Events arrive out of order. When can we close a window?

Example:
  Real time    →  10:00  10:01  10:02  10:03  10:04
  Events arrive:  [A-9:58] [B-9:55] [C-10:01] [D-9:59] [E-10:02]
  
  When can we emit the [9:00-10:00) window?
  We might receive a 9:xx event at any time!

Solution: WATERMARKS
  "I believe all events with time <= W have arrived"
  When watermark passes window end, window can fire.
"""

from dataclasses import dataclass
from datetime import datetime, timedelta
from typing import Optional


@dataclass
class Watermark:
    """
    Represents progress of event time.
    
    Watermark(t) means: "No more events with event_time <= t expected"
    """
    timestamp: datetime


class WatermarkStrategy:
    """
    Strategy for generating watermarks.
    """
    
    def __init__(self, max_out_of_orderness: timedelta):
        """
        Args:
            max_out_of_orderness: How late events can be.
                If events can be up to 5 minutes late,
                watermark = max_event_time - 5 minutes
        """
        self.max_out_of_orderness = max_out_of_orderness
        self.current_max_timestamp: Optional[datetime] = None
    
    def on_event(self, event_time: datetime) -> Optional[Watermark]:
        """Update watermark based on new event."""
        
        if self.current_max_timestamp is None:
            self.current_max_timestamp = event_time
        else:
            self.current_max_timestamp = max(
                self.current_max_timestamp,
                event_time
            )
        
        # Watermark = max seen - max out-of-orderness
        watermark_time = self.current_max_timestamp - self.max_out_of_orderness
        
        return Watermark(watermark_time)
    
    def get_current_watermark(self) -> Optional[Watermark]:
        """Get current watermark."""
        if self.current_max_timestamp:
            return Watermark(
                self.current_max_timestamp - self.max_out_of_orderness
            )
        return None


class LateDataHandler:
    """
    Handles events that arrive after their window closed.
    """
    
    def __init__(
        self,
        allowed_lateness: timedelta,
        late_output_tag: str = "late-data"
    ):
        """
        Args:
            allowed_lateness: How long to keep window state after closing
            late_output_tag: Tag for side output of late events
        """
        self.allowed_lateness = allowed_lateness
        self.late_output_tag = late_output_tag
    
    def is_late(
        self,
        event_time: datetime,
        watermark: datetime,
        window_end: datetime
    ) -> bool:
        """Check if event is late for its window."""
        
        # Window hasn't closed yet
        if watermark < window_end:
            return False
        
        # Window closed, but within allowed lateness
        if watermark < window_end + self.allowed_lateness:
            return False  # Can still update window
        
        # Too late, drop or side output
        return True

# streaming/state_management.py

"""
State management in stream processing.

State is the memory of past events needed for current processing.

Examples:
- Running sum: Need previous sum
- Distinct count: Need set of seen IDs
- Session windows: Need events in current session
- Joins: Need buffer of events from both streams
"""

from dataclasses import dataclass
from typing import Dict, Any, Optional, Generic, TypeVar
from abc import ABC, abstractmethod
from datetime import datetime, timedelta
import pickle


T = TypeVar('T')


class StateBackend(ABC):
    """
    Backend for storing stream processing state.
    
    Options:
    - Memory: Fast, limited by heap (restored from checkpoints)
    - RocksDB: Larger-than-memory, persistent locally
    - Remote (Redis, etc.): Shared across nodes
    """
    
    @abstractmethod
    def get(self, key: str) -> Optional[Any]:
        pass
    
    @abstractmethod
    def put(self, key: str, value: Any):
        pass
    
    @abstractmethod
    def delete(self, key: str):
        pass
    
    @abstractmethod
    def checkpoint(self) -> bytes:
        """Serialize all state for checkpointing."""
        pass
    
    @abstractmethod
    def restore(self, data: bytes):
        """Restore state from checkpoint."""
        pass


class InMemoryStateBackend(StateBackend):
    """Simple in-memory state backend."""
    
    def __init__(self):
        self._state: Dict[str, Any] = {}
    
    def get(self, key: str) -> Optional[Any]:
        return self._state.get(key)
    
    def put(self, key: str, value: Any):
        self._state[key] = value
    
    def delete(self, key: str):
        self._state.pop(key, None)
    
    def checkpoint(self) -> bytes:
        return pickle.dumps(self._state)
    
    def restore(self, data: bytes):
        self._state = pickle.loads(data)


class RocksDBStateBackend(StateBackend):
    """
    RocksDB-backed state for larger-than-memory state.
    
    Used in production Flink deployments.
    """
    
    def __init__(self, db_path: str):
        # In practice: import rocksdb; self.db = rocksdb.DB(db_path)
        self.db_path = db_path
        self._db = {}  # Simplified
    
    def get(self, key: str) -> Optional[Any]:
        value_bytes = self._db.get(key.encode())
        if value_bytes:
            return pickle.loads(value_bytes)
        return None
    
    def put(self, key: str, value: Any):
        self._db[key.encode()] = pickle.dumps(value)
    
    def delete(self, key: str):
        self._db.pop(key.encode(), None)
    
    def checkpoint(self) -> bytes:
        # In practice: Use RocksDB snapshots
        return pickle.dumps(dict(self._db))
    
    def restore(self, data: bytes):
        self._db = pickle.loads(data)


class ValueState(Generic[T]):
    """
    State holding a single value per key.
    
    Example: Current sum for a user
    """
    
    def __init__(self, backend: StateBackend, name: str, key: str):
        self.backend = backend
        self.state_key = f"{name}:{key}"
    
    def value(self) -> Optional[T]:
        return self.backend.get(self.state_key)
    
    def update(self, value: T):
        self.backend.put(self.state_key, value)
    
    def clear(self):
        self.backend.delete(self.state_key)


class ListState(Generic[T]):
    """
    State holding a list of values per key.
    
    Example: All events in a session
    """
    
    def __init__(self, backend: StateBackend, name: str, key: str):
        self.backend = backend
        self.state_key = f"{name}:{key}"
    
    def get(self) -> list:
        return self.backend.get(self.state_key) or []
    
    def add(self, value: T):
        current = self.get()
        current.append(value)
        self.backend.put(self.state_key, current)
    
    def clear(self):
        self.backend.delete(self.state_key)


class MapState(Generic[T]):
    """
    State holding a map of values per key.
    
    Example: Count per category for a user
    """
    
    def __init__(self, backend: StateBackend, name: str, key: str):
        self.backend = backend
        self.state_key = f"{name}:{key}"
    
    def get(self, map_key: str) -> Optional[T]:
        current = self.backend.get(self.state_key) or {}
        return current.get(map_key)
    
    def put(self, map_key: str, value: T):
        current = self.backend.get(self.state_key) or {}
        current[map_key] = value
        self.backend.put(self.state_key, current)
    
    def keys(self) -> list:
        current = self.backend.get(self.state_key) or {}
        return list(current.keys())
    
    def clear(self):
        self.backend.delete(self.state_key)


# =============================================================================
# Stateful Processing Example
# =============================================================================

class UserSessionProcessor:
    """
    Stateful processor that tracks user sessions.
    
    A session ends after 30 minutes of inactivity.
    """
    
    SESSION_TIMEOUT = timedelta(minutes=30)
    
    def __init__(self, state_backend: StateBackend):
        self.backend = state_backend
    
    def process_event(self, user_id: str, event: Dict) -> Optional[Dict]:
        """
        Process event, maintain session state, emit session on close.
        """
        
        # Get session state for this user
        session_state = ValueState[Dict](self.backend, "session", user_id)
        timer_state = ValueState[datetime](self.backend, "timer", user_id)
        
        event_time = datetime.fromisoformat(event["event_time"])
        current_session = session_state.value()
        
        result = None
        
        if current_session is None:
            # New session
            current_session = {
                "user_id": user_id,
                "start_time": event_time.isoformat(),
                "events": [],
                "event_count": 0
            }
        
        else:
            # Check if session timed out
            last_timer = timer_state.value()
            if last_timer and event_time > last_timer:
                # Session closed, emit it
                result = {
                    "type": "session_complete",
                    "session": current_session
                }
                
                # Start new session
                current_session = {
                    "user_id": user_id,
                    "start_time": event_time.isoformat(),
                    "events": [],
                    "event_count": 0
                }
        
        # Add event to session
        current_session["events"].append(event)
        current_session["event_count"] += 1
        current_session["last_event_time"] = event_time.isoformat()
        
        # Update state
        session_state.update(current_session)
        timer_state.update(event_time + self.SESSION_TIMEOUT)
        
        return result

# batch/spark_processor.py

"""
Apache Spark batch processing for analytics.

Spark concepts:
- DataFrame: Distributed table of data
- Transformations: Lazy operations (filter, map, join)
- Actions: Trigger computation (collect, write)
- Partitioning: How data is distributed
- Caching: Keep intermediate results in memory
"""

from dataclasses import dataclass
from datetime import datetime, date, timedelta, timezone
from typing import Dict, Any, List, Optional
import json
from pyspark.sql import functions as F


@dataclass
class SparkConfig:
    """Spark application configuration."""
    app_name: str
    master: str = "local[*]"  # or "yarn", "k8s://..."
    executor_memory: str = "4g"
    executor_cores: int = 4
    num_executors: int = 10


class AnalyticsBatchJob:
    """
    Batch analytics job using Spark.
    
    Processes historical data to compute accurate metrics.
    """
    
    def __init__(self, spark_session):
        self.spark = spark_session
    
    def compute_daily_metrics(
        self,
        events_path: str,
        output_path: str,
        target_date: date
    ) -> Dict[str, Any]:
        """
        Compute all daily metrics for a specific date.
        
        This is the batch layer of Lambda architecture.
        """
        
        # Read events for the day
        events = self.spark.read.parquet(events_path).filter(
            f"date(event_time) = '{target_date}'"
        )
        
        # Cache for multiple aggregations
        events.cache()
        
        metrics = {}
        
        # Revenue metrics
        revenue_row = events.filter(
            "event_type = 'order.placed'"
        ).agg(
            F.sum("payload.total_amount").alias("revenue"),
            F.count("*").alias("order_count")
        ).collect()[0]
        
        metrics["revenue"] = revenue_row["revenue"] or 0
        metrics["order_count"] = revenue_row["order_count"] or 0
        
        # User metrics
        user_metrics = events.agg(
            F.countDistinct("user_id").alias("dau")
        ).collect()[0]
        
        metrics["dau"] = user_metrics["dau"] or 0
        
        # Funnel metrics
        metrics["funnel"] = self._compute_funnel(events)
        
        # Product metrics
        metrics["top_products"] = self._compute_top_products(events)
        
        # Write results
        self._write_metrics(output_path, target_date, metrics)
        
        return metrics
    
    def _compute_funnel(self, events) -> Dict[str, int]:
        """Compute conversion funnel."""
        
        funnel = {}
        
        # View → Cart → Purchase funnel
        funnel["product_views"] = events.filter(
            "event_type = 'product.viewed'"
        ).select("user_id").distinct().count()
        
        funnel["add_to_cart"] = events.filter(
            "event_type = 'cart.updated' AND payload.action = 'add'"
        ).select("user_id").distinct().count()
        
        funnel["purchases"] = events.filter(
            "event_type = 'order.placed'"
        ).select("user_id").distinct().count()
        
        # Conversion rates
        if funnel["product_views"] > 0:
            funnel["view_to_cart_rate"] = (
                funnel["add_to_cart"] / funnel["product_views"]
            )
            funnel["view_to_purchase_rate"] = (
                funnel["purchases"] / funnel["product_views"]
            )
        
        if funnel["add_to_cart"] > 0:
            funnel["cart_to_purchase_rate"] = (
                funnel["purchases"] / funnel["add_to_cart"]
            )
        
        return funnel
    
    def _compute_top_products(self, events, limit: int = 100) -> List[Dict]:
        """Compute top products by various metrics."""
        
        is_view = F.col("event_type") == "product.viewed"
        is_order = F.col("event_type") == "order.placed"
        
        product_stats = events.filter(
            "event_type IN ('product.viewed', 'order.placed')"
        ).groupBy(
            F.col("payload.product_id").alias("product_id")
        ).agg(
            F.count(F.when(is_view, 1)).alias("view_count"),
            F.sum(F.when(is_order, F.col("payload.total_amount"))).alias("revenue")
        ).orderBy(
            F.desc("view_count")
        ).limit(limit).collect()
        
        return [
            {
                "product_id": row["product_id"],
                "view_count": row["view_count"],
                "revenue": row["revenue"] or 0
            }
            for row in product_stats
        ]
    
    def _write_metrics(
        self,
        output_path: str,
        target_date: date,
        metrics: Dict
    ):
        """Write metrics to output storage."""
        
        # Write to partitioned parquet
        self.spark.createDataFrame([{
            "date": target_date.isoformat(),
            "metrics": json.dumps(metrics),  # Mixed nested types → JSON
            "computed_at": datetime.now(timezone.utc).isoformat()
        }]).write.mode("overwrite").parquet(
            f"{output_path}/date={target_date}"
        )


class IncrementalBatchJob:
    """
    Incremental batch job that processes only new data.
    
    More efficient than full recomputation.
    """
    
    def __init__(self, spark_session):
        self.spark = spark_session
    
    def process_incremental(
        self,
        events_path: str,
        state_path: str,
        next_state_path: str,
        output_path: str,
        watermark: datetime
    ):
        """
        Process only events since last watermark.
        
        Merge with existing state for complete picture. The merged state
        goes to next_state_path: Spark can't overwrite state_path while
        it is still (lazily) reading from it.
        """
        
        # Read new events only
        new_events = self.spark.read.parquet(events_path).filter(
            f"event_time > '{watermark.isoformat()}'"
        )
        
        # Read existing aggregates
        existing = self.spark.read.parquet(state_path)
        
        # Compute incremental aggregates
        incremental = new_events.groupBy(
            F.to_date("event_time").alias("date")
        ).agg(
            F.sum("payload.total_amount").alias("revenue"),
            F.count("*").alias("event_count")
        )
        
        # Merge: Add incremental to existing (same column names)
        merged = existing.unionByName(incremental).groupBy("date").agg(
            F.sum("revenue").alias("revenue"),
            F.sum("event_count").alias("event_count")
        )
        
        # Write updated state to the next version (not the path being read)
        merged.write.mode("overwrite").parquet(next_state_path)
        
        # Also write to serving layer
        merged.write.mode("overwrite").parquet(output_path)

# batch/spark_streaming.py

"""
Spark Structured Streaming for unified batch + stream.

Structured Streaming treats streams as unbounded tables,
using the same DataFrame API as batch.
"""

from pyspark.sql import SparkSession
from pyspark.sql.functions import *
from pyspark.sql.types import *


class UnifiedAnalyticsProcessor:
    """
    Unified processor that works for both batch and streaming.
    
    Same code, different execution modes.
    """
    
    def __init__(self, spark: SparkSession):
        self.spark = spark
    
    def create_streaming_source(self, kafka_brokers: str, topic: str):
        """Create streaming DataFrame from Kafka."""
        
        return self.spark.readStream \
            .format("kafka") \
            .option("kafka.bootstrap.servers", kafka_brokers) \
            .option("subscribe", topic) \
            .option("startingOffsets", "latest") \
            .load() \
            .select(
                from_json(
                    col("value").cast("string"),
                    self._get_event_schema()
                ).alias("event")
            ).select("event.*")
    
    def create_batch_source(self, path: str):
        """Create batch DataFrame from files."""
        
        return self.spark.read.parquet(path)
    
    def compute_revenue_metrics(self, events_df):
        """
        Compute revenue metrics.
        
        Works identically for batch and streaming DataFrames!
        """
        
        return events_df \
            .filter(col("event_type") == "order.placed") \
            .withWatermark("event_time", "1 hour") \
            .groupBy(
                window("event_time", "1 hour"),
                "payload.category"
            ) \
            .agg(
                sum("payload.total_amount").alias("revenue"),
                count("*").alias("order_count"),
                avg("payload.total_amount").alias("avg_order_value")
            )
    
    def write_streaming(self, df, output_path: str, checkpoint_path: str):
        """Write streaming DataFrame with checkpointing."""
        
        return df.writeStream \
            .format("parquet") \
            .option("path", output_path) \
            .option("checkpointLocation", checkpoint_path) \
            .outputMode("append") \
            .trigger(processingTime="1 minute") \
            .start()
    
    def write_batch(self, df, output_path: str):
        """Write batch DataFrame."""
        
        df.withColumn("date", to_date(col("window.start"))) \
            .write \
            .mode("overwrite") \
            .partitionBy("date") \
            .parquet(output_path)
    
    def _get_event_schema(self):
        """Define event schema."""
        
        return StructType([
            StructField("event_id", StringType()),
            StructField("event_type", StringType()),
            StructField("event_time", TimestampType()),
            StructField("user_id", StringType()),
            StructField("payload", MapType(StringType(), StringType()))
        ])


# =============================================================================
# Usage Examples
# =============================================================================

USAGE_EXAMPLES = """
UNIFIED BATCH + STREAMING

# Same processing logic for both!

processor = UnifiedAnalyticsProcessor(spark)

# STREAMING MODE
stream_df = processor.create_streaming_source("kafka:9092", "events")
revenue = processor.compute_revenue_metrics(stream_df)
query = processor.write_streaming(revenue, "/output/stream", "/checkpoint")

# BATCH MODE  
batch_df = processor.create_batch_source("/data/events")
revenue = processor.compute_revenue_metrics(batch_df)
processor.write_batch(revenue, "/output/batch")


THE POWER:
- Same aggregation code
- Same business logic
- Different execution modes
- Consistent results
"""