Kappa-архитектура на Spark Structured Streaming: стриминг как единый источник правды

# Создание событийного топика с годовым retention (для Kappa)
kafka-topics.sh --create \
  --bootstrap-server kafka:9092 \
  --topic ecommerce.purchases \
  --partitions 12 \           # Параллелизм = количество партиций
  --replication-factor 3 \    # Надёжность: 3 реплики
  --config retention.ms=31536000000 \   # 365 дней
  --config segment.bytes=536870912 \    # Сегмент 512MB (оптимально для долгого хранения)
  --config compression.type=lz4         # Сжатие: lz4 быстрее gzip, лучше чем none

# Создание компактного топика для состояний сущностей
kafka-topics.sh --create \
  --bootstrap-server kafka:9092 \
  --topic ecommerce.user_profiles \
  --partitions 12 \
  --replication-factor 3 \
  --config cleanup.policy=compact \
  --config min.compaction.lag.ms=3600000 \   # Compaction не раньше чем через час
  --config max.compaction.lag.ms=86400000    # Compaction не позже чем через сутки

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

spark = SparkSession.builder \
    .appName("Kappa-Architecture-Pipeline") \
    .config("spark.sql.extensions",
            "org.apache.iceberg.spark.extensions.IcebergSparkSessionExtensions") \
    .config("spark.sql.catalog.lakehouse", "org.apache.iceberg.spark.SparkCatalog") \
    .config("spark.sql.catalog.lakehouse.type", "rest") \
    .config("spark.sql.catalog.lakehouse.uri", "http://catalog:8181") \
    .config("spark.sql.catalog.lakehouse.warehouse", "s3://bucket/warehouse") \
    # RocksDB State Store: гораздо эффективнее default HashMapStateStore
    # для больших stateful операций (деdup, windowed agg)
    .config("spark.sql.streaming.stateStore.providerClass",
            "org.apache.spark.sql.execution.streaming.state.RocksDBStateStoreProvider") \
    # Компактим RocksDB при каждом commit для предотвращения раздувания
    .config("spark.sql.streaming.stateStore.rocksdb.compactOnCommit", "true") \
    .getOrCreate()

event_schema = StructType([
    StructField("event_id",   StringType(),    False),
    StructField("user_id",    StringType(),    False),
    StructField("session_id", StringType(),    True),
    StructField("event_type", StringType(),    False),
    StructField("product_id", StringType(),    True),
    StructField("amount",     DoubleType(),    True),
    StructField("event_ts",   TimestampType(), False),
])

# === РЕЖИМ 1: Начальный Replay (пересчёт истории) ===
# startingOffsets="earliest": читаем Kafka с самого начала
# Используется при:
# - первом запуске нового пайплайна
# - смене бизнес-логики (новая версия V2)
# - восстановлении после потери checkpoint
REPLAY_MODE = True   # Управляется конфигурацией или ENV-переменной

starting_offsets = "earliest" if REPLAY_MODE else "latest"

# Для точного управления можно передать JSON со смещениями по партициям:
# starting_offsets = json.dumps({
#     "ecommerce.purchases": {"0": 0, "1": 0, "2": 0}  # offset 0 в каждой партиции
# })

raw_kafka_stream = (
    spark.readStream
         .format("kafka")
         .option("kafka.bootstrap.servers", "kafka:9092")
         .option("subscribe", "ecommerce.purchases")
         .option("startingOffsets", starting_offsets)
         # В Replay режиме снимаем ограничение: читаем как можно быстрее
         # В production режиме ставим 50000 для защиты от OOM
         .option("maxOffsetsPerTrigger", "500000" if REPLAY_MODE else "50000")
         # failOnDataLoss=false для replay: старые сегменты могут быть удалены,
         # если retention не настроен достаточно долго
         .option("failOnDataLoss", "false" if REPLAY_MODE else "true")
         .load()
         .select(
             F.col("partition"),
             F.col("offset"),
             F.col("timestamp").alias("kafka_ts"),
             F.from_json(F.col("value").cast("string"), event_schema).alias("d")
         )
         .select("partition", "offset", "kafka_ts", "d.*")
)

from pyspark.sql import functions as F

