Medallion Architecture: bronze, silver, gold - принципы и границы слоёв

from pyspark.sql import SparkSession
from pyspark.sql import functions as F

spark = SparkSession.builder \
    .appName("Bronze-Ingestion") \
    .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", "/tmp/warehouse") \
    .getOrCreate()

# Создаём Bronze-таблицу с Iceberg
# PARTITIONED BY (ingestion_date) - партиционирование по дате загрузки
# для быстрого чтения инкрементальных обновлений
spark.sql("""
    CREATE TABLE IF NOT EXISTS lakehouse.bronze.orders (
        order_id        STRING,
        customer_id     STRING,
        status          STRING,
        amount          STRING,    -- намеренно STRING: Bronze хранит оригинал
        created_at      STRING,    -- тоже STRING: не делаем cast на Bronze
        raw_payload     STRING,    -- полный оригинальный JSON (для отладки)
        _ingested_at    TIMESTAMP  COMMENT 'Время загрузки в Bronze',
        _source_file    STRING     COMMENT 'Путь к исходному файлу',
        _pipeline_run_id STRING    COMMENT 'ID запуска пайплайна',
        _source_system  STRING     COMMENT 'Имя системы-источника',
        ingestion_date  DATE       COMMENT 'Партиция по дате загрузки'
    )
    USING iceberg
    PARTITIONED BY (ingestion_date)
""")

# Имитируем «грязные» сырые данные из источника
# Обратите внимание: amount - строка, дата в нестандартном формате,
# есть дубликат (order_id=1001 пришёл дважды), одна запись явно битая
raw_orders_data = [
    ("1001", "cust_A", "confirmed", "1500.00", "2026-05-23 10:00:00", None),
    ("1001", "cust_A", "confirmed", "1500.00", "2026-05-23 10:00:00", None),  # дубликат
    ("1002", "cust_B", "pending",   "750.50",  "23-05-2026 11:30:00", None),  # другой формат даты
    ("1003", None,     "shipped",   "N/A",     "2026-05-23 12:00:00", None),  # битая запись
    ("1004", "cust_C", "cancelled", "-200.00", "2026-05-23 13:00:00", None),
]

raw_df = spark.createDataFrame(
    raw_orders_data,
    ["order_id", "customer_id", "status", "amount", "created_at", "raw_payload"]
)

# Добавляем технические метаданные - ЕДИНСТВЕННАЯ трансформация на Bronze
bronze_df = raw_df \
    .withColumn("raw_payload",
                F.to_json(F.struct([F.col(c) for c in raw_df.columns]))) \
    .withColumn("_ingested_at",    F.current_timestamp()) \
    .withColumn("_source_file",    F.lit("s3://raw/orders/2026-05-23.json")) \
    .withColumn("_pipeline_run_id", F.lit("run-20260523-001")) \
    .withColumn("_source_system",  F.lit("orders_db")) \
    .withColumn("ingestion_date",  F.current_date())

# APPEND - единственный режим записи на Bronze
# Дубликаты, битые строки - всё пишем как есть
bronze_df.writeTo("lakehouse.bronze.orders").append()

print(f"Записано в Bronze: {bronze_df.count()} строк (включая дубликаты и битые)")
spark.table("lakehouse.bronze.orders").show(truncate=False)

# Пример: источник добавил новое поле discount_code
# Bronze просто принимает расширенную схему
new_data_with_extra_field = spark.createDataFrame([
    ("1005", "cust_D", "confirmed", "2000.00", "2026-05-24 09:00:00", "SALE20"),
], ["order_id", "customer_id", "status", "amount", "created_at", "discount_code"])

# Iceberg автоматически добавит колонку discount_code в таблицу
new_bronze_df = new_data_with_extra_field \
    .withColumn("raw_payload",    F.to_json(F.struct([F.col(c)
                                  for c in new_data_with_extra_field.columns]))) \
    .withColumn("_ingested_at",   F.current_timestamp()) \
    .withColumn("_source_file",   F.lit("s3://raw/orders/2026-05-24.json")) \
    .withColumn("_pipeline_run_id", F.lit("run-20260524-001")) \
    .withColumn("_source_system", F.lit("orders_db")) \
    .withColumn("ingestion_date", F.current_date())

