Compaction: стратегии укрупнения файлов через Spark и Iceberg

# ❌ Наивный compaction: БЛОКИРУЕТ таблицу для читателей
df = spark.read.parquet("s3a://data/events/event_date=2026-01-15/")

# Во время этой операции:
# - Читатели получают либо старые данные (до overwrite) либо новые (после)
# - Если упадёт на середине - таблица в несогласованном состоянии
# - Нет атомарности: часть файлов удалена, часть ещё не записана
df.coalesce(4) \
  .write \
  .mode("overwrite") \
  .parquet("s3a://data/events/event_date=2026-01-15/")

# Что происходит в S3 при mode("overwrite"):
# 1. Spark удаляет все существующие файлы в папке
# 2. Spark пишет новые файлы
# Если шаг 2 упадёт на 50% - у вас пустая папка!
# Нет атомарности, нет изоляции, нет time-travel.

Классический Hive:
  S3 LIST s3://data/events/ → 1000 объектов
  S3 LIST s3://data/events/dt=2026-01-01/ → 1000 объектов
  ... 365 запросов × 1000 объектов = 365 000 файлов → медленно

Iceberg:
  Читаем metadata/current → snapshot-id
  Читаем manifest-list (один Avro-файл) → список Manifest
  Читаем нужные Manifests (Avro-файлы) → список data files с статистикой
  Применяем Partition Pruning и Column Statistics → читаем только нужные файлы
  Всё это - несколько GET запросов вместо 365+ LIST запросов

# Bin-Pack: просто объединить мелкие файлы в крупные
result = spark.sql("""
    CALL lakehouse.system.rewrite_data_files(
        table => 'db.events',
        strategy => 'binpack',
        options => map(
            'target-file-size-bytes', '536870912',
            -- ↑ целевой размер файла: 512 МБ
            -- Iceberg будет пытаться создать файлы близкие к этому размеру

            'min-input-files', '5',
            -- ↑ минимальное число файлов в группе для объединения
            -- Если в партиции только 3 файла - не трогать (не стоит overhead)

            'min-file-size-bytes', '134217728',
            -- ↑ файлы меньше 128 МБ считаются "мелкими" и подлежат объединению
            -- Файлы больше этого порога пропускаются (уже оптимальные)

            'max-file-size-bytes', '644245094'
            -- ↑ максимальный размер создаваемого файла: 614 МБ
            -- Файлы больше этого порога тоже подлежат расщеплению (слишком большие)
        )
    )
""")
result.show()
# +---------------------------+---------------------------+-----------------------+
# |rewritten_data_files_count |added_data_files_count     |rewritten_bytes_count  |
# +---------------------------+---------------------------+-----------------------+
# |1500                       |3                          |5368709120             |
# +---------------------------+---------------------------+-----------------------+
# Интерпретация: 1500 мелких файлов объединены в 3 крупных файла по ~1.8 ГБ каждый.
# Общий объём данных не изменился: 5368709120 байт ≈ 5 ГБ.

# Sort по одной колонке
result_sort = spark.sql("""
    CALL lakehouse.system.rewrite_data_files(
        table => 'db.events',
        strategy => 'sort',
        sort_order => 'user_id ASC NULLS LAST',
        -- ↑ сортировать строки внутри файлов по user_id по возрастанию
        -- Null-значения в конце (NULLS LAST - удобнее для аналитики)
        options => map(
            'target-file-size-bytes', '536870912',
            'min-input-files', '5'
        )
    )
""")
result_sort.show()

# Sort по нескольким колонкам (составной ключ)
result_sort_multi = spark.sql("""
    CALL lakehouse.system.rewrite_data_files(
        table => 'db.events',
        strategy => 'sort',
        sort_order => 'event_date ASC, user_id ASC NULLS LAST',
        -- ↑ сначала по дате, внутри даты - по user_id
        -- Идеально для запросов: WHERE event_date = '2026-01-15' AND user_id = 500
        options => map(
            'target-file-size-bytes', '536870912',
            'min-input-files', '3'
        )
    )
""")