# === Tumbling Window (фиксированное, непересекающееся) ===
# Каждая покупка попадает ровно в один час.
# Результат: выручка за час - обновляется каждые 30 секунд (trigger interval)
tumbling_revenue = (
    raw_kafka_stream
    # ОБЯЗАТЕЛЬНО: watermark ПЕРЕД stateful операцией
    # 10 минут - максимальное ожидаемое опоздание покупок
    .withWatermark("event_ts", "10 minutes")
    .groupBy(
        # window(timeColumn, windowDuration)
        # Размер окна: 1 час. Смещение: нет (окна по UTC-часам: 10:00-11:00, 11:00-12:00)
        F.window("event_ts", "1 hour"),
        "event_type"
    )
    .agg(
        F.sum("amount").alias("revenue"),
        F.count("*").alias("tx_count"),
        F.approx_count_distinct("user_id", rsd=0.01).alias("unique_buyers")
    )
    # Разворачиваем структуру window {start, end} в отдельные колонки
    .select(
        F.col("window.start").alias("hour_start"),
        F.col("window.end").alias("hour_end"),
        "event_type", "revenue", "tx_count", "unique_buyers"
    )
)

# === Sliding Window (скользящее, пересекающееся) ===
# Окно 15 минут, смещение 5 минут: 10:00-10:15, 10:05-10:20, 10:10-10:25...
# Каждое событие попадает в 3 окна (15/5 = 3)
# Используется для: скользящее среднее, moving KPI

sliding_activity = (
    raw_kafka_stream
    .withWatermark("event_ts", "20 minutes")   # Больше delay = больше late data принимаем
    .groupBy(
        # window(timeColumn, windowDuration, slideDuration)
        # Окно 15 минут, обновляется каждые 5 минут
        F.window("event_ts", "15 minutes", "5 minutes"),
        "event_type"
    )
    .agg(
        F.count("*").alias("event_count"),
        F.sum("amount").alias("window_revenue")
    )
    .select(
        F.col("window.start").alias("window_start"),
        F.col("window.end").alias("window_end"),
        "event_type", "event_count", "window_revenue"
    )
)

from pyspark.sql.streaming.state import GroupState, GroupStateTimeout
from pyspark.sql.types import (StructType, StructField,
                                StringType, TimestampType, LongType, IntegerType)

# Схема состояния сессии (хранится в State Store для каждого user_id)
session_state_schema = StructType([
    StructField("session_id",     StringType(),    False),
    StructField("user_id",        StringType(),    False),
    StructField("session_start",  TimestampType(), False),
    StructField("last_event_ts",  TimestampType(), False),
    StructField("event_count",    IntegerType(),   False),
    StructField("total_amount",   DoubleType(),    False),
])

# Схема выходного события (когда сессия завершается)
session_output_schema = StructType([
    StructField("session_id",    StringType(),    False),
    StructField("user_id",       StringType(),    False),
    StructField("session_start", TimestampType(), False),
    StructField("session_end",   TimestampType(), False),
    StructField("duration_sec",  LongType(),      False),
    StructField("event_count",   IntegerType(),   False),
    StructField("total_amount",  DoubleType(),    False),
])

import uuid
from datetime import datetime, timedelta

SESSION_TIMEOUT_MINUTES = 30   # Сессия завершается через 30 минут тишины

def update_session_state(user_id: str, events, state: GroupState):
    """
    flatMapGroupsWithState функция: вызывается для каждого user_id
    при каждом новом микро-батче.

    Логика:
    - Если есть открытая сессия и новые события укладываются в timeout - расширяем сессию
    - Если timeout истёк - закрываем сессию, возвращаем её как результат, открываем новую
    - При истечении GroupState timeout - тоже закрываем сессию
    """
    SESSION_TIMEOUT = timedelta(minutes=SESSION_TIMEOUT_MINUTES)
    sorted_events = sorted(events, key=lambda e: e.event_ts)

    # Обработка timeout (GroupState timeout истёк - пользователь молчал долго)
    if state.hasTimedOut:
        if state.exists:
            # Сессия была открыта - закрываем её и возвращаем как результат
            s = state.get
            yield (s.session_id, s.user_id, s.session_start,
                   s.last_event_ts, int((s.last_event_ts - s.session_start).total_seconds()),
                   s.event_count, s.total_amount)
        state.remove()
        return

    completed_sessions = []
    current_state = state.get if state.exists else None

    for event in sorted_events:
        if current_state is None:
            # Нет открытой сессии - создаём новую
            current_state = {
                "session_id": str(uuid.uuid4()),
                "user_id": user_id,
                "session_start": event.event_ts,
                "last_event_ts": event.event_ts,
                "event_count": 1,
                "total_amount": event.amount or 0.0,
            }
        else:
            gap = event.event_ts - current_state["last_event_ts"]
            if gap > SESSION_TIMEOUT:
                # Gap слишком большой - закрываем текущую сессию
                completed_sessions.append(current_state)
                # Открываем новую сессию
                current_state = {
                    "session_id": str(uuid.uuid4()),
                    "user_id": user_id,
                    "session_start": event.event_ts,
                    "last_event_ts": event.event_ts,
                    "event_count": 1,
                    "total_amount": event.amount or 0.0,
                }
            else:
                # Продолжаем текущую сессию
                current_state["last_event_ts"] = event.event_ts
                current_state["event_count"] += 1
                current_state["total_amount"] += event.amount or 0.0

    # Сохраняем открытую сессию в State Store
    if current_state:
        state.update((
            current_state["session_id"],
            current_state["user_id"],
            current_state["session_start"],
            current_state["last_event_ts"],
            current_state["event_count"],
            current_state["total_amount"],
        ))
        # Устанавливаем timeout: если 30 минут не будет событий - закроем сессию
        state.setTimeoutDuration(f"{SESSION_TIMEOUT_MINUTES} minutes")

    # Возвращаем все завершённые сессии этого микро-батча
    for s in completed_sessions:
        yield (s["session_id"], s["user_id"], s["session_start"],
               s["last_event_ts"],
               int((s["last_event_ts"] - s["session_start"]).total_seconds()),
               s["event_count"], s["total_amount"])


