Lambda-архитектура в Lakehouse: интеграция Batch и Real-time потоков на Spark

# ПЛОХО: результат зависит от времени запуска
df.withColumn("is_recent", F.col("event_ts") > F.current_timestamp() - F.expr("INTERVAL 30 DAYS"))

# ХОРОШО: детерминированно, зависит только от данных
df.withColumn("is_recent", F.col("event_ts") > F.lit("2026-04-23").cast("timestamp"))

# ПЛОХО: случайная выборка меняется при каждом запуске
df.sample(fraction=0.1, seed=None)

# ХОРОШО: фиксированный seed гарантирует воспроизводимость
df.sample(fraction=0.1, seed=42)

# ПЛОХО: зависимость от внешнего состояния (текущего курса валют)
df.withColumn("usd_amount", F.col("rub_amount") / get_current_exchange_rate())

# ХОРОШО: исторический курс валют читается из справочника
exchange_rates = spark.table("lakehouse.reference.exchange_rates")
df.join(exchange_rates, on=["date", "currency"], how="left")

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

spark = SparkSession.builder \
    .appName("Lambda-SpeedLayer") \
    .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", "hadoop") \
    .config("spark.sql.catalog.lakehouse.warehouse", "s3://my-bucket/warehouse") \
    .getOrCreate()

# Схема события из Kafka
# Kafka хранит сообщения как байтовые строки: value - JSON с бизнес-данными
event_schema = StructType([
    StructField("event_id",    StringType(),    nullable=False),
    StructField("user_id",     StringType(),    nullable=False),
    StructField("event_type",  StringType(),    nullable=False),
    StructField("amount",      LongType(),      nullable=True),
    StructField("event_ts",    TimestampType(), nullable=False),
])

# Читаем из Kafka
# startingOffsets="latest": при первом запуске начинаем с конца топика.
# Это правильно для Speed Layer: нам нужны только свежие события,
# история уже обработана Batch Layer.
# При перезапуске checkpoint автоматически восстановит позицию.
raw_kafka_stream = (
    spark.readStream
         .format("kafka")
         .option("kafka.bootstrap.servers", "kafka:9092")
         .option("subscribe",               "ecommerce.events")
         .option("startingOffsets",         "latest")
         # Максимальное количество сообщений за один микро-батч.
         # Защищает от ситуации, когда consumer lag вырос и один микро-батч
         # пытается обработать миллионы сообщений, вызывая OOM.
         .option("maxOffsetsPerTrigger",    "50000")
         # Если топик не существует - упасть, а не ждать молча
         .option("failOnDataLoss",          "true")
         .load()
)

# Десериализуем JSON из поля value
# Kafka хранит value как BINARY, поэтому CAST к STRING обязателен
parsed_stream = (
    raw_kafka_stream
    .select(
        # Метаданные Kafka: всегда полезны для debugging
        F.col("topic"),
        F.col("partition"),
        F.col("offset"),
        F.col("timestamp").alias("kafka_ts"),      # Время записи в Kafka
        # Десериализуем бизнес-данные
        F.from_json(
            F.col("value").cast("string"),
            event_schema
        ).alias("data")
    )
    .select(
        "topic", "partition", "offset", "kafka_ts",
        "data.*"   # Разворачиваем структуру: event_id, user_id, event_type, amount, event_ts
    )
)

# Вариант 1: Fixed Interval - триггер каждые N секунд
# Используется: когда нужна предсказуемая задержка (например, обновление дашборда раз в 30 сек)
# Поведение: если обработка предыдущего батча заняла больше N секунд,
#            следующий запускается сразу после завершения, без ожидания
query_fixed = (
    parsed_stream.writeStream
                 .trigger(processingTime="30 seconds")
                 .format("iceberg")
                 .outputMode("append")
                 .toTable("lakehouse.silver.speed_layer_events")
)

# Вариант 2: Once - запустить один микро-батч и остановиться
# Используется: в Lambda Architecture как "pseudo-streaming" —
# запускается по cron каждые 5 минут, обрабатывает накопившееся, завершается.
# Преимущество: не нужен постоянно работающий Spark-процесс.
# Недостаток: задержка = интервал cron (5-15 минут).
query_once = (
    parsed_stream.writeStream
                 .trigger(once=True)
                 .format("iceberg")
                 .outputMode("append")
                 .toTable("lakehouse.silver.speed_layer_events")
)

# Вариант 3: AvailableNow - обработать всё доступное сейчас, затем остановиться
# Spark 3.3+: аналог Once, но разбивает работу на несколько микро-батчей,
# что даёт более эффективное использование ресурсов при большом lag.
query_available_now = (
    parsed_stream.writeStream
                 .trigger(availableNow=True)
                 .format("iceberg")
                 .outputMode("append")
                 .toTable("lakehouse.silver.speed_layer_events")
)