# Z-Order по нескольким колонкам
result_zorder = spark.sql("""
    CALL lakehouse.system.rewrite_data_files(
        table => 'db.events',
        strategy => 'sort',
        sort_order => 'zorder(user_id, country, event_type)',
        -- ↑ Z-Order: близость одновременно по user_id, country И event_type
        -- Iceberg использует zorder() как специальный синтаксис в sort_order
        options => map(
            'target-file-size-bytes', '536870912',
            'min-input-files', '5',
            'rewrite-job-order', 'files-desc'
            -- ↑ начинать с самых больших файлов (приоритет на группы с наибольшим объёмом)
        )
    )
""")
result_zorder.show()

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

spark = SparkSession.builder \
    .appName("Iceberg-Compaction-Deep-Dive") \
    .config("spark.sql.extensions",
            "org.apache.iceberg.spark.extensions.IcebergSparkSessionExtensions") \
    # ↑ подключаем Iceberg SQL-расширения для CALL-процедур

    .config("spark.sql.catalog.lakehouse",
            "org.apache.iceberg.spark.SparkCatalog") \
    # ↑ регистрируем каталог с именем "lakehouse"

    .config("spark.sql.catalog.lakehouse.type", "hadoop") \
    # ↑ тип хранилища: hadoop-совместимая файловая система
    # В production: "hive" (Hive Metastore) или "rest" (Iceberg REST Catalog)

    .config("spark.sql.catalog.lakehouse.warehouse", "/tmp/iceberg-warehouse") \
    # ↑ путь к warehouse (корень хранилища)
    # В production: s3a://bucket/warehouse/

    .config("spark.executor.memory", "4g") \
    .config("spark.driver.memory", "2g") \
    .getOrCreate()

import datetime

# ─── Шаг 1: создать Iceberg таблицу ──────────────────────────────────────────
spark.sql("CREATE DATABASE IF NOT EXISTS lakehouse.db")

spark.sql("""
    CREATE TABLE IF NOT EXISTS lakehouse.db.events (
        event_id    BIGINT,
        user_id     BIGINT,
        event_type  STRING,
        country     STRING,
        amount      DOUBLE,
        event_ts    TIMESTAMP
    )
    USING iceberg
    PARTITIONED BY (days(event_ts))
    TBLPROPERTIES (
        'write.target-file-size-bytes' = '134217728',
        'write.distribution-mode' = 'hash'
    )
""")

# ─── Шаг 2: симулируем streaming запись (много мелких батчей) ─────────────────
# Записываем 100 «микробатчей» по 10 000 строк каждый
# Имитируем то, что Structured Streaming делал бы каждые 10 секунд

COUNTRIES = ["RU", "US", "DE", "CN", "BR", "IN", "GB", "FR"]
EVENT_TYPES = ["click", "purchase", "view", "add_to_cart", "checkout"]

print("Записываем 100 микробатчей...")
for batch_id in range(100):
    micro_batch = (
        spark.range(batch_id * 10_000, (batch_id + 1) * 10_000)
        .withColumn("user_id",    (F.col("id") % 100_000).cast("long"))
        .withColumn("event_type", F.element_at(
            F.array([F.lit(e) for e in EVENT_TYPES]),
            (F.col("id") % 5 + 1).cast("int")
        ))
        .withColumn("country", F.element_at(
            F.array([F.lit(c) for c in COUNTRIES]),
            (F.col("id") % 8 + 1).cast("int")
        ))
        .withColumn("amount",    (F.rand(seed=batch_id) * 5000 + 1).cast("double"))
        .withColumn("event_ts",  F.from_unixtime(
            1_735_689_600 + (F.col("id") % (7 * 24 * 3600)).cast("long")
        ).cast("timestamp"))
        .withColumnRenamed("id", "event_id")
    )

    micro_batch.writeTo("lakehouse.db.events").append()

print("Готово. Таблица создана с 100 микробатчами.")