# Применяем сессионизацию
sessions_stream = (
    raw_kafka_stream
    .withWatermark("event_ts", "30 minutes")
    .groupBy("user_id")
    .applyInPandasWithState(
        update_session_state,
        outputStructType=session_output_schema,
        stateStructType=session_state_schema,
        outputMode="append",
        timeoutConf=GroupStateTimeout.EventTimeTimeout
    )
)

# === Этап 1: Stateless Replay (быстрый, не требует State Store) ===
# Читает Kafka → применяет чистые трансформации → пишет в промежуточный Bronze Iceberg
# Скорость: ограничена только Kafka throughput и CPU на десериализацию
# Параллелизм: partitions * executors

def stateless_transform(df):
    """Чистые, безstate трансформации. Детерминированы, быстры."""
    return (
        df
        .filter(F.col("amount") > 0)
        .withColumn("net_amount",
            F.when(F.col("discount_pct") <= 50,
                   F.col("amount") * (1 - F.col("discount_pct") / 100))
            .otherwise(F.lit(0.0))
        )
        .withColumn("event_date", F.to_date("event_ts"))
        .withColumn("event_hour", F.date_trunc("hour", F.col("event_ts")))
    )

bronze_replay_query = (
    raw_kafka_stream                       # startingOffsets="earliest"
    .transform(stateless_transform)        # Stateless: быстро
    .writeStream
    .trigger(processingTime="10 seconds")  # Частый триггер для быстрого replay
    .outputMode("append")
    .format("iceberg")
    .option("checkpointLocation", "s3://bucket/checkpoints/replay-bronze-v2")
    .toTable("lakehouse.bronze.purchases_v2")
)

# Запускаем Этап 1, ждём завершения (когда job догонит current offsets)
bronze_replay_query.awaitTermination(timeout=86400)  # timeout 24 часа

# === Этап 2: Stateful Aggregation (медленнее, но работает с уже загруженными данными) ===
# Читает из Bronze Iceberg (batch!), строит stateful агрегаты
# Преимущество: Spark Batch параллелизм не ограничен числом Kafka-партиций
silver_from_bronze = (
    spark.read                           # BATCH read, не streaming!
         .format("iceberg")
         .load("lakehouse.bronze.purchases_v2")
         .transform(stateless_transform)  # Применяем ту же логику
)

# Stateful Window Aggregation - теперь это просто Spark Batch
hourly_revenue = (
    silver_from_bronze
    .groupBy(
        "event_hour",
        "event_type"
    )
    .agg(
        F.sum("net_amount").alias("net_revenue"),
        F.count("*").alias("tx_count"),
        F.countDistinct("user_id").alias("unique_buyers")
    )
)

hourly_revenue.writeTo("lakehouse.gold.revenue_by_hour_v2").overwritePartitions()

def kappa_foreachbatch_with_dedup(batch_df, batch_id: int) -> None:
    """
    Идемпотентная запись с дедупликацией при Replay.
    Использует MERGE INTO для обнаружения уже существующих event_id.
    """
    if batch_df.isEmpty():
        return

    # Дедупликация внутри батча (могут быть дубли из Kafka при at-least-once)
    deduped = batch_df.dropDuplicates(["event_id"])
    deduped.createOrReplaceTempView("incoming")

    # MERGE INTO Bronze: если event_id уже есть - пропускаем (не обновляем!)
    # При Replay это защищает от двойного счёта
    # WHEN MATCHED: ничего не делаем (событие уже записано корректно)
    # WHEN NOT MATCHED: вставляем новое событие
    spark.sql("""
        MERGE INTO lakehouse.bronze.purchases_v2 AS target
        USING incoming AS source
        ON target.event_id = source.event_id
        WHEN NOT MATCHED THEN INSERT *
    """)
    # Нет WHEN MATCHED: существующие записи неизменны (иммутабельность Bronze)

    # Альтернатива для чистого Bronze (только APPEND):
    # Держите отдельную таблицу processed_event_ids и фильтруйте перед записью
    # Это дороже, но не требует MERGE INTO для Bronze