# Watermark: ждём опоздавшие события до 10 минут
# Это означает: если последнее событие имело event_ts = 10:00,
# мы ещё принимаем события с event_ts >= 09:50.
# Событие с event_ts = 09:40 считается опоздавшим и игнорируется.
windowed_stream = (
    parsed_stream
    .withWatermark("event_ts", "10 minutes")   # Ждём опоздавших до 10 минут
    .groupBy(
        F.window("event_ts", "1 minute"),       # Группируем по 1-минутным окнам
        "event_type"
    )
    .agg(
        F.count("*").alias("event_count"),
        F.sum("amount").alias("total_amount"),
        F.approx_count_distinct("user_id").alias("unique_users")
    )
)

# Записываем агрегированные Realtime Views
query = (
    windowed_stream
    .writeStream
    .trigger(processingTime="30 seconds")
    .outputMode("update")            # update: обновляем только изменившиеся строки окон
    .format("iceberg")
    .option("checkpointLocation", "s3://my-bucket/checkpoints/speed-layer-agg")
    .toTable("lakehouse.gold.realtime_views")
)

query.awaitTermination()

-- Представление Serving Layer, объединяющее Batch и Speed данные
-- Ключевой принцип: Speed Layer покрывает только периоды ПОЗЖЕ последнего Batch run
CREATE OR REPLACE VIEW lakehouse.gold.serving_revenue_by_hour AS

WITH batch_coverage AS (
    -- Определяем, до какого момента покрыт Batch Layer
    -- Читаем метаданные последнего Batch View
    SELECT MAX(processed_up_to_ts) as batch_max_ts
    FROM lakehouse.gold.batch_view_metadata
),

batch_data AS (
    -- Данные из Batch Layer: полностью точные, за все завершённые периоды
    SELECT
        hour_ts,
        event_type,
        total_amount,
        event_count,
        'batch' as source_layer
    FROM lakehouse.gold.batch_revenue_by_hour
    WHERE hour_ts < (SELECT batch_max_ts FROM batch_coverage)
),

speed_data AS (
    -- Данные из Speed Layer: только за период ПОСЛЕ последнего Batch run
    -- Это предотвращает двойной счёт
    SELECT
        window.start as hour_ts,
        event_type,
        total_amount,
        event_count,
        'speed' as source_layer
    FROM lakehouse.gold.realtime_views
    WHERE window.start >= (SELECT batch_max_ts FROM batch_coverage)
)

-- UNION ALL без дублирования: batch и speed покрывают непересекающиеся временные периоды
SELECT * FROM batch_data
UNION ALL
SELECT * FROM speed_data
ORDER BY hour_ts DESC;

def serve_unified_query(metric: str, start_date: str, end_date: str) -> None:
    """
    Пример Serving Layer: возвращает метрику за период,
    объединяя Batch Views и Realtime Views без дублирования.

    metric: имя метрики (например, 'revenue', 'orders_count')
    start_date: начало периода, формат 'YYYY-MM-DD'
    end_date: конец периода, формат 'YYYY-MM-DD'
    """
    # 1. Определяем временную границу Batch Layer
    # Читаем метаданные, которые Batch-джоб обновляет при каждом запуске
    batch_metadata = spark.table("lakehouse.gold.batch_view_metadata")
    batch_max_ts = batch_metadata.select(F.max("processed_up_to_ts")).collect()[0][0]

    print(f"Batch Layer покрывает данные до: {batch_max_ts}")
    print(f"Speed Layer покрывает данные с: {batch_max_ts} до сейчас")

    # 2. Читаем Batch Views: точные исторические данные
    batch_views = (
        spark.table("lakehouse.gold.batch_revenue_by_hour")
        .filter(F.col("hour_ts").between(start_date, end_date))
        # Берём ТОЛЬКО данные до границы Batch Layer
        .filter(F.col("hour_ts") < F.lit(batch_max_ts))
        .select(
            "hour_ts", "event_type",
            F.col("total_amount"),
            F.col("event_count"),
            F.lit("batch").alias("source_layer")
        )
    )

    # 3. Читаем Realtime Views: свежие данные
    realtime_views = (
        spark.table("lakehouse.gold.realtime_views")
        .filter(F.col("window.start").between(start_date, end_date))
        # Берём ТОЛЬКО данные ПОСЛЕ границы Batch Layer - избегаем дублирования
        .filter(F.col("window.start") >= F.lit(batch_max_ts))
        .select(
            F.col("window.start").alias("hour_ts"),
            "event_type",
            F.col("total_amount"),
            F.col("event_count"),
            F.lit("speed").alias("source_layer")
        )
    )

    # 4. Объединяем два источника
    # UNION ALL корректен: временные диапазоны не пересекаются
    unified = batch_views.unionByName(realtime_views, allowMissingColumns=True)

    # 5. Финальная агрегация по дням для дашборда
    result = (
        unified
        .groupBy(F.to_date("hour_ts").alias("date"), "event_type")
        .agg(
            F.sum("total_amount").alias("total_amount"),
            F.sum("event_count").alias("event_count"),
            F.countDistinct("source_layer").alias("layers_used")  # 1=только batch, 2=batch+speed
        )
        .orderBy("date", "event_type")
    )

    print("Результат unified query (Batch + Speed):")
    result.show(truncate=False)