# ─── Шаг 3: проверить состояние таблицы до compaction ────────────────────────
files_before = spark.sql("""
    SELECT
        count(*) AS file_count,
        sum(file_size_in_bytes) / (1024*1024) AS total_mb,
        avg(file_size_in_bytes) / (1024*1024) AS avg_file_mb,
        min(file_size_in_bytes) / (1024*1024) AS min_file_mb,
        max(file_size_in_bytes) / (1024*1024) AS max_file_mb
    FROM lakehouse.db.events.files
""")
files_before.show()
# +----------+---------+-------------+-------------+-------------+
# |file_count|total_mb |avg_file_mb  |min_file_mb  |max_file_mb  |
# +----------+---------+-------------+-------------+-------------+
# |~700-900  |~180-220 |~0.2-0.3     |~0.05        |~1.5         |
# Примерно 800 файлов по 200-300 КБ - типичная Small Files Disease

# ─── Bin-Pack: быстрое уплотнение без сортировки ─────────────────────────────
print("Запускаем Bin-Pack compaction...")
binpack_result = spark.sql("""
    CALL lakehouse.system.rewrite_data_files(
        table => 'lakehouse.db.events',
        strategy => 'binpack',
        options => map(
            'target-file-size-bytes', '268435456',
            'min-input-files', '5',
            'min-file-size-bytes', '134217728',
            'max-file-size-bytes', '536870912'
        )
    )
""")
binpack_result.show(truncate=False)
# +---------------------------+---------------------------+-----------------------+--------+
# |rewritten_data_files_count |added_data_files_count     |rewritten_bytes_count  |...     |
# +---------------------------+---------------------------+-----------------------+--------+
# |780                        |4                          |189054976              |...     |
# Интерпретация: 780 файлов → 4 файла, 180 МБ данных без изменений

# Проверить состояние ПОСЛЕ compaction
files_after = spark.sql("""
    SELECT count(*) AS file_count,
           avg(file_size_in_bytes) / (1024*1024) AS avg_file_mb
    FROM lakehouse.db.events.files
""")
files_after.show()
# +----------+------------+
# |file_count|avg_file_mb |
# +----------+------------+
# |4         |45.0        |  ← 4 файла вместо 780!

# ─── Sort по user_id: ускорение запросов WHERE user_id = ? ───────────────────
print("Запускаем Sort compaction по user_id...")

# Сначала делаем новую запись с мелкими файлами (как будто продолжился стриминг)
for batch_id in range(100, 150):
    micro_batch = spark.range(batch_id * 10_000, (batch_id + 1) * 10_000) \
        .withColumn("user_id",   (F.col("id") % 100_000).cast("long")) \
        .withColumn("event_type", F.lit("click")) \
        .withColumn("country",   F.lit("RU")) \
        .withColumn("amount",    (F.rand() * 1000).cast("double")) \
        .withColumn("event_ts",  F.from_unixtime(
            1_735_689_600 + (F.col("id") % (7 * 24 * 3600)).cast("long")
        ).cast("timestamp")) \
        .withColumnRenamed("id", "event_id")
    micro_batch.writeTo("lakehouse.db.events").append()

sort_result = spark.sql("""
    CALL lakehouse.system.rewrite_data_files(
        table => 'lakehouse.db.events',
        strategy => 'sort',
        sort_order => 'user_id ASC NULLS LAST, event_ts ASC',
        options => map(
            'target-file-size-bytes', '268435456',
            'min-input-files', '3',
            'rewrite-job-order', 'bytes-asc'
            -- ↑ начинать с самых маленьких файлов (укрупняем приоритетно мелкое)
        )
    )
""")
sort_result.show(truncate=False)

# Проверить эффект: запрос с фильтром по user_id
spark.sql("""
    SELECT user_id, sum(amount) AS total
    FROM lakehouse.db.events
    WHERE user_id = 42
    GROUP BY user_id
""").explain(mode="formatted")
# В плане теперь должны быть минимальные numFiles и хорошие PartitionFilters

# ─── Z-Order по (user_id, country, event_type) ───────────────────────────────
print("Запускаем Z-Order compaction...")
zorder_result = spark.sql("""
    CALL lakehouse.system.rewrite_data_files(
        table => 'lakehouse.db.events',
        strategy => 'sort',
        sort_order => 'zorder(user_id, country, event_type)',
        -- ↑ Z-Order по трём измерениям: запросы по любой из этих колонок ускорятся
        options => map(
            'target-file-size-bytes', '536870912',
            'min-input-files', '5',
            'max-concurrent-file-group-rewrites', '5'
            -- ↑ максимально 5 групп файлов перезаписываются параллельно
            -- Больше → быстрее compaction, но больше нагрузка на кластер
        )
    )
""")
zorder_result.show(truncate=False)