# === Стандартная стриминговая запись в Iceberg ===
# Spark + Iceberg нативно поддерживают exactly-once через WAL Iceberg
purchase_stream_query = (
    raw_kafka_stream
    .transform(stateless_transform)
    .writeStream
    .trigger(processingTime="30 seconds")
    .outputMode("append")
    .format("iceberg")
    # checkpoint хранит: последние обработанные Kafka offsets + Iceberg commit ID
    # Вместе они гарантируют exactly-once при перезапуске
    .option("checkpointLocation", "s3://bucket/checkpoints/iceberg-stream-v1")
    # fanout_enabled: разрешает запись в несколько партиций за один task
    # Нужен при стриминге, когда один executor пишет в несколько partition key
    .option("fanout-enabled", "true")
    .toTable("lakehouse.bronze.purchases")
)

# Схема потока показов рекламы
impression_schema = StructType([
    StructField("impression_id", StringType(), False),
    StructField("ad_id",         StringType(), False),
    StructField("user_id",       StringType(), False),
    StructField("campaign_id",   StringType(), True),
    StructField("shown_at",      TimestampType(), False),
])

# Схема потока кликов
click_schema = StructType([
    StructField("click_id",  StringType(),    False),
    StructField("ad_id",     StringType(),    False),
    StructField("user_id",   StringType(),    False),
    StructField("clicked_at", TimestampType(), False),
])

# Читаем поток показов
impressions_stream = (
    spark.readStream
         .format("kafka")
         .option("kafka.bootstrap.servers", "kafka:9092")
         .option("subscribe", "ad_impressions")
         .option("startingOffsets", "latest")
         .load()
         .select(F.from_json(F.col("value").cast("string"), impression_schema).alias("d"))
         .select("d.*")
         # Watermark: ждём опоздавшие показы до 2 часов
         .withWatermark("shown_at", "2 hours")
)

# Читаем поток кликов
clicks_stream = (
    spark.readStream
         .format("kafka")
         .option("kafka.bootstrap.servers", "kafka:9092")
         .option("subscribe", "ad_clicks")
         .option("startingOffsets", "latest")
         .load()
         .select(F.from_json(F.col("value").cast("string"), click_schema).alias("d"))
         .select("d.*")
         # Watermark: ждём опоздавшие клики до 2 часов
         .withWatermark("clicked_at", "2 hours")
)

# Stream-to-Stream Join
# Inner Join: выдаём только пары показ+клик (не показы без клика)
# State: Spark держит в памяти показы и клики до истечения watermark
ad_performance = (
    impressions_stream.alias("imp")
    .join(
        clicks_stream.alias("clk"),
        on=[
            # Условие JOIN: тот же ad_id, тот же пользователь
            F.col("imp.ad_id")   == F.col("clk.ad_id"),
            F.col("imp.user_id") == F.col("clk.user_id"),
            # Временное ограничение: клик должен быть после показа и не позже чем через 2 часа
            # Это критически важно: без временного ограничения State Store бесконечно растёт
            F.col("clk.clicked_at") >= F.col("imp.shown_at"),
            F.col("clk.clicked_at") <= F.col("imp.shown_at") + F.expr("INTERVAL 2 HOURS")
        ],
        how="inner"
    )
    .select(
        F.col("imp.impression_id"),
        F.col("imp.ad_id"),
        F.col("imp.user_id"),
        F.col("imp.campaign_id"),
        F.col("imp.shown_at"),
        F.col("clk.click_id"),
        F.col("clk.clicked_at"),
        # Время от показа до клика
        (F.col("clk.clicked_at").cast("long") - F.col("imp.shown_at").cast("long"))
            .alias("click_latency_sec")
    )
)

# Записываем в Iceberg: каждый клик с информацией о показе
(
    ad_performance
    .writeStream
    .trigger(processingTime="1 minute")
    .outputMode("append")
    .format("iceberg")
    .option("checkpointLocation", "s3://bucket/checkpoints/ad-performance")
    .toTable("lakehouse.gold.ad_click_performance")
)