def compute_revenue_metrics(df):
    """
    Эта функция работает ОДИНАКОВО для:
    1. Batch DataFrame (исторические данные)
    2. Streaming DataFrame (поток событий в реальном времени)
    Бизнес-логика написана один раз.
    """
    return (
        df
        # Фильтрация: только успешные транзакции с положительной суммой
        .filter(F.col("event_type") == "purchase")
        .filter(F.col("amount") > 0)
        # Обогащение: категоризация суммы
        .withColumn("amount_tier",
            F.when(F.col("amount") < 1000, "micro")
             .when(F.col("amount") < 10000, "mid")
             .otherwise("large")
        )
        # Агрегация по временному окну
        .groupBy(
            F.window("event_ts", "1 hour").alias("hour_window"),
            "amount_tier"
        )
        .agg(
            F.sum("amount").alias("revenue"),
            F.count("*").alias("transactions"),
            F.approx_count_distinct("user_id", rsd=0.01).alias("unique_buyers")
        )
    )

# --- BATCH режим (Batch Layer) ---
historical_df = (
    spark.table("lakehouse.bronze.events")
         .filter(F.col("event_date").between("2026-01-01", "2026-05-22"))
)
# Используем ту же функцию
batch_result = compute_revenue_metrics(historical_df)
batch_result.write.format("iceberg").mode("overwrite").saveAsTable("lakehouse.gold.batch_revenue")

# --- STREAMING режим (Speed Layer) ---
# Тот же вызов функции, но df - это streaming DataFrame
streaming_df = (
    spark.readStream
         .format("kafka")
         .option("kafka.bootstrap.servers", "kafka:9092")
         .option("subscribe", "ecommerce.events")
         .option("startingOffsets", "latest")
         .load()
         .selectExpr("CAST(value AS STRING) as json_payload")
         .select(F.from_json("json_payload", event_schema).alias("data"))
         .select("data.*")
)

# Используем ту же функцию - она работает и для streaming!
streaming_result = compute_revenue_metrics(streaming_df.withWatermark("event_ts", "10 minutes"))

streaming_query = (
    streaming_result
    .writeStream
    .trigger(processingTime="30 seconds")
    .outputMode("update")
    .format("iceberg")
    .option("checkpointLocation", "s3://my-bucket/checkpoints/speed-revenue")
    .toTable("lakehouse.gold.realtime_revenue")
)