# mergeSchema=True позволяет добавлять новые колонки автоматически
new_bronze_df.writeTo("lakehouse.bronze.orders") \
    .option("mergeSchema", "true") \
    .append()

from pyspark.sql.window import Window

# Читаем из Bronze - берём только новые данные за нужную дату
bronze_df = spark.table("lakehouse.bronze.orders") \
    .filter(F.col("ingestion_date") == F.current_date())

# Подход 1: dropDuplicates - простой, но не всегда точный
# Хорошо работает, когда дубликат - это точная копия строки
dedup_simple = bronze_df.dropDuplicates(["order_id"])

# Подход 2: ROW_NUMBER по бизнес-ключу с сортировкой по времени
# Лучше: выбираем «самую позднюю» версию по _ingested_at
window_spec = Window.partitionBy("order_id").orderBy(F.col("_ingested_at").desc())

dedup_advanced = (
    bronze_df
    .withColumn("rn", F.row_number().over(window_spec))
    .filter(F.col("rn") == 1)
    .drop("rn")
)

print("Строк в Bronze (с дубликатами):", bronze_df.count())
print("Строк после дедупликации:", dedup_advanced.count())

# Разделяем записи на валидные и невалидные ПЕРЕД трансформацией
# Это паттерн "fail soft" - не падаем на ошибке, а помещаем в карантин

# Условия валидности бизнес-правил Silver-слоя
valid_condition = (
    F.col("order_id").isNotNull() &
    F.col("customer_id").isNotNull() &
    F.col("amount").rlike(r"^-?\d+(\.\d+)?$")  # Число (можно отрицательное)
)

valid_records    = dedup_advanced.filter(valid_condition)
invalid_records  = dedup_advanced.filter(~valid_condition)

# Карантинная зона - сохраняем для расследования, не удаляем
invalid_records \
    .withColumn("rejection_reason",
                F.when(F.col("order_id").isNull(),     F.lit("null_order_id"))
                 .when(F.col("customer_id").isNull(),  F.lit("null_customer_id"))
                 .when(~F.col("amount").rlike(r"^-?\d+(\.\d+)?$"),
                                                        F.lit("invalid_amount"))
                 .otherwise(F.lit("unknown"))) \
    .withColumn("quarantined_at", F.current_timestamp()) \
    .writeTo("lakehouse.silver.orders_quarantine") \
    .append()

# Типизация и стандартизация валидных записей
silver_df = (
    valid_records
    .withColumn("order_id",     F.col("order_id").cast("integer"))
    .withColumn("amount",       F.col("amount").cast("decimal(10,2)"))
    # Парсим оба формата даты: "2026-05-23 10:00:00" и "23-05-2026 11:30:00"
    .withColumn("created_at",
        F.coalesce(
            F.to_timestamp("created_at", "yyyy-MM-dd HH:mm:ss"),
            F.to_timestamp("created_at", "dd-MM-yyyy HH:mm:ss"),
        )
    )
    # Стандартизация статусов: приводим к нижнему регистру
    .withColumn("status",       F.lower(F.trim(F.col("status"))))
    # Отрицательный amount для cancelled-заказов - бизнес-аномалия, ставим 0
    .withColumn("amount",
        F.when(
            (F.col("status") == "cancelled") & (F.col("amount") < 0),
            F.lit(0).cast("decimal(10,2)")
        ).otherwise(F.col("amount"))
    )
    # Добавляем метаданные обработки Silver
    .withColumn("_silver_processed_at", F.current_timestamp())
    .select(
        "order_id", "customer_id", "status", "amount", "created_at",
        "_silver_processed_at"
    )
)

print(f"Валидных строк: {silver_df.count()}, в карантине: {invalid_records.count()}")

# Регистрируем Silver-данные как временное представление для Spark SQL
silver_df.createOrReplaceTempView("incoming_silver_orders")

# Создаём Silver-таблицу если не существует
spark.sql("""
    CREATE TABLE IF NOT EXISTS lakehouse.silver.orders (
        order_id              INTEGER,
        customer_id           STRING,
        status                STRING,
        amount                DECIMAL(10,2),
        created_at            TIMESTAMP,
        _silver_processed_at  TIMESTAMP
    )
    USING iceberg
    PARTITIONED BY (month(created_at))
""")