import time

def monitor_streaming_job(query, check_interval_sec: int = 60):
    """
    Мониторинг здоровья Streaming Job в production.
    Логирует ключевые метрики и поднимает алерты при аномалиях.

    query: активный StreamingQuery (результат .start())
    check_interval_sec: интервал проверки метрик
    """
    while query.isActive:
        progress = query.lastProgress

        if progress is None:
            time.sleep(check_interval_sec)
            continue

        # === Базовые метрики производительности ===
        input_rate   = progress.get("inputRowsPerSecond", 0)
        process_rate = progress.get("processedRowsPerSecond", 0)
        batch_dur_ms = progress.get("batchDuration", 0)
        num_rows     = progress.get("numInputRows", 0)

        print(f"\n=== Streaming Health Check ===")
        print(f"Input Rate:     {input_rate:.0f} строк/сек")
        print(f"Process Rate:   {process_rate:.0f} строк/сек")
        print(f"Batch Duration: {batch_dur_ms}ms")
        print(f"Rows in batch:  {num_rows}")

        # === Алерт: Consumer Lag растёт ===
        if input_rate > 0 and process_rate < input_rate * 0.9:
            lag_ratio = input_rate / process_rate
            print(f"WARN: Process Rate отстаёт от Input Rate в {lag_ratio:.1f}x!")
            print("      Возможная причина: медленный foreachBatch, медленный S3, GC паузы")

        # === Алерт: Батч слишком долгий ===
        # Если батч длится дольше trigger interval - следующий батч задерживается
        # Это первый признак растущего Consumer Lag
        trigger_interval_ms = 30_000  # наш trigger: 30 секунд
        if batch_dur_ms > trigger_interval_ms * 1.5:
            print(f"WARN: Batch Duration {batch_dur_ms}ms > trigger * 1.5 ({trigger_interval_ms * 1.5}ms)")
            print("      Оптимизируйте foreachBatch или увеличьте trigger interval")

        # === Метрики State Store ===
        state_operators = progress.get("stateOperators", [])
        for i, so in enumerate(state_operators):
            rows_total   = so.get("numRowsTotal", 0)
            memory_bytes = so.get("memoryUsedBytes", 0)
            mem_mb = memory_bytes / 1024 / 1024

            print(f"\nState Operator [{i}]:")
            print(f"  Строк в state:  {rows_total:,}")
            print(f"  Память:         {mem_mb:.1f} MB")

            # Алерт: State Store занимает много памяти
            executor_heap_mb = 4096   # executor.memory = 4GB
            if mem_mb > executor_heap_mb * 0.7:
                print(f"  КРИТИЧНО: State Store > 70% heap executor!")
                print(f"  Действие: уменьшить watermark_delay или добавить executor'ы")

        # === Watermark Gap ===
        event_time_watermark = progress.get("eventTime", {}).get("watermark", "N/A")
        print(f"\nWatermark: {event_time_watermark}")

        time.sleep(check_interval_sec)


# Запускаем мониторинг в отдельном потоке
import threading
monitor_thread = threading.Thread(
    target=monitor_streaming_job,
    args=(purchase_stream_query, 60),
    daemon=True
)
monitor_thread.start()

# АНТИПАТТЕРН: Kafka с коротким retention для Kappa Source of Truth
# 7 дней retention = нельзя сделать Replay за месяц
# Бизнес решает пересчитать данные за прошлый квартал - невозможно

# СЛЕДСТВИЕ: приходится держать отдельный S3 архив событий
# Это фактически возвращает к Lambda Architecture (S3 = Master Dataset)

# ПРАВИЛЬНО: retention должен покрывать максимальный горизонт Replay
# Если пересчёты нужны за год - retention минимум 365 дней
# Стоимость хранения 1TB в Kafka ≈ $30-100/месяц (зависит от репликации)
# vs $23/TB/месяц в S3 Standard
# Для больших объёмов: Kafka → S3 через MirrorMaker2 или Kafka S3 Sink
# Тогда Replay читает S3, а не Kafka - снимает ограничение по партициям

# АНТИПАТТЕРН: неидемпотентный sink при Replay
# При Replay пайплайн читает те же события снова
# Если sink не идемпотентен - данные дублируются

def bad_foreachbatch(batch_df, batch_id):
    # ОШИБКА: простой INSERT без дедупликации
    # При Replay те же event_id записываются второй раз
    batch_df.write.format("iceberg").mode("append").saveAsTable("bronze.events")
    # Результат: 2x дубликаты в таблице