def process_microbatch(batch_df, batch_id: int) -> None:
    """
    foreachBatch функция: вызывается для каждого микро-батча.
    batch_df: обычный (не streaming) DataFrame с данными текущего микро-батча
    batch_id: уникальный ID батча, монотонно возрастающий

    Идемпотентность: функция должна давать одинаковый результат при повторном вызове
    с теми же данными. Spark может вызвать её повторно при сбое.
    Идемпотентность гарантируется через дедупликацию по batch_id и event_id.
    """
    if batch_df.isEmpty():
        print(f"Батч {batch_id}: пустой, пропускаем")
        return

    # 1. Дедупликация: защита от повторного вызова (at-least-once semantics foreachBatch)
    deduped_df = batch_df.dropDuplicates(["event_id"])

    # 2. Кешируем DataFrame: он будет использован несколько раз
    # Без cache() Spark читал бы Kafka повторно для каждого .write
    deduped_df.cache()

    record_count = deduped_df.count()
    print(f"Батч {batch_id}: {record_count} уникальных событий после дедупликации")

    # 3. Запись в Bronze (append-only)
    # Все сырые события сохраняются в неизменяемый лог
    (
        deduped_df
        .withColumn("event_date", F.to_date("event_ts"))
        .writeTo("lakehouse.bronze.events")
        .append()
    )

    # 4. MERGE INTO Silver: обновление текущего состояния пользователей
    # Регистрируем временное представление для использования в SQL
    deduped_df.createOrReplaceTempView("incoming_events_view")

    spark.sql("""
        MERGE INTO lakehouse.silver.user_profiles AS target
        USING (
            SELECT
                user_id,
                SUM(amount) as session_revenue,
                COUNT(*) as session_events,
                MAX(event_ts) as last_seen_at
            FROM incoming_events_view
            WHERE event_type = 'purchase'
            GROUP BY user_id
        ) AS source
        ON target.user_id = source.user_id
        WHEN MATCHED THEN UPDATE SET
            target.total_revenue  = target.total_revenue + source.session_revenue,
            target.total_events   = target.total_events + source.session_events,
            target.last_seen_at   = source.last_seen_at,
            target.updated_at     = current_timestamp()
        WHEN NOT MATCHED THEN INSERT
            (user_id, total_revenue, total_events, last_seen_at, created_at, updated_at)
            VALUES (source.user_id, source.session_revenue, source.session_events,
                    source.last_seen_at, current_timestamp(), current_timestamp())
    """)

    # 5. Запись в Gold Realtime View (агрегаты для дашборда)
    (
        deduped_df
        .groupBy(
            F.window("event_ts", "1 minute").alias("time_window"),
            "event_type"
        )
        .agg(
            F.sum("amount").alias("total_amount"),
            F.count("*").alias("event_count")
        )
        .writeTo("lakehouse.gold.realtime_views")
        .append()
    )

    # 6. Фоновая компакция: каждые 60 батчей (~30 минут при 30-секундном триггере)
    if batch_id % 60 == 0:
        print(f"Батч {batch_id}: запускаем компакцию Bronze...")
        spark.sql("""
            CALL lakehouse.system.rewrite_data_files(
                table => 'lakehouse.bronze.events',
                strategy => 'sort',
                sort_order => 'event_date ASC, user_id ASC',
                options => map(
                    'target-file-size-bytes', '134217728',
                    'partial-progress.enabled', 'true'
                )
            )
        """)

    deduped_df.unpersist()


# Запускаем стриминговый пайплайн
streaming_query = (
    parsed_stream
    .writeStream
    .trigger(processingTime="30 seconds")
    .foreachBatch(process_microbatch)
    .option("checkpointLocation", "s3://my-bucket/checkpoints/lambda-speed-layer")
    .start()
)

streaming_query.awaitTermination()

# Критически важно: всегда устанавливайте watermark ПЕРЕД stateful операцией
# Watermark говорит Spark: "окна старше X минут от последнего события можно закрыть
# и выбросить из State Store"

revenue_by_window = (
    streaming_df
    # Шаг 1: watermark на колонке event_ts
    # "10 minutes" означает: опоздавшие события принимаем до 10 минут,
    # более старые игнорируем
    .withWatermark("event_ts", "10 minutes")
    # Шаг 2: stateful агрегация
    .groupBy(
        F.window("event_ts", "1 hour"),   # Часовые окна
        "event_type"
    )
    .agg(
        F.sum("amount").alias("revenue"),
        F.count("*").alias("tx_count")
    )
)
# Без watermark: State Store хранит ВСЕ окна с начала времён → OOM через N дней
# С watermark: Spark удаляет из State Store окна старше
#              (last_event_ts - 10 minutes - window_duration)

# Программный доступ к метрикам стриминга
query = streaming_df.writeStream \
    .trigger(processingTime="30 seconds") \
    .foreachBatch(process_microbatch) \
    .option("checkpointLocation", "s3://bucket/checkpoints/speed-layer") \
    .start()

# После запуска - получаем метрики текущего состояния
status = query.status
progress = query.lastProgress

print("=== Streaming Metrics ===")
print(f"Статус: {status['message']}")
print(f"Триггер активен: {status['isTriggerActive']}")

if progress:
    # Input Rate: сколько сообщений поступает в Kafka в секунду
    input_rate = progress.get("inputRowsPerSecond", 0)
    # Process Rate: сколько сообщений Spark обрабатывает в секунду
    process_rate = progress.get("processedRowsPerSecond", 0)

    print(f"Input Rate:   {input_rate:.1f} строк/сек (приходит в Kafka)")
    print(f"Process Rate: {process_rate:.1f} строк/сек (обрабатывает Spark)")

    # КЛЮЧЕВОЙ СИГНАЛ: если Input Rate > Process Rate - Consumer Lag растёт
    if input_rate > process_rate * 1.1:  # 10% допустимое отклонение
        print("ПРЕДУПРЕЖДЕНИЕ: Consumer Lag растёт!")
        print("Причины: недостаточно ресурсов исполнителей, долгая foreachBatch, I/O S3")

    # Дополнительные метрики
    batch_duration_ms = progress.get("batchDuration", 0)
    num_input_rows = progress.get("numInputRows", 0)
    print(f"Длительность батча: {batch_duration_ms}ms")
    print(f"Строк в последнем батче: {num_input_rows}")