# ─── Шаг 1: expire_snapshots - пометить старые снапшоты как истёкшие ─────────
print("Очищаем старые снапшоты...")
expire_result = spark.sql("""
    CALL lakehouse.system.expire_snapshots(
        table => 'lakehouse.db.events',
        older_than => TIMESTAMP '2026-01-22 00:00:00',
        -- ↑ удалить снапшоты старше этой даты
        -- На практике: CURRENT_TIMESTAMP - interval 7 days (держать 7 дней Time Travel)
        retain_last => 10
        -- ↑ всегда оставлять минимум 10 последних снапшотов
        -- Даже если они старше older_than - это защита от случайного удаления всего
    )
""")
expire_result.show()
# +-------------------+----------------------+-----------------------+
# |deleted_data_files |deleted_manifest_files|deleted_manifest_lists |
# +-------------------+----------------------+-----------------------+
# |780                |15                    |8                      |
# Физически удалено 780 старых мелких файлов!

# ─── Шаг 2: remove_orphan_files - удалить «осиротевшие» файлы ────────────────
# Orphan files - файлы на диске без ссылок в Manifest (например, от прерванной записи)
orphan_result = spark.sql("""
    CALL lakehouse.system.remove_orphan_files(
        table => 'lakehouse.db.events',
        older_than => TIMESTAMP '2026-01-22 00:00:00'
        -- ↑ только файлы старше этого времени
        -- Не удалять очень свежие - вдруг запись ещё идёт
    )
""")
orphan_result.show()

# ─── Шаг 3: rewrite_manifests - уплотнение самих Manifest-файлов ──────────────
# После тысячи операций записи накопились тысячи мелких Manifest-файлов
# Это тоже замедляет планирование запросов
manifest_result = spark.sql("""
    CALL lakehouse.system.rewrite_manifests(
        table => 'lakehouse.db.events'
    )
""")
manifest_result.show()
# +-------------------------+-----------------------+
# |rewritten_manifests_count|added_manifests_count  |
# +-------------------------+-----------------------+
# |142                      |4                      |
# 142 мелких Manifest → 4 крупных = быстрее планирование запросов

# ─── Шаблон Compaction DAG для Apache Airflow ─────────────────────────────────
# (псевдокод - реальный Airflow DAG будет в курсе по оркестрации)

def run_compaction(table_name: str, partition_date: str, strategy: str = "binpack"):
    """
    Запустить compaction для конкретной партиции.
    Вызывается из Airflow с параметрами.
    """
    from pyspark.sql import SparkSession

    spark = SparkSession.builder \
        .appName(f"Compaction-{table_name}-{partition_date}") \
        .config("spark.sql.extensions",
                "org.apache.iceberg.spark.extensions.IcebergSparkSessionExtensions") \
        .getOrCreate()

    if strategy == "binpack":
        sort_clause = ""
        strategy_val = "binpack"
    elif strategy == "sort":
        sort_clause = ", sort_order => 'user_id ASC NULLS LAST'"
        strategy_val = "sort"
    else:
        sort_clause = ", sort_order => 'zorder(user_id, country)'"
        strategy_val = "sort"

    # Compaction только за конкретную дату (не трогаем всю историю!)
    result = spark.sql(f"""
        CALL lakehouse.system.rewrite_data_files(
            table => 'lakehouse.db.{table_name}',
            strategy => '{strategy_val}',
            where => 'event_ts >= TIMESTAMP ''{partition_date} 00:00:00''
                      AND event_ts < TIMESTAMP ''{partition_date} 23:59:59''',
            {sort_clause}
            options => map(
                'target-file-size-bytes', '268435456',
                'min-input-files', '5'
            )
        )
    """)

    metrics = result.first()
    print(f"Compaction {table_name}/{partition_date}: "
          f"{metrics['rewritten_data_files_count']} → {metrics['added_data_files_count']} файлов, "
          f"{metrics['rewritten_bytes_count'] / (1024**3):.2f} ГБ")

    # Expire старых снапшотов (7 дней retention)
    spark.sql(f"""
        CALL lakehouse.system.expire_snapshots(
            table => 'lakehouse.db.{table_name}',
            older_than => TIMESTAMP '{partition_date} 00:00:00',
            retain_last => 5
        )
    """)

    spark.stop()