def good_foreachbatch(batch_df, batch_id):
    # ПРАВИЛЬНО: MERGE INTO с дедупликацией по event_id
    # При Replay: WHEN MATCHED - пропускаем, WHEN NOT MATCHED - вставляем
    batch_df.createOrReplaceTempView("incoming")
    spark.sql("""
        MERGE INTO lakehouse.bronze.events AS t
        USING (SELECT DISTINCT * FROM incoming) AS s
        ON t.event_id = s.event_id
        WHEN NOT MATCHED THEN INSERT *
    """)
    # Результат: идемпотентно, Replay не создаёт дубликатов

# АНТИПАТТЕРН: stateful операция без watermark
# Spark хранит state для КАЖДОГО ключа и КАЖДОГО окна навсегда

bad_agg = (
    stream_df
    # НЕТ withWatermark → нет очистки State Store
    .groupBy(F.window("event_ts", "1 hour"), "user_id")
    .agg(F.sum("amount"))
    # После 7 дней: State Store хранит 7 * 24 = 168 часовых окон × N пользователей
    # После месяца: 720 окон × N пользователей → OOM
)

# ПРАВИЛЬНО: watermark освобождает State Store по мере закрытия окон
good_agg = (
    stream_df
    .withWatermark("event_ts", "30 minutes")    # Ждём опоздавших 30 минут
    .groupBy(F.window("event_ts", "1 hour"), "user_id")
    .agg(F.sum("amount"))
    # После watermark > window.end: Spark удаляет окно из State Store
    # Стабильный объём state ≈ (watermark_delay / trigger_interval) × active_users_in_window
)

# АНТИПАТТЕРН: несколько стриминговых джобов читают один топик
# с разными checkpoint - нет гарантии согласованности

query_revenue = (stream_df.writeStream
    .option("checkpointLocation", "s3://bucket/checkpoints/revenue")
    .toTable("gold.revenue"))

query_users = (stream_df.writeStream
    .option("checkpointLocation", "s3://bucket/checkpoints/users")
    .toTable("gold.users"))

# Проблема: при сбое revenue job откатится к своему checkpoint (offset 1000),
# а users job продолжит с offset 1100.
# Таблицы revenue и users теперь рассогласованы по времени.

# ПРАВИЛЬНО: один foreachBatch пишет в несколько таблиц атомарно
def unified_foreachbatch(batch_df, batch_id):
    # Один checkpoint, один микро-батч, несколько целевых таблиц
    batch_df.createOrReplaceTempView("batch")
    spark.sql("INSERT INTO gold.revenue SELECT ... FROM batch")
    spark.sql("INSERT INTO gold.users SELECT ... FROM batch")
    # Если любая запись падает - весь batch_id откатывается

unified_query = (
    stream_df.writeStream
             .foreachBatch(unified_foreachbatch)
             .option("checkpointLocation", "s3://bucket/checkpoints/unified")
             .start()
)

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

spark = SparkSession.builder \
    .appName("Kappa-ECommerce-Pipeline-V2") \
    .config("spark.sql.extensions",
            "org.apache.iceberg.spark.extensions.IcebergSparkSessionExtensions") \
    .config("spark.sql.catalog.lakehouse", "org.apache.iceberg.spark.SparkCatalog") \
    .config("spark.sql.catalog.lakehouse.type", "rest") \
    .config("spark.sql.catalog.lakehouse.uri", "http://catalog:8181") \
    .config("spark.sql.catalog.lakehouse.warehouse", "s3://datalake/warehouse") \
    .config("spark.sql.streaming.stateStore.providerClass",
            "org.apache.spark.sql.execution.streaming.state.RocksDBStateStoreProvider") \
    .config("spark.sql.streaming.stateStore.rocksdb.compactOnCommit", "true") \
    # Оптимизация: не записывать лишние метаданные при каждом микро-батче
    .config("spark.sql.streaming.minBatchesToRetain", "2") \
    .getOrCreate()

# === Схема события ===
event_schema = StructType([
    StructField("event_id",    StringType(),    False),
    StructField("user_id",     StringType(),    False),
    StructField("session_id",  StringType(),    True),
    StructField("event_type",  StringType(),    False),
    StructField("product_id",  StringType(),    True),
    StructField("category",    StringType(),    True),
    StructField("amount",      DoubleType(),    True),
    StructField("discount_pct", DoubleType(),   True),
    StructField("event_ts",    TimestampType(), False),
    StructField("device_type", StringType(),    True),
])

# === Чтение из Kafka ===
# PIPELINE_VERSION управляет: latest для нового деплоя, earliest для replay
PIPELINE_VERSION = "V2"
IS_REPLAY = True   # True = читаем историю, False = только real-time