# ПРОБЛЕМА: изменили схему входных данных, но checkpoint хранит старую схему
# Spark не может прочитать старый checkpoint с новой схемой → ошибка запуска

# СИМПТОМ: AnalysisException при старте стриминга после изменения схемы
# "Input schema changes are not supported in streaming. ..."

# АНТИПАТТЕРН: удалить checkpoint и начать заново
# Это приведёт к потере прогресса и к повторной обработке с нуля
# или к пропуску данных (если startingOffsets=latest)

# ПРАВИЛЬНОЕ РЕШЕНИЕ 1: Schema Evolution в Iceberg
# Добавляем колонку в таблицу ПРЕЖДЕ, чем менять стриминг
spark.sql("""
    ALTER TABLE lakehouse.bronze.events
    ADD COLUMN new_column STRING
    AFTER existing_column
""")
# Теперь стриминг может продолжить с тем же checkpoint

# ПРАВИЛЬНОЕ РЕШЕНИЕ 2: Миграция checkpoint
# Если схема изменилась несовместимо - создаём новый стриминговый джоб
# с новым checkpoint path и startingOffsets=earliest
# параллельно со старым, затем переключаем Serving Layer

# АНТИПАТТЕРН: очень длинное окно без агрессивного watermark
problematic_query = (
    streaming_df
    .withWatermark("event_ts", "7 days")    # Spark хранит state 7 дней!
    .groupBy(F.window("event_ts", "30 days")) # Окно 30 дней × миллионы ключей
    .agg(F.sum("amount"))
    # Проблема: State Store = (30 days / trigger interval) × keys_count × state_size
    # Через неделю работы: гигабайты RAM на каждый executor
)

# ПРАВИЛЬНЫЙ ПОДХОД: разбить на несколько шагов
# Шаг 1: короткие окна в стриминге (watermark 10 минут, окно 1 час)
hourly_stream = (
    streaming_df
    .withWatermark("event_ts", "10 minutes")
    .groupBy(F.window("event_ts", "1 hour"))
    .agg(F.sum("amount").alias("hourly_revenue"))
    .writeStream.toTable("lakehouse.silver.hourly_revenue")
)

# Шаг 2: длинные агрегации в Batch Layer (читает Silver, агрегирует по 30 дням)
# Нет ограничений на длину периода, нет State Store
monthly_batch = (
    spark.table("lakehouse.silver.hourly_revenue")
         .groupBy(F.window("window.start", "30 days"))
         .agg(F.sum("hourly_revenue").alias("monthly_revenue"))
)

# АНТИПАТТЕРН: checkpoint в /tmp или на локальном диске
bad_query = (
    streaming_df.writeStream
                .option("checkpointLocation", "/tmp/checkpoint")  # ОШИБКА!
                .start()
)
# При рестарте контейнера или перезапуске Spark на другом узле - checkpoint потерян
# Стриминг начнёт с нуля → потеря прогресса или дубликаты

# ПРАВИЛЬНО: checkpoint на S3 или HDFS
good_query = (
    streaming_df.writeStream
                .option("checkpointLocation", "s3://my-bucket/checkpoints/job-name-v1")
                .start()
)
# S3 доступен из любого узла кластера, переживает рестарты

# ВАЖНО: включайте версию в имя checkpoint
# При несовместимом изменении джоба - меняйте v1 на v2
# Это создаёт новый checkpoint (джоб стартует заново) без риска конфликта со старым

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

# === Инициализация Spark Session ===
spark = SparkSession.builder \
    .appName("ECommerce-Lambda-SpeedLayer") \
    .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://iceberg-catalog:8181") \
    .config("spark.sql.catalog.lakehouse.warehouse", "s3://my-bucket/warehouse") \
    # Оптимизации для Structured Streaming
    .config("spark.sql.streaming.stateStore.providerClass",
            "org.apache.spark.sql.execution.streaming.state.RocksDBStateStoreProvider") \
    .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),  # purchase, view, cart_add, cart_remove
    StructField("product_id",  StringType(),    True),
    StructField("amount",      DoubleType(),    True),
    StructField("discount_pct", DoubleType(),   True),   # процент скидки 0-100
    StructField("event_ts",    TimestampType(), False),
    StructField("ip_address",  StringType(),    True),
])