# MERGE INTO: идемпотентный UPSERT
# При повторном запуске order_id=1001 обновится (не продублируется)
spark.sql("""
    MERGE INTO lakehouse.silver.orders AS target
    USING incoming_silver_orders AS source
    ON target.order_id = source.order_id
    WHEN MATCHED AND (
        target.status  != source.status  OR
        target.amount  != source.amount
    ) THEN UPDATE SET *
    WHEN NOT MATCHED THEN INSERT *
""")

print("Silver обновлён. Финальное состояние:")
spark.table("lakehouse.silver.orders").show()

# Silver-таблица клиентов (предположим, что уже обработана)
customers_df = spark.table("lakehouse.silver.customers") \
    .select("customer_id", "customer_name", "city", "segment")

# Enrichment через LEFT JOIN
# LEFT JOIN: заказы без совпадающего клиента всё равно остаются (не теряются)
enriched_orders = (
    spark.table("lakehouse.silver.orders")
    .join(
        F.broadcast(customers_df),   # BROADCAST hint: customers_df маленький
        on="customer_id",
        how="left"
    )
)
# Результат: каждый заказ теперь содержит customer_name, city, segment

# Gold-витрина: ежедневная выручка по сегменту клиентов
# Потребитель: BI-дашборд для коммерческого директора

daily_revenue_mart = (
    enriched_orders
    .filter(F.col("status") == "confirmed")
    .withColumn("order_date", F.to_date("created_at"))
    .groupBy("order_date", "segment")
    .agg(
        F.sum("amount").alias("total_revenue"),
        F.count("order_id").alias("orders_count"),
        F.avg("amount").alias("avg_order_value"),
        F.countDistinct("customer_id").alias("unique_customers"),
    )
    .withColumn("revenue_per_customer",
                F.col("total_revenue") / F.col("unique_customers"))
    .orderBy("order_date", "segment")
)

# Создаём Gold-таблицу
spark.sql("""
    CREATE TABLE IF NOT EXISTS lakehouse.gold.daily_revenue_by_segment (
        order_date          DATE,
        segment             STRING,
        total_revenue       DECIMAL(15,2),
        orders_count        BIGINT,
        avg_order_value     DECIMAL(10,2),
        unique_customers    BIGINT,
        revenue_per_customer DECIMAL(10,2),
        _updated_at         TIMESTAMP
    )
    USING iceberg
    PARTITIONED BY (order_date)
""")

# Overwrite Partitions: для Gold часто используем перезапись партиций
# (вместо MERGE INTO) - проще и быстрее для агрегатов
(
    daily_revenue_mart
    .withColumn("_updated_at", F.current_timestamp())
    .writeTo("lakehouse.gold.daily_revenue_by_segment")
    .overwritePartitions()
)

print("Gold витрина обновлена:")
spark.table("lakehouse.gold.daily_revenue_by_segment").show()

# ROLLUP: иерархическая агрегация (order_date → segment → итого)
rollup_df = (
    enriched_orders
    .filter(F.col("status") == "confirmed")
    .withColumn("order_date", F.to_date("created_at"))
    .rollup("order_date", "segment")
    .agg(
        F.sum("amount").alias("total_revenue"),
        F.count("order_id").alias("orders_count")
    )
    # grouping() == 1 означает: это итоговая строка для данного уровня
    .withColumn("level",
        F.when(F.grouping("order_date") == 1, F.lit("GRAND_TOTAL"))
         .when(F.grouping("segment")    == 1, F.lit("DATE_TOTAL"))
         .otherwise(F.lit("DETAIL"))
    )
)

rollup_df.show(20)
# Результат содержит:
# DETAIL:     конкретная дата + сегмент → их выручка
# DATE_TOTAL: конкретная дата + null    → итог за день
# GRAND_TOTAL: null + null             → итог за всё время

# Паттерн "Overwrite Partitions" для Gold
# При повторном запуске партиция за дату просто перезаписывается
# Нет дублей, нет MERGE overhead

spark.conf.set("spark.sql.sources.partitionOverwriteMode", "dynamic")