raw_stream = (
    spark.readStream
         .format("kafka")
         .option("kafka.bootstrap.servers", "kafka:9092")
         .option("subscribe", "ecommerce.events")
         .option("startingOffsets", "earliest" if IS_REPLAY else "latest")
         # При Replay: большой батч для скорости. При real-time: защита от OOM
         .option("maxOffsetsPerTrigger", "500000" if IS_REPLAY else "50000")
         .option("failOnDataLoss", "false")
         .load()
         .select(
             F.col("partition").alias("kafka_partition"),
             F.col("offset").alias("kafka_offset"),
             F.col("timestamp").alias("kafka_ts"),
             F.from_json(F.col("value").cast("string"), event_schema).alias("d")
         )
         .select("kafka_partition", "kafka_offset", "kafka_ts", "d.*")
)


# === Единая функция трансформации (Kappa принцип) ===
def apply_business_rules_v2(df):
    """
    V2 бизнес-логика. Отличие от V1: добавлен category_revenue_share.
    Используется ОДИНАКОВО при Replay и при real-time processing.
    Детерминирована: результат зависит только от event_ts (не от current_timestamp).
    """
    return (
        df
        # Фильтрация невалидных записей
        .filter(F.col("amount").isNull() | (F.col("amount") >= 0))
        # V2 правило: скидки > 50% не учитываются
        .withColumn(
            "net_amount",
            F.when(
                (F.col("event_type") == "purchase") &
                (F.col("discount_pct") <= 50),
                F.col("amount") * (1 - F.col("discount_pct") / 100)
            ).when(
                F.col("event_type") == "purchase",
                F.lit(0.0)   # Скидка > 50%: не считаем
            ).otherwise(F.lit(0.0))
        )
        # НОВОЕ в V2: категория вклада в выручку
        .withColumn(
            "revenue_tier",
            F.when(F.col("net_amount") == 0,     "no_revenue")
             .when(F.col("net_amount") < 1000,   "micro")
             .when(F.col("net_amount") < 10000,  "mid")
             .otherwise("large")
        )
        # Производные временны́е поля
        .withColumn("event_date", F.to_date("event_ts"))
        .withColumn("event_hour", F.date_trunc("hour", F.col("event_ts")))
        .withColumn("event_minute", F.date_trunc("minute", F.col("event_ts")))
    )


# === foreachBatch: единый обработчик ===
def kappa_main_batch(batch_df, batch_id: int) -> None:
    """
    Центральный foreachBatch для Kappa Pipeline.
    Обрабатывает каждый микро-батч: Bronze → Silver → Gold.
    Идемпотентен: повторный вызов с теми же данными даёт тот же результат.
    """
    if batch_df.isEmpty():
        return

    # Дедупликация: защита от Kafka at-least-once и повторных foreachBatch вызовов
    deduped = batch_df.dropDuplicates(["event_id"]).cache()
    record_count = deduped.count()

    if record_count == 0:
        deduped.unpersist()
        return

    print(f"[Pipeline-{PIPELINE_VERSION} | Batch {batch_id}] {record_count} событий")

    # Применяем V2 бизнес-правила
    enriched = apply_business_rules_v2(deduped)
    enriched.cache()

    # --- Bronze: идемпотентная запись (MERGE IN, игнорируем дубли по event_id) ---
    enriched.createOrReplaceTempView("incoming_events")

    spark.sql(f"""
        MERGE INTO lakehouse.bronze.events AS t
        USING (SELECT * FROM incoming_events) AS s
        ON t.event_id = s.event_id
        WHEN NOT MATCHED THEN INSERT *
    """)

    # --- Silver: текущие агрегаты по пользователям (MERGE INTO, обновляем) ---
    purchases = enriched.filter(F.col("event_type") == "purchase")

    if not purchases.isEmpty():
        purchases.createOrReplaceTempView("incoming_purchases")

        spark.sql("""
            MERGE INTO lakehouse.silver.user_purchase_summary AS t
            USING (
                SELECT
                    user_id,
                    SUM(net_amount)   AS batch_revenue,
                    COUNT(*)          AS batch_count,
                    MAX(event_ts)     AS last_purchase_ts,
                    MAX(revenue_tier) AS max_tier
                FROM incoming_purchases
                GROUP BY user_id
            ) AS s
            ON t.user_id = s.user_id
            WHEN MATCHED THEN UPDATE SET
                t.total_revenue   = t.total_revenue + s.batch_revenue,
                t.total_purchases = t.total_purchases + s.batch_count,
                t.last_purchase_ts = s.last_purchase_ts,
                t.updated_at      = current_timestamp()
            WHEN NOT MATCHED THEN INSERT
                (user_id, total_revenue, total_purchases,
                 last_purchase_ts, created_at, updated_at)
            VALUES (s.user_id, s.batch_revenue, s.batch_count,
                    s.last_purchase_ts, current_timestamp(), current_timestamp())
        """)

    # --- Gold: 1-минутные агрегаты для live дашборда ---
    gold_agg = (
        enriched
        .groupBy("event_minute", "event_type", "revenue_tier", "device_type")
        .agg(
            F.sum("net_amount").alias("net_revenue"),
            F.count("*").alias("event_count"),
            F.approx_count_distinct("user_id", rsd=0.02).alias("unique_users"),
            F.approx_count_distinct("session_id", rsd=0.02).alias("unique_sessions")
        )
    )

    # Append: Gold-таблица читается через Serving View с ограничением по времени
    # Компакция удалит старые мелкие файлы
    (
        gold_agg
        .writeTo("lakehouse.gold.realtime_revenue")
        .append()
    )

    # --- Фоновая компакция Bronze каждые 120 батчей (~1 час) ---
    if batch_id % 120 == 0:
        try:
            spark.sql("""
                CALL lakehouse.system.rewrite_data_files(
                    table => 'lakehouse.bronze.events',
                    where => "event_date = current_date()",
                    options => map(
                        'target-file-size-bytes', '268435456',
                        'max-concurrent-file-group-rewrites', '6',
                        'partial-progress.enabled', 'true'
                    )
                )
            """)
            print(f"[Batch {batch_id}] Компакция Bronze завершена")
        except Exception as ex:
            print(f"[Batch {batch_id}] Компакция пропущена: {ex}")

    # Очищаем кеш
    enriched.unpersist()
    deduped.unpersist()