# === Читаем из Kafka ===
raw_stream = (
    spark.readStream
         .format("kafka")
         .option("kafka.bootstrap.servers", "kafka:9092")
         .option("subscribe", "ecommerce.purchases,ecommerce.views,ecommerce.carts")
         .option("startingOffsets", "latest")
         .option("maxOffsetsPerTrigger", "100000")  # защита от OOM при большом lag
         .option("kafka.group.id", "spark-lambda-speed-layer")
         .load()
         .select(
             F.col("topic"),
             F.col("partition"),
             F.col("offset"),
             F.col("timestamp").alias("kafka_ingestion_ts"),
             F.from_json(F.col("value").cast("string"), event_schema).alias("d")
         )
         .select("topic", "partition", "offset", "kafka_ingestion_ts", "d.*")
)


# === Общая функция обогащения (используется в Batch И в Streaming) ===
def enrich_events(df):
    """
    Обогащение событий: одинаковая логика для Batch и Speed Layer.
    Добавляет бизнес-правила: фильтрация невалидных записей,
    категоризация, расчёт net_amount.
    """
    return (
        df
        # Фильтрация: только валидные суммы для финансовых событий
        .filter(
            (F.col("event_type") != "purchase") | (F.col("amount") > 0)
        )
        # Бизнес-правило: скидки > 50% не учитываются в выручке
        # Это применяется ОДИНАКОВО и в Batch, и в Speed
        .withColumn(
            "net_amount",
            F.when(
                (F.col("event_type") == "purchase") & (F.col("discount_pct") <= 50),
                F.col("amount") * (1 - F.col("discount_pct") / 100)
            ).otherwise(F.lit(0.0))
        )
        # Категоризация суммы (для сегментации аналитики)
        .withColumn(
            "amount_tier",
            F.when(F.col("amount") < 500,   "micro")
             .when(F.col("amount") < 5000,  "mid")
             .when(F.col("amount") < 50000, "large")
             .otherwise("enterprise")
        )
    )


# === foreachBatch функция ===
def process_ecommerce_batch(batch_df, batch_id: int) -> None:
    """
    Центральная функция обработки каждого микро-батча Speed Layer.
    Выполняет запись в 3 слоя Lakehouse параллельно.
    """
    if batch_df.isEmpty():
        return

    # 1. Дедупликация по event_id (защита от at-least-once Kafka)
    deduped = batch_df.dropDuplicates(["event_id"]).cache()
    count = deduped.count()

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

    print(f"[Batch {batch_id}] Обрабатываем {count} уникальных событий")

    # 2. Обогащаем (та же функция, что и в Batch Layer)
    enriched = enrich_events(deduped)

    # 3. Bronze: пишем сырые события (append-only)
    # Сохраняем ВСЕ поля, включая topic, partition, offset для дебаггинга
    (
        enriched
        .withColumn("event_date", F.to_date("event_ts"))
        .writeTo("lakehouse.bronze.ecommerce_events")
        .append()
    )

    # 4. Silver: обновляем профили пользователей через MERGE INTO
    # Только для событий типа purchase
    enriched.filter(F.col("event_type") == "purchase") \
            .createOrReplaceTempView("speed_purchases")

    spark.sql("""
        MERGE INTO lakehouse.silver.user_purchase_profiles AS t
        USING (
            SELECT
                user_id,
                COUNT(*)         AS purchase_count,
                SUM(net_amount)  AS total_net_revenue,
                MAX(event_ts)    AS last_purchase_at,
                MAX(amount_tier) AS highest_tier
            FROM speed_purchases
            GROUP BY user_id
        ) AS s
        ON t.user_id = s.user_id
        WHEN MATCHED THEN UPDATE SET
            t.purchase_count     = t.purchase_count + s.purchase_count,
            t.total_net_revenue  = t.total_net_revenue + s.total_net_revenue,
            t.last_purchase_at   = s.last_purchase_at,
            t.updated_at         = current_timestamp()
        WHEN NOT MATCHED THEN INSERT *
    """)

    # 5. Gold Realtime View: агрегаты для live дашборда
    (
        enriched
        .groupBy(
            F.date_trunc("minute", F.col("event_ts")).alias("minute_ts"),
            "event_type",
            "amount_tier"
        )
        .agg(
            F.sum("net_amount").alias("net_revenue"),
            F.count("*").alias("event_count"),
            F.approx_count_distinct("user_id").alias("unique_users"),
            F.approx_count_distinct("session_id").alias("unique_sessions")
        )
        .writeTo("lakehouse.gold.realtime_revenue_by_minute")
        .append()
    )

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

    deduped.unpersist()