(
    daily_revenue_mart
    .withColumn("_updated_at", F.current_timestamp())
    .write
    .format("iceberg")
    .mode("overwrite")  # Перезаписываем только тронутые партиции
    .option("partitionOverwriteMode", "dynamic")
    .saveAsTable("lakehouse.gold.daily_revenue_by_segment")
)

# Компакция мелких файлов в Silver-таблице
# Рекомендуемый целевой размер: 128-512 МБ на файл
spark.sql("""
    CALL lakehouse.system.rewrite_data_files(
        table => 'silver.orders',
        strategy => 'binpack',
        options => map('target-file-size-bytes', '134217728')
    )
""")

# Удаление устаревших снапшотов Iceberg
# Оставляем последние 7 дней для Time Travel
spark.sql("""
    CALL lakehouse.system.expire_snapshots(
        table => 'silver.orders',
        older_than => TIMESTAMP '2026-05-17 00:00:00',
        retain_last => 10
    )
""")

# Обновление статистики для Cost-Based Optimizer
spark.sql("ANALYZE TABLE lakehouse.silver.orders COMPUTE STATISTICS FOR ALL COLUMNS")

# Смотрим на Silver-таблицу, какой она была вчера
spark.sql("""
    SELECT * FROM lakehouse.silver.orders
    TIMESTAMP AS OF '2026-05-22 23:59:59'
""")

# Сравниваем текущее состояние с прошлым
spark.sql("""
    SELECT
        current.order_id,
        current.status  AS status_now,
        past.status     AS status_yesterday,
        current.amount  AS amount_now,
        past.amount     AS amount_yesterday
    FROM lakehouse.silver.orders current
    LEFT JOIN (
        SELECT * FROM lakehouse.silver.orders
        TIMESTAMP AS OF '2026-05-22 23:59:59'
    ) past USING (order_id)
    WHERE current.status != past.status
""")

import datetime
from pyspark.sql import SparkSession, functions as F
from pyspark.sql.window import Window

spark = SparkSession.builder \
    .appName("Medallion-E2E") \
    .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", "/tmp/warehouse") \
    .getOrCreate()

TODAY = datetime.date.today().isoformat()
PIPELINE_RUN_ID = f"run-{TODAY}-001"

# ─── ШАГ 1: BRONZE INGESTION ─────────────────────────────────────────────────
print("=== ШАГ 1: BRONZE ===")

raw_data = [
    ("1001","cust_A","confirmed","1500.00","2026-05-23 10:00:00"),
    ("1001","cust_A","confirmed","1500.00","2026-05-23 10:00:00"),  # дубликат
    ("1002","cust_B","pending",  "750.50", "2026-05-23 11:30:00"),
    ("1003",None,    "shipped",  "N/A",    "2026-05-23 12:00:00"),  # битая
    ("1004","cust_C","confirmed","200.00", "2026-05-23 13:00:00"),
]

raw_df = spark.createDataFrame(
    raw_data, ["order_id","customer_id","status","amount","created_at"]
)

bronze_df = (
    raw_df
    .withColumn("_ingested_at",     F.current_timestamp())
    .withColumn("_pipeline_run_id", F.lit(PIPELINE_RUN_ID))
    .withColumn("_source_system",   F.lit("orders_db"))
    .withColumn("ingestion_date",   F.current_date())
)

spark.sql("""CREATE TABLE IF NOT EXISTS lakehouse.bronze.orders (
    order_id STRING, customer_id STRING, status STRING,
    amount STRING, created_at STRING,
    _ingested_at TIMESTAMP, _pipeline_run_id STRING,
    _source_system STRING, ingestion_date DATE
) USING iceberg PARTITIONED BY (ingestion_date)""")

bronze_df.writeTo("lakehouse.bronze.orders").append()
print(f"Bronze: {bronze_df.count()} строк записано (включая дубли и битые)")

# ─── ШАГ 2: SILVER TRANSFORMATION ────────────────────────────────────────────
print("\n=== ШАГ 2: SILVER ===")

bronze_today = spark.table("lakehouse.bronze.orders") \
    .filter(F.col("ingestion_date") == F.current_date())

# Дедупликация: берём последнюю версию по _ingested_at
w = Window.partitionBy("order_id").orderBy(F.col("_ingested_at").desc())
deduped = bronze_today \
    .withColumn("rn", F.row_number().over(w)) \
    .filter(F.col("rn") == 1).drop("rn")