# === Запуск основного Kappa Pipeline ===
# При IS_REPLAY=True: читает историю 90 дней, затем переходит в real-time
# При IS_REPLAY=False: читает только новые события

trigger_config = {"processingTime": "10 seconds"} if IS_REPLAY else {"processingTime": "30 seconds"}

main_query = (
    raw_stream
    .writeStream
    .trigger(**trigger_config)
    .foreachBatch(kappa_main_batch)
    .option("checkpointLocation",
            f"s3://bucket/checkpoints/kappa-main-{PIPELINE_VERSION.lower()}")
    .start()
)

print(f"Kappa Pipeline {PIPELINE_VERSION} запущен")
print(f"Режим: {'Replay (история + real-time)' if IS_REPLAY else 'Real-time'}")
print(f"Checkpoint: s3://bucket/checkpoints/kappa-main-{PIPELINE_VERSION.lower()}")

# === Алгоритм migration: V1 → V2 без остановки сервиса ===

# Шаг 1: V2 запускается с IS_REPLAY=True, пишет в gold.realtime_revenue_v2
# Параллельно V1 продолжает работать: gold.realtime_revenue_v1

# Шаг 2: Ждём, пока V2 «догонит» V2 до real-time
# Мониторим отставание через query.lastProgress
def wait_for_catchup(query_v2, max_lag_sec: int = 60):
    """Ждём, пока V2 догонит real-time поток."""
    import time
    while True:
        progress = query_v2.lastProgress
        if progress:
            # watermarkGap: разница между последним event_ts и watermark
            # Когда V2 в real-time: watermarkGap ≈ watermark_delay
            batch_dur = progress.get("batchDuration", 999999)
            trigger_ms = 30_000   # наш trigger

            # Если batch duration < trigger: V2 обрабатывает меньше данных чем приходит
            # Это означает: V2 догнал или почти догнал real-time
            if batch_dur < trigger_ms:
                print("V2 догнал real-time поток. Готов к переключению.")
                break

        print(f"V2 ещё догоняет историю... Ждём {max_lag_sec}s")
        time.sleep(max_lag_sec)


wait_for_catchup(main_query)

# Шаг 3: Атомарное переключение Serving View
# Аналитики читают через view gold.realtime_serving
# Переключение мгновенное, нет downtime
spark.sql("""
    CREATE OR REPLACE VIEW lakehouse.gold.realtime_serving AS
    SELECT * FROM lakehouse.gold.realtime_revenue_v2
    WHERE event_minute >= current_timestamp() - INTERVAL 48 HOURS
""")

print("Serving View переключён на V2")
print("Аналитики теперь видят данные из V2 пайплайна")

# Шаг 4: Останавливаем V1 (после переключения View)
# query_v1.stop()
# V1 checkpoint можно удалить через 7 дней (если Replay V2 подтверждён)