# === Fraud Detection: отдельный стриминговый джоб с коротким триггером ===
def detect_fraud_batch(batch_df, batch_id: int) -> None:
    """
    Обнаружение аномалий: пользователи с > 10 покупками за 2 минуты
    или с суммой > 100 000 руб за 5 минут.
    """
    if batch_df.isEmpty():
        return

    batch_df.createOrReplaceTempView("recent_purchases")

    # Детектируем аномальную активность
    alerts = spark.sql("""
        SELECT
            user_id,
            COUNT(*) as purchase_count_2min,
            SUM(amount) as total_amount_2min,
            CASE
                WHEN COUNT(*) > 10             THEN 'HIGH_FREQUENCY'
                WHEN SUM(amount) > 100000      THEN 'HIGH_AMOUNT'
                ELSE 'NORMAL'
            END as alert_type,
            current_timestamp() as detected_at
        FROM recent_purchases
        WHERE event_type = 'purchase'
          AND event_ts >= current_timestamp() - INTERVAL 2 MINUTES
        GROUP BY user_id
        HAVING COUNT(*) > 10 OR SUM(amount) > 100000
    """)

    alert_count = alerts.count()
    if alert_count > 0:
        print(f"[Fraud Batch {batch_id}] АЛЕРТ: {alert_count} подозрительных пользователей!")
        (
            alerts
            .writeTo("lakehouse.gold.fraud_alerts")
            .append()
        )


# === Запуск Speed Layer ===
main_query = (
    raw_stream
    .writeStream
    .trigger(processingTime="30 seconds")
    .foreachBatch(process_ecommerce_batch)
    .option("checkpointLocation", "s3://my-bucket/checkpoints/ecommerce-speed-v2")
    .start()
)

# Fraud detection - отдельный джоб с более частым триггером
fraud_query = (
    raw_stream
    .filter(F.col("event_type") == "purchase")
    .withWatermark("event_ts", "5 minutes")
    .writeStream
    .trigger(processingTime="5 seconds")
    .foreachBatch(detect_fraud_batch)
    .option("checkpointLocation", "s3://my-bucket/checkpoints/fraud-detection-v1")
    .start()
)

# Ждём завершения (в production работает постоянно)
spark.streams.awaitAnyTermination()

# === Batch Layer: ежечасный инкрементальный пересчёт ===
# Запускается по расписанию (Apache Airflow, Prefect, cron)

from datetime import datetime, timedelta

def run_hourly_batch_job(target_hour: datetime) -> None:
    """
    Инкрементальный Batch Job: обрабатывает данные за конкретный час.
    Используется для ежечасного обновления Batch Views.

    target_hour: datetime объект часа для обработки (напр., 2026-05-23 14:00)
    """
    hour_start = target_hour.replace(minute=0, second=0, microsecond=0)
    hour_end   = hour_start + timedelta(hours=1)

    print(f"Batch Job: обрабатываем {hour_start} - {hour_end}")

    # Читаем из Bronze только нужный час (Partition Pruning по event_date + фильтр)
    bronze_hour = (
        spark.table("lakehouse.bronze.ecommerce_events")
             .filter(F.col("event_date") == hour_start.date())       # Partition Pruning
             .filter(F.col("event_ts").between(hour_start, hour_end)) # Точный диапазон
    )

    # Применяем ту же функцию обогащения (единая бизнес-логика!)
    enriched_hour = enrich_events(bronze_hour)

    # Агрегация за час
    hourly_aggregates = (
        enriched_hour
        .groupBy(
            F.date_trunc("hour", F.col("event_ts")).alias("hour_ts"),
            "event_type",
            "amount_tier"
        )
        .agg(
            F.sum("net_amount").alias("net_revenue"),
            F.count("*").alias("event_count"),
            F.countDistinct("user_id").alias("unique_users"),
            F.countDistinct("session_id").alias("unique_sessions"),
            F.lit("batch").alias("source_layer")
        )
    )

    # Записываем Batch View: перезаписываем партицию нужного часа
    # Это идемпотентно: повторный запуск за тот же час даст тот же результат
    (
        hourly_aggregates
        .writeTo("lakehouse.gold.batch_revenue_by_hour")
        .overwritePartitions()
    )

    # Обновляем метаданные: до какого момента покрыт Batch Layer
    spark.sql(f"""
        MERGE INTO lakehouse.gold.batch_view_metadata AS t
        USING (SELECT '{hour_end}' as processed_up_to_ts) AS s
        ON 1 = 1
        WHEN MATCHED AND t.processed_up_to_ts < s.processed_up_to_ts
            THEN UPDATE SET t.processed_up_to_ts = s.processed_up_to_ts,
                            t.updated_at = current_timestamp()
        WHEN NOT MATCHED THEN INSERT *
    """)

    print(f"Batch Job завершён: {hour_start} - {hour_end}")
    print(f"Batch Views покрывают данные до: {hour_end}")