# Валидация и карантин
valid_cond = (
    F.col("order_id").isNotNull() &
    F.col("customer_id").isNotNull() &
    F.col("amount").rlike(r"^\d+(\.\d+)?$")
)
valid    = deduped.filter(valid_cond)
invalid  = deduped.filter(~valid_cond)

if invalid.count() > 0:
    spark.sql("""CREATE TABLE IF NOT EXISTS lakehouse.silver.orders_quarantine (
        order_id STRING, customer_id STRING, status STRING,
        amount STRING, rejection_reason STRING, quarantined_at TIMESTAMP
    ) USING iceberg""")
    invalid.withColumn("rejection_reason", F.lit("validation_failed")) \
           .withColumn("quarantined_at", F.current_timestamp()) \
           .select("order_id","customer_id","status","amount",
                   "rejection_reason","quarantined_at") \
           .writeTo("lakehouse.silver.orders_quarantine").append()
    print(f"В карантин отправлено: {invalid.count()} строк")

# Типизация
silver_df = (
    valid
    .withColumn("order_id", F.col("order_id").cast("integer"))
    .withColumn("amount", F.col("amount").cast("decimal(10,2)"))
    .withColumn("created_at",
        F.coalesce(
            F.to_timestamp("created_at","yyyy-MM-dd HH:mm:ss"),
            F.to_timestamp("created_at","dd-MM-yyyy HH:mm:ss"),
        ))
    .withColumn("status", F.lower(F.trim("status")))
    .withColumn("_silver_processed_at", F.current_timestamp())
    .select("order_id","customer_id","status","amount",
            "created_at","_silver_processed_at")
)

spark.sql("""CREATE TABLE IF NOT EXISTS lakehouse.silver.orders (
    order_id INTEGER, customer_id STRING, status STRING,
    amount DECIMAL(10,2), created_at TIMESTAMP, _silver_processed_at TIMESTAMP
) USING iceberg PARTITIONED BY (month(created_at))""")

silver_df.createOrReplaceTempView("incoming_silver")
spark.sql("""
    MERGE INTO lakehouse.silver.orders t
    USING incoming_silver s ON t.order_id = s.order_id
    WHEN MATCHED AND (t.status != s.status OR t.amount != s.amount)
    THEN UPDATE SET *
    WHEN NOT MATCHED THEN INSERT *
""")
print(f"Silver обновлён: {silver_df.count()} строк")

# ─── ШАГ 3: GOLD AGGREGATION ──────────────────────────────────────────────────
print("\n=== ШАГ 3: GOLD ===")

gold_df = (
    spark.table("lakehouse.silver.orders")
    .filter(F.col("status") == "confirmed")
    .withColumn("order_date", F.to_date("created_at"))
    .groupBy("order_date")
    .agg(
        F.sum("amount").alias("total_revenue"),
        F.count("order_id").alias("orders_count"),
        F.avg("amount").alias("avg_order_value"),
        F.countDistinct("customer_id").alias("unique_customers"),
    )
    .withColumn("_updated_at", F.current_timestamp())
)

spark.sql("""CREATE TABLE IF NOT EXISTS lakehouse.gold.daily_revenue (
    order_date DATE, total_revenue DECIMAL(15,2),
    orders_count BIGINT, avg_order_value DECIMAL(10,2),
    unique_customers BIGINT, _updated_at TIMESTAMP
) USING iceberg PARTITIONED BY (order_date)""")

gold_df.writeTo("lakehouse.gold.daily_revenue").overwritePartitions()

print("=== ИТОГ ПАЙПЛАЙНА ===")
print("Gold витрина:")
spark.table("lakehouse.gold.daily_revenue").show()

=== ШАГ 1: BRONZE ===
Bronze: 5 строк записано (включая дубли и битые)

=== ШАГ 2: SILVER ===
В карантин отправлено: 1 строк   (order_id=1003, customer_id=null, amount="N/A")
Silver обновлён: 3 строк         (1001 deduped, 1002, 1004)

=== ШАГ 3: GOLD ===
Gold витрина:
order_date  | total_revenue | orders_count | avg_order_value | unique_customers
2026-05-23  | 1700.00       | 2            | 850.00          | 2