# ─── Ежедневная стратегия: компактируем «вчера» ───────────────────────────────
from datetime import date, timedelta

def daily_compaction_job():
    """
    Ежедневный compaction:
    - Вчера: Bin-Pack (всегда, убираем мелкие файлы от вчерашнего стрима)
    - 7 дней назад: Sort (раз в неделю, оптимизируем для аналитики)
    - 30 дней назад: Z-Order (раз в месяц, максимальная оптимизация для BI)
    """
    today = date.today()

    # Bin-Pack вчерашней партиции
    yesterday = (today - timedelta(days=1)).isoformat()
    run_compaction("events", yesterday, strategy="binpack")

    # Sort неделю назад (по понедельникам)
    if today.weekday() == 0:  # понедельник
        week_ago = (today - timedelta(days=7)).isoformat()
        run_compaction("events", week_ago, strategy="sort")

    # Z-Order месяц назад (первого числа)
    if today.day == 1:
        month_ago = (today - timedelta(days=30)).isoformat()
        run_compaction("events", month_ago, strategy="zorder")

# ─── Мониторинг: сколько файлов и какого размера ─────────────────────────────
spark.sql("""
    SELECT
        partition.event_ts_day AS partition_date,
        count(*) AS file_count,
        sum(file_size_in_bytes) / (1024*1024) AS total_mb,
        avg(file_size_in_bytes) / (1024*1024) AS avg_file_mb,
        min(file_size_in_bytes) / (1024*1024) AS min_file_mb,
        max(file_size_in_bytes) / (1024*1024) AS max_file_mb
    FROM lakehouse.db.events.files
    GROUP BY 1
    HAVING count(*) > 10  -- показать только партиции с >10 файлами (кандидаты)
    ORDER BY file_count DESC
""").show(20, truncate=False)

# ─── Мониторинг: история снапшотов ──────────────────────────────────────────
spark.sql("""
    SELECT snapshot_id,
           committed_at,
           operation,
           summary['spark.app.id'] AS app_id,
           summary['added-data-files'] AS added_files,
           summary['deleted-data-files'] AS deleted_files
    FROM lakehouse.db.events.snapshots
    ORDER BY committed_at DESC
    LIMIT 20
""").show(truncate=False)

# ─── Time Travel: читать данные до compaction ─────────────────────────────────
# Можно убедиться, что данные не изменились
old_snapshot_id = 123456789  # ID снапшота до compaction (из таблицы snapshots)

before = spark.sql(f"""
    SELECT count(*), sum(amount)
    FROM lakehouse.db.events VERSION AS OF {old_snapshot_id}
""")

after = spark.sql("""
    SELECT count(*), sum(amount)
    FROM lakehouse.db.events
""")

before.show()  # должны совпасть с after
after.show()   # compaction не меняет данные, только физическую организацию

Стоит делать compaction ЕСЛИ:
  CPU(compaction) < N × (CPU(запрос без compaction) - CPU(запрос с compaction))

Где N - число запросов, которые будут сделаны до следующего compaction.

Пример:
  Bin-Pack compaction: 5 CPU-минут
  Скорость одного запроса: 10 мин → 2 мин после compaction (экономия 8 мин/запрос)
  Число запросов в день: 20
  Экономия в день: 20 × 8 = 160 CPU-минут
  Стоимость compaction: 5 CPU-минут
  ROI: 160/5 = 32x - однозначно стоит!

  Sort compaction: 60 CPU-минут
  Ускорение одного запроса: 0.5 мин (слабая аналитика, и так быстро)
  Число запросов в день: 5
  Экономия в день: 5 × 0.5 = 2.5 CPU-минуты
  Стоимость compaction: 60 CPU-минут
  ROI: 2.5/60 = 0.04x - НЕ стоит! Тратим больше чем экономим.