# === Полный исторический пересчёт (при исправлении бизнес-логики) ===
def run_full_historical_recompute(start_date: str, end_date: str) -> None:
    """
    Полный пересчёт Batch Views за исторический период.
    Запускается при изменении бизнес-логики (например, пересмотр правила скидок).

    Ключевой принцип Lambda: Bronze нетронут, пересчитываем только Gold Batch Views.
    """
    print(f"Исторический пересчёт: {start_date} - {end_date}")
    print("Bronze Layer не изменяется - только Gold Batch Views пересчитываются")

    # Читаем весь исторический Bronze за период
    bronze_historical = (
        spark.table("lakehouse.bronze.ecommerce_events")
             .filter(F.col("event_date").between(start_date, end_date))
    )

    # Применяем ТЕКУЩУЮ (исправленную) бизнес-логику
    enriched_historical = enrich_events(bronze_historical)  # новая версия функции

    # Агрегируем по дням для исторических отчётов
    daily_aggregates = (
        enriched_historical
        .groupBy(
            F.to_date("event_ts").alias("date"),
            "event_type"
        )
        .agg(
            F.sum("net_amount").alias("net_revenue"),
            F.count("*").alias("event_count"),
            F.countDistinct("user_id").alias("unique_users")
        )
    )

    # Перезаписываем все партиции за период
    (
        daily_aggregates
        .writeTo("lakehouse.gold.batch_revenue_by_day")
        .overwritePartitions()
    )

    print(f"Исторический пересчёт завершён: {daily_aggregates.count()} строк записано")

# === Serving Layer: объединение Batch и Speed ===
# Этот запрос используется Live Dashboard (обновляется каждые 30 секунд)

def get_dashboard_data(period_start: str, period_end: str):
    """
    Возвращает данные для Live Dashboard за период.
    Автоматически объединяет точные Batch данные и свежие Speed данные.
    Нет двойного счёта.
    """
    # Определяем границу Batch Layer
    batch_meta = spark.table("lakehouse.gold.batch_view_metadata")
    batch_max_ts_row = batch_meta.select(F.max("processed_up_to_ts")).first()
    batch_max_ts = batch_max_ts_row[0] if batch_max_ts_row[0] else "2000-01-01"

    print(f"Dashboard: Batch покрывает до {batch_max_ts}, Speed - остальное")

    # Batch данные: точные исторические агрегаты (до batch_max_ts)
    batch_data = (
        spark.table("lakehouse.gold.batch_revenue_by_hour")
             .filter(F.col("hour_ts").between(period_start, period_end))
             .filter(F.col("hour_ts") < F.lit(batch_max_ts))
             .select(
                 "hour_ts", "event_type", "amount_tier",
                 "net_revenue", "event_count", "unique_users",
                 F.lit("batch").alias("data_source")
             )
    )

    # Speed данные: свежие агрегаты (начиная с batch_max_ts)
    speed_data = (
        spark.table("lakehouse.gold.realtime_revenue_by_minute")
             .filter(F.col("minute_ts").between(period_start, period_end))
             .filter(F.col("minute_ts") >= F.lit(batch_max_ts))
             .withColumn("hour_ts", F.date_trunc("hour", F.col("minute_ts")))
             .groupBy("hour_ts", "event_type", "amount_tier")
             .agg(
                 F.sum("net_revenue").alias("net_revenue"),
                 F.sum("event_count").alias("event_count"),
                 F.sum("unique_users").alias("unique_users")  # приблизительно
             )
             .withColumn("data_source", F.lit("speed"))
    )

    # Объединяем (непересекающиеся периоды, нет двойного счёта)
    unified = (
        batch_data.unionByName(speed_data, allowMissingColumns=True)
                  .orderBy("hour_ts", "event_type")
    )

    return unified


# Пример вызова для дашборда
dashboard_df = get_dashboard_data(
    period_start="2026-05-23 00:00:00",
    period_end="2026-05-23 23:59:59"
)

dashboard_df.show(truncate=False)
# hour_ts             | event_type | net_revenue | data_source
# 2026-05-23 00:00:00 | purchase   | 45 000.00   | batch
# 2026-05-23 01:00:00 | purchase   | 38 500.00   | batch
# ...
# 2026-05-23 14:00:00 | purchase   | 72 100.00   | batch
# 2026-05-23 15:00:00 | purchase   | 12 300.00   | speed  <- свежие данные
# 2026-05-23 15:30:00 | purchase   | 5 800.00    | speed  <- за последние минуты