Compaction стратегии: Spark job, Delta OPTIMIZE, Iceberg rewrite_data_files

import math

def calculate_target_file_count(
    total_size_bytes: int,
    target_file_size_bytes: int = 256 * 1024 * 1024,  # 256 МБ
    min_files: int = 1
) -> int:
    """
    Рассчитывает оптимальное число файлов после компакции.

    total_size_bytes: суммарный размер всех файлов до компакции
    target_file_size_bytes: целевой размер каждого выходного файла
    min_files: минимальное число файлов (не схлопывать в 1)
    """
    if total_size_bytes <= 0:
        return min_files

    raw_count = math.ceil(total_size_bytes / target_file_size_bytes)
    return max(raw_count, min_files)


# Пример расчёта
total_gb = 45  # 45 ГБ данных в разделе
total_bytes = total_gb * 1024**3

target_files = calculate_target_file_count(
    total_size_bytes=total_bytes,
    target_file_size_bytes=256 * 1024 * 1024  # 256 МБ
)
print(f"45 ГБ данных → {target_files} файлов по 256 МБ")
# 45 ГБ данных → 180 файлов по 256 МБ

from pyspark.sql import SparkSession

spark = SparkSession.builder.getOrCreate()

# Простейшая компакция: читаем → переразбиваем → перезаписываем
(
    spark.read.parquet("s3a://bucket/data/events/")
    .repartition(180)          # 45 ГБ / 256 МБ = 180 файлов
    .write
    .mode("overwrite")
    .parquet("s3a://bucket/data/events/")
)

import math
from dataclasses import dataclass
from pyspark.sql import SparkSession, DataFrame
from typing import Optional


@dataclass
class CompactionConfig:
    """Конфигурация задачи компакции."""
    source_path: str
    output_path: str              # может совпадать с source_path при использовании tmp
    temp_path: str                # временный путь для записи
    target_file_size_mb: int = 256
    sort_columns: Optional[list] = None   # сортировать данные во время компакции?
    partition_columns: Optional[list] = None  # колонки партиционирования
    compression: str = "snappy"


def get_table_size_bytes(spark: SparkSession, path: str) -> int:
    """
    Получает суммарный размер всех файлов по пути.
    Использует Hadoop FileSystem API через Spark.
    """
    sc = spark.sparkContext
    hadoop_conf = sc._jvm.org.apache.hadoop.conf.Configuration()
    fs = sc._jvm.org.apache.hadoop.fs.FileSystem.get(
        sc._jvm.java.net.URI.create(path), hadoop_conf
    )
    content_summary = fs.getContentSummary(
        sc._jvm.org.apache.hadoop.fs.Path(path)
    )
    return content_summary.getLength()


def calculate_target_partitions(
    total_bytes: int,
    target_file_size_mb: int = 256,
    min_partitions: int = 1
) -> int:
    """Рассчитывает оптимальное число партиций Spark (= число файлов)."""
    target_bytes = target_file_size_mb * 1024 * 1024
    raw = math.ceil(total_bytes / target_bytes)
    return max(raw, min_partitions)


def compact_dataframe(
    df: DataFrame,
    config: CompactionConfig,
    total_bytes: int
) -> DataFrame:
    """
    Применяет компакцию к DataFrame:
    1. Рассчитывает оптимальное число партиций
    2. Опционально сортирует для кластеризации
    3. Возвращает готовый к записи DataFrame
    """
    target_partitions = calculate_target_partitions(
        total_bytes=total_bytes,
        target_file_size_mb=config.target_file_size_mb
    )
    print(f"Компакция: {total_bytes / 1024**3:.2f} ГБ → {target_partitions} файлов")

    if config.sort_columns:
        # sortWithinPartitions: сортировка внутри каждой партиции
        # НЕ то же самое, что orderBy (не вызывает глобальный shuffle)
        df = df.sortWithinPartitions(*config.sort_columns)

    return df.repartition(target_partitions)


def run_manual_compaction(spark: SparkSession, config: CompactionConfig) -> None:
    """
    Безопасная компакция через временный путь:
    1. Читаем источник
    2. Пишем во временный путь
    3. Атомарно переименовываем (или Hadoop rename на HDFS)
    """
    sc = spark.sparkContext
    hadoop_conf = sc._jvm.org.apache.hadoop.conf.Configuration()
    fs = sc._jvm.org.apache.hadoop.fs.FileSystem.get(
        sc._jvm.java.net.URI.create(config.source_path), hadoop_conf
    )

    # Шаг 1: измеряем объём данных
    total_bytes = get_table_size_bytes(spark, config.source_path)
    print(f"Источник: {config.source_path}")
    print(f"Объём: {total_bytes / 1024**3:.2f} ГБ")

    # Шаг 2: читаем данные
    df = spark.read.parquet(config.source_path)

    # Шаг 3: применяем компакцию
    compacted_df = compact_dataframe(df, config, total_bytes)

    # Шаг 4: пишем во временный путь
    writer = (
        compacted_df
        .write
        .mode("overwrite")
        .option("compression", config.compression)
    )

    if config.partition_columns:
        writer = writer.partitionBy(*config.partition_columns)

    writer.parquet(config.temp_path)
    print(f"Записано во временный путь: {config.temp_path}")

    # Шаг 5: удаляем старые данные и переименовываем
    # ВНИМАНИЕ: на S3 rename = copy + delete, это не атомарная операция!
    # Для production используйте Delta Lake или Iceberg
    src_path = sc._jvm.org.apache.hadoop.fs.Path(config.temp_path)
    dst_path = sc._jvm.org.apache.hadoop.fs.Path(config.output_path)

    if fs.exists(dst_path):
        fs.delete(dst_path, True)  # рекурсивное удаление

    fs.rename(src_path, dst_path)
    print(f"Компакция завершена: {config.output_path}")

from datetime import date, timedelta
from pyspark.sql import SparkSession


def compact_partitioned_table(
    spark: SparkSession,
    base_path: str,
    partition_column: str,
    partitions: list,
    target_file_size_mb: int = 256,
    sort_columns: Optional[list] = None
) -> None:
    """
    Компактирует партиционированную таблицу, обрабатывая каждый раздел отдельно.

    Пример вызова:
        compact_partitioned_table(
            spark=spark,
            base_path="s3a://bucket/events/",
            partition_column="event_date",
            partitions=["2024-01-01", "2024-01-02"],
            sort_columns=["user_id", "event_time"]
        )
    """
    for partition_value in partitions:
        partition_path = f"{base_path}/{partition_column}={partition_value}"
        temp_path = f"{base_path}/_compaction_tmp/{partition_column}={partition_value}"

        print(f"\nКомпактируем: {partition_column}={partition_value}")

        config = CompactionConfig(
            source_path=partition_path,
            output_path=partition_path,
            temp_path=temp_path,
            target_file_size_mb=target_file_size_mb,
            sort_columns=sort_columns
        )

        run_manual_compaction(spark, config)

    # Очищаем временную директорию
    sc = spark.sparkContext
    hadoop_conf = sc._jvm.org.apache.hadoop.conf.Configuration()
    fs = sc._jvm.org.apache.hadoop.fs.FileSystem.get(
        sc._jvm.java.net.URI.create(base_path), hadoop_conf
    )
    tmp_path = sc._jvm.org.apache.hadoop.fs.Path(f"{base_path}/_compaction_tmp")
    if fs.exists(tmp_path):
        fs.delete(tmp_path, True)
    print("\nКомпакция всех партиций завершена")


# Компактируем последние 7 дней
today = date.today()
recent_partitions = [
    str(today - timedelta(days=i))
    for i in range(7)
]

compact_partitioned_table(
    spark=spark,
    base_path="s3a://bucket/events/",
    partition_column="event_date",
    partitions=recent_partitions,
    target_file_size_mb=256,
    sort_columns=["user_id", "event_type"]
)

-- Простой OPTIMIZE всей таблицы
OPTIMIZE events;

-- OPTIMIZE с фильтром по партиции (рекомендуется для больших таблиц)
-- Compact только один раздел - быстрее и экономичнее
OPTIMIZE events
WHERE event_date = '2024-01-15';

-- OPTIMIZE с диапазоном дат
OPTIMIZE events
WHERE event_date BETWEEN '2024-01-01' AND '2024-01-31';

-- OPTIMIZE с Z-Order - компакция + кластеризация данных
-- Z-Order физически располагает похожие значения рядом в файле
-- Это позволяет Row Group Filtering пропускать больше данных
OPTIMIZE events
ZORDER BY (user_id, event_type);

-- OPTIMIZE конкретной партиции + Z-Order
OPTIMIZE events
WHERE event_date = '2024-01-15'
ZORDER BY (user_id, event_type);

from delta import DeltaTable
from pyspark.sql import SparkSession


def optimize_delta_table(
    spark: SparkSession,
    table_path: str,
    partition_filter: Optional[str] = None,
    zorder_columns: Optional[list] = None
) -> dict:
    """
    Выполняет OPTIMIZE для Delta Lake таблицы.

    Возвращает словарь с метриками:
    - numFilesAdded: число добавленных файлов (после компакции)
    - numFilesRemoved: число удалённых файлов (до компакции)
    - filesAdded.avg: средний размер новых файлов в байтах
    - filesRemoved.avg: средний размер старых файлов в байтах
    """
    dt = DeltaTable.forPath(spark, table_path)

    if partition_filter and zorder_columns:
        result = (
            dt.optimize()
            .where(partition_filter)
            .executeZOrderBy(*zorder_columns)
        )
    elif partition_filter:
        result = (
            dt.optimize()
            .where(partition_filter)
            .executeCompaction()
        )
    elif zorder_columns:
        result = (
            dt.optimize()
            .executeZOrderBy(*zorder_columns)
        )
    else:
        result = dt.optimize().executeCompaction()

    # result.show() - выводит метрики компакции
    metrics = result.first().asDict()
    return metrics


# Использование
metrics = optimize_delta_table(
    spark=spark,
    table_path="s3a://bucket/events/",
    partition_filter="event_date = '2024-01-15'",
    zorder_columns=["user_id", "event_type"]
)

print(f"Файлов до: {metrics['numFilesRemoved']}")
print(f"Файлов после: {metrics['numFilesAdded']}")
if metrics.get('filesRemoved') and metrics['filesRemoved'].get('avg'):
    print(f"Средний размер ДО: {metrics['filesRemoved']['avg'] / 1024**2:.1f} МБ")
if metrics.get('filesAdded') and metrics['filesAdded'].get('avg'):
    print(f"Средний размер ПОСЛЕ: {metrics['filesAdded']['avg'] / 1024**2:.1f} МБ")

from pyspark.sql import SparkSession

spark = SparkSession.builder \
    .config("spark.sql.extensions", "io.delta.sql.DeltaSparkSessionExtension") \
    .config("spark.sql.catalog.spark_catalog",
            "org.apache.spark.sql.delta.catalog.DeltaCatalog") \
    # Целевой размер файла после OPTIMIZE (default: 1 ГБ в Delta 2.x)
    .config("spark.databricks.delta.optimize.maxFileSize",
            str(256 * 1024 * 1024)) \
    # Минимальный размер файла для участия в компакции
    # Файлы крупнее этого порога пропускаются
    .config("spark.databricks.delta.optimize.minFileSize",
            str(32 * 1024 * 1024)) \
    .getOrCreate()

# Или через ALTER TABLE (per-table конфигурация)
spark.sql("""
    ALTER TABLE events
    SET TBLPROPERTIES (
        'delta.targetFileSize' = '268435456',     -- 256 МБ
        'delta.tuneFileSizesForRewrites' = 'true'
    )
""")

# Optimized Writes: Spark сам коалесцирует партиции перед записью
# Вместо 200 мелких файлов пишет 10 больших
# Работает для append и overwrite
spark.conf.set("spark.databricks.delta.optimizeWrite.enabled", "true")

# Auto Compaction: после каждой записи Delta проверяет,
# не накопилось ли слишком много мелких файлов, и компактирует автоматически
# Триггерится асинхронно в фоне
spark.conf.set("spark.databricks.delta.autoCompact.enabled", "true")

# Задаём порог авто-компакции: если в партиции больше N файлов,
# запускаем компакцию
spark.conf.set("spark.databricks.delta.autoCompact.minNumFiles", "50")

# Пишем данные - Delta автоматически оптимизирует запись и запускает компакцию
df.write \
    .format("delta") \
    .mode("append") \
    .option("mergeSchema", "true") \
    .save("s3a://bucket/events/")

-- Удалить все файлы, которые Delta не отслеживает и которым > 7 дней (168 часов)
VACUUM events RETAIN 168 HOURS;

-- Режим DRY RUN: показывает, что БЫЛО БЫ удалено, но не удаляет
-- Рекомендуется запускать перед первым VACUUM на production
VACUUM events RETAIN 168 HOURS DRY RUN;

-- ОПАСНО: удалить файлы старше 0 часов
-- Ломает time travel и может сломать concurrent транзакции
-- Delta блокирует это по умолчанию (retentionDurationCheck)
SET spark.databricks.delta.retentionDurationCheck.enabled = false;
VACUUM events RETAIN 0 HOURS;  -- крайне не рекомендуется в production!

from delta import DeltaTable

def vacuum_delta_table(
    spark: SparkSession,
    table_path: str,
    retain_hours: int = 168,   # 7 дней
    dry_run: bool = True       # по умолчанию только показываем
) -> None:
    """Безопасное удаление устаревших файлов через Delta VACUUM."""
    dt = DeltaTable.forPath(spark, table_path)

    if dry_run:
        print("=== DRY RUN: файлы, которые будут удалены ===")
        dt.vacuum(retentionHours=retain_hours)
        # В режиме dry_run нужен spark.conf.set ниже
    else:
        if retain_hours < 168:
            print(f"ПРЕДУПРЕЖДЕНИЕ: retain_hours={retain_hours} < 168 (7 дней)!")
            print("Это может сломать concurrent транзакции и time travel")
            # В production НИКОГДА не уменьшать ниже 168
            return

        dt.vacuum(retentionHours=retain_hours)
        print(f"VACUUM завершён. Удалены файлы старше {retain_hours}ч ({retain_hours//24} дней)")


# Правильный production workflow:
# 1. Сначала OPTIMIZE
optimize_delta_table(
    spark, "s3a://bucket/events/",
    partition_filter="event_date = '2024-01-15'"
)
# 2. Потом VACUUM (через 7+ дней)
vacuum_delta_table(
    spark, "s3a://bucket/events/",
    retain_hours=168,
    dry_run=False
)

-- Базовая компакция с binpack-стратегией
-- binpack: упаковывает мелкие файлы в крупные без изменения порядка строк
CALL catalog.system.rewrite_data_files(
    table => 'mydb.events'
);

-- Компакция только конкретной партиции
CALL catalog.system.rewrite_data_files(
    table => 'mydb.events',
    where => 'event_date = ''2024-01-15'''
);

-- Компакция с настройкой целевого размера файла
CALL catalog.system.rewrite_data_files(
    table => 'mydb.events',
    options => map(
        'target-file-size-bytes', '268435456'   -- 256 МБ
    )
);

-- Компакция с фильтром по размеру: компактировать только файлы < 128 МБ
-- Крупные файлы (уже оптимального размера) пропускаются
CALL catalog.system.rewrite_data_files(
    table => 'mydb.events',
    options => map(
        'target-file-size-bytes',   '268435456',  -- цель: 256 МБ
        'min-file-size-bytes',      '1048576',    -- компактировать файлы > 1 МБ...
        'max-file-size-bytes',      '134217728'   -- ...и < 128 МБ (мелкие)
    )
);

-- Sort-стратегия: перетасовывает данные так, чтобы похожие значения
-- оказались в одних Row Groups - ускоряет запросы с фильтрами
CALL catalog.system.rewrite_data_files(
    table => 'mydb.events',
    strategy => 'sort',
    sort_order => 'user_id ASC NULLS LAST, event_date ASC NULLS LAST'
);

-- Partial progress: для очень больших таблиц
-- Делает промежуточные коммиты каждые N групп файлов
-- Если компакция упадёт на 50% - половина работы уже закоммичена
CALL catalog.system.rewrite_data_files(
    table => 'mydb.events',
    options => map(
        'target-file-size-bytes',       '268435456',
        'partial_progress.enabled',     'true',
        'partial_progress.max_commits', '10'
    )
);

from pyspark.sql import SparkSession


def iceberg_compact(
    spark: SparkSession,
    catalog: str,
    database: str,
    table: str,
    partition_filter: Optional[str] = None,
    strategy: str = "binpack",
    sort_order: Optional[str] = None,
    target_file_size_mb: int = 256,
    min_file_size_mb: int = 1,
    max_file_size_mb: int = 128,
    partial_progress: bool = False,
    max_commits: int = 10
) -> None:
    """
    Выполняет компакцию Iceberg-таблицы через rewrite_data_files.

    strategy:
        'binpack' - упаковывает мелкие файлы без сортировки (быстро)
        'sort'    - сортирует данные (медленнее, но лучше для запросов)
    """
    options = {
        "target-file-size-bytes": str(target_file_size_mb * 1024 * 1024),
        "min-file-size-bytes": str(min_file_size_mb * 1024 * 1024),
        "max-file-size-bytes": str(max_file_size_mb * 1024 * 1024),
    }

    if partial_progress:
        options["partial_progress.enabled"] = "true"
        options["partial_progress.max_commits"] = str(max_commits)

    # Строим SQL опции как Spark map
    options_sql = ", ".join(
        f"'{k}', '{v}'" for k, v in options.items()
    )

    if strategy == "sort" and sort_order:
        sql = f"""
            CALL {catalog}.system.rewrite_data_files(
                table => '{database}.{table}',
                strategy => 'sort',
                sort_order => '{sort_order}',
                options => map({options_sql})
                {f", where => '{partition_filter}'" if partition_filter else ""}
            )
        """
    else:
        sql = f"""
            CALL {catalog}.system.rewrite_data_files(
                table => '{database}.{table}',
                strategy => '{strategy}',
                options => map({options_sql})
                {f", where => '{partition_filter}'" if partition_filter else ""}
            )
        """

    result = spark.sql(sql)
    row = result.first()

    if row:
        print(f"rewrite_data_files результат:")
        print(f"  Файлов переписано: {row['rewritten_data_files_count']}")
        print(f"  Файлов добавлено:  {row['added_data_files_count']}")
        print(f"  Байт переписано:   {row['rewritten_bytes_count'] / 1024**3:.2f} ГБ")


# Binpack - ежедневная компакция вчерашнего раздела
iceberg_compact(
    spark=spark,
    catalog="iceberg",
    database="analytics",
    table="events",
    partition_filter="event_date = '2024-01-14'",
    strategy="binpack",
    target_file_size_mb=256
)

# Sort - еженедельная компакция с сортировкой для ускорения запросов
iceberg_compact(
    spark=spark,
    catalog="iceberg",
    database="analytics",
    table="events",
    strategy="sort",
    sort_order="user_id ASC NULLS LAST, event_date ASC NULLS LAST",
    target_file_size_mb=256,
    partial_progress=True
)

-- Удалить snapshot'ы старше указанного времени
CALL catalog.system.expire_snapshots(
    table => 'mydb.events',
    older_than => TIMESTAMP '2024-01-08 00:00:00'
);

-- Оставить минимум N последних snapshot'ов (независимо от возраста)
CALL catalog.system.expire_snapshots(
    table => 'mydb.events',
    older_than => TIMESTAMP '2024-01-08 00:00:00',
    retain_last => 5
);

-- Удалить snapshot'ы И сразу же физически удалить файлы
-- (иначе нужен отдельный вызов remove_orphan_files)
CALL catalog.system.expire_snapshots(
    table => 'mydb.events',
    older_than => TIMESTAMP '2024-01-08 00:00:00',
    retain_last => 5
    -- delete_data = true  -- осторожно! удаляет файлы немедленно
);

-- Удалить файлы-сироты (orphan files)
-- Это файлы, которые существуют в S3, но не упоминаются ни в одном snapshot'е
-- Может возникнуть после упавшей записи, ручных операций с файлами, etc.
CALL catalog.system.remove_orphan_files(
    table => 'mydb.events',
    older_than => TIMESTAMP '2024-01-08 00:00:00'   -- файлы старше 7 дней
);

from datetime import datetime, timedelta
from pyspark.sql import SparkSession


def iceberg_maintenance(
    spark: SparkSession,
    catalog: str,
    database: str,
    table: str,
    retain_days: int = 7,
    retain_last_snapshots: int = 5
) -> None:
    """
    Полный maintenance цикл для Iceberg таблицы:
    1. expire_snapshots (удаляем старые snapshot'ы)
    2. remove_orphan_files (удаляем файлы-сироты)
    """
    cutoff = datetime.utcnow() - timedelta(days=retain_days)
    cutoff_str = cutoff.strftime("%Y-%m-%d %H:%M:%S")
    full_table = f"{database}.{table}"

    print(f"Iceberg Maintenance: {full_table}")
    print(f"Порог: {cutoff_str} (retain={retain_days} дней)")

    # Шаг 1: expire_snapshots
    result = spark.sql(f"""
        CALL {catalog}.system.expire_snapshots(
            table => '{full_table}',
            older_than => TIMESTAMP '{cutoff_str}',
            retain_last => {retain_last_snapshots}
        )
    """)
    row = result.first()
    if row:
        print(f"expire_snapshots: {row['deleted_data_files_count']} файлов,"
              f" {row['deleted_manifest_files_count']} манифестов удалено")

    # Шаг 2: remove_orphan_files
    # older_than должен быть >= cutoff у expire_snapshots
    orphan_result = spark.sql(f"""
        CALL {catalog}.system.remove_orphan_files(
            table => '{full_table}',
            older_than => TIMESTAMP '{cutoff_str}'
        )
    """)
    orphan_row = orphan_result.first()
    if orphan_row:
        print(f"remove_orphan_files: {orphan_row['orphan_file_location']} файлов удалено")

    print(f"Maintenance завершён: {full_table}")


# Запуск ежедневного maintenance (после ночной компакции)
iceberg_maintenance(
    spark=spark,
    catalog="iceberg",
    database="analytics",
    table="events",
    retain_days=7,
    retain_last_snapshots=10
)

# Пример Airflow DAG для ночной компакции Delta Lake
from airflow import DAG
from airflow.providers.apache.spark.operators.spark_submit import SparkSubmitOperator
from airflow.operators.python import PythonOperator
from datetime import datetime, timedelta


with DAG(
    dag_id="delta_nightly_compaction",
    start_date=datetime(2024, 1, 1),
    schedule_interval="0 2 * * *",   # 02:00 UTC каждый день
    catchup=False,
    default_args={
        "owner": "data-platform",
        "retries": 1,
        "retry_delay": timedelta(minutes=10)
    }
) as dag:

    def get_yesterday_partitions(**context) -> list:
        """Возвращает партиции вчерашнего дня."""
        yesterday = context["ds"]   # YYYY-MM-DD вчерашний день
        return [yesterday]

    # Задача 1: получаем список партиций для компакции
    get_partitions = PythonOperator(
        task_id="get_partitions",
        python_callable=get_yesterday_partitions,
        provide_context=True
    )

    # Задача 2: компакция с OPTIMIZE
    optimize_events = SparkSubmitOperator(
        task_id="optimize_events",
        application="/jobs/compaction/optimize_events.py",
        conf={
            "spark.sql.extensions": "io.delta.sql.DeltaSparkSessionExtension",
            "spark.sql.catalog.spark_catalog":
                "org.apache.spark.sql.delta.catalog.DeltaCatalog",
        },
        application_args=[
            "--table-path", "s3a://bucket/events/",
            "--partition-date", "{{ ds }}"
        ]
    )

    # Задача 3: VACUUM (старше 7 дней)
    vacuum_events = SparkSubmitOperator(
        task_id="vacuum_events",
        application="/jobs/compaction/vacuum_events.py",
        application_args=[
            "--table-path", "s3a://bucket/events/",
            "--retain-hours", "168"
        ]
    )

    get_partitions >> optimize_events >> vacuum_events

from pyspark.sql import SparkSession


def check_compaction_needed(
    spark: SparkSession,
    table_path: str,
    max_small_files: int = 1000,
    min_avg_size_mb: float = 32.0
) -> dict:
    """
    Проверяет, нужна ли компакция на основе метрик файлов.

    Возвращает словарь:
    {
        'needs_compaction': bool,
        'file_count': int,
        'avg_size_mb': float,
        'total_size_gb': float,
        'reason': str
    }
    """
    dt = DeltaTable.forPath(spark, table_path)
    detail = dt.detail().first()

    file_count = detail["numFiles"]
    total_bytes = detail["sizeInBytes"]
    avg_size_mb = (total_bytes / file_count / 1024**2) if file_count > 0 else 0

    reasons = []

    if file_count > max_small_files:
        reasons.append(f"Файлов: {file_count} > {max_small_files}")

    if avg_size_mb < min_avg_size_mb:
        reasons.append(f"Средний размер: {avg_size_mb:.1f} МБ < {min_avg_size_mb} МБ")

    needs_compaction = len(reasons) > 0

    return {
        "needs_compaction": needs_compaction,
        "file_count": file_count,
        "avg_size_mb": avg_size_mb,
        "total_size_gb": total_bytes / 1024**3,
        "reason": "; ".join(reasons) if reasons else "Компакция не нужна"
    }


def smart_compaction(spark: SparkSession, table_path: str) -> None:
    """Запускает компакцию только если нужно."""
    metrics = check_compaction_needed(spark, table_path)

    print(f"Таблица: {table_path}")
    print(f"Файлов: {metrics['file_count']}")
    print(f"Средний размер: {metrics['avg_size_mb']:.1f} МБ")
    print(f"Общий объём: {metrics['total_size_gb']:.2f} ГБ")

    if not metrics["needs_compaction"]:
        print(f"✓ Компакция не нужна: {metrics['reason']}")
        return

    print(f"✗ Нужна компакция: {metrics['reason']}")
    optimize_delta_table(spark, table_path)
    print("✓ Компакция завершена")

from pyspark.sql import SparkSession, DataFrame
from pyspark.sql.streaming import StreamingQuery


class StreamingCompactionManager:
    """
    Управляет компакцией для стриминговой таблицы.
    Запускает компакцию каждые N батчей или по таймеру.
    """

    def __init__(
        self,
        spark: SparkSession,
        output_path: str,
        compaction_interval_batches: int = 20,
        target_file_size_mb: int = 256
    ):
        self.spark = spark
        self.output_path = output_path
        self.compaction_interval = compaction_interval_batches
        self.target_file_size_mb = target_file_size_mb
        self.batch_count = 0

    def process_batch(self, df: DataFrame, batch_id: int) -> None:
        """
        foreachBatch handler:
        1. Пишет батч с coalesce (предотвращаем мелкие файлы)
        2. Каждые N батчей запускает OPTIMIZE
        """
        # Считаем размер батча для оптимального числа файлов
        row_count = df.count()
        # Предполагаем ~1 КБ на строку (подстраивайте под реальные данные)
        estimated_bytes = row_count * 1024
        target_files = calculate_target_partitions(
            total_bytes=estimated_bytes,
            target_file_size_mb=64   # для стриминга - файлы меньше
        )

        # Запись с coalesce: предотвращаем сотни мелких файлов
        (
            df
            .coalesce(target_files)
            .write
            .format("delta")
            .mode("append")
            .save(self.output_path)
        )

        self.batch_count += 1

        # Каждые N батчей - OPTIMIZE
        if self.batch_count % self.compaction_interval == 0:
            print(f"[Batch {batch_id}] Запуск компакции (каждые {self.compaction_interval} батчей)")
            optimize_delta_table(
                self.spark,
                self.output_path,
                # Компактируем только последние 2 часа данных
                partition_filter="event_date >= current_date() - interval 1 day"
            )


def start_streaming_with_compaction(spark: SparkSession) -> StreamingQuery:
    """Запускает стриминг с периодической компакцией."""
    manager = StreamingCompactionManager(
        spark=spark,
        output_path="s3a://bucket/events/",
        compaction_interval_batches=20,   # компактировать каждые 20 батчей
        target_file_size_mb=256
    )

    return (
        spark.readStream
        .format("kafka")
        .option("kafka.bootstrap.servers", "kafka:9092")
        .option("subscribe", "events")
        .load()
        .writeStream
        .foreachBatch(manager.process_batch)
        .trigger(processingTime="1 minute")
        .option("checkpointLocation", "s3a://bucket/checkpoints/events/")
        .start()
    )

# Delta Lake: OPTIMIZE с Z-Order
def optimize_with_zorder(
    spark: SparkSession,
    table_path: str,
    partition_filter: str,
    zorder_columns: list
) -> None:
    """
    Компакция с Z-Order кластеризацией.
    Z-Order оптимален для запросов с фильтрами по нескольким полям.
    """
    dt = DeltaTable.forPath(spark, table_path)

    print(f"OPTIMIZE ZORDER BY {zorder_columns}")
    result = (
        dt.optimize()
        .where(partition_filter)
        .executeZOrderBy(*zorder_columns)
    )

    row = result.first()
    if row:
        print(f"Завершено: {row['numFilesRemoved']} → {row['numFilesAdded']} файлов")


# Пример: компакция таблицы events с Z-Order по user_id + event_type
optimize_with_zorder(
    spark=spark,
    table_path="s3a://bucket/events/",
    partition_filter="event_date >= '2024-01-01'",
    zorder_columns=["user_id", "event_type"]
)

-- Iceberg: sort strategy = аналог Z-Order
-- Сортирует данные во время rewrite_data_files
CALL catalog.system.rewrite_data_files(
    table => 'analytics.events',
    strategy => 'sort',
    sort_order => 'user_id ASC NULLS LAST, event_date ASC NULLS LAST'
);

-- Iceberg также поддерживает hidden partitioning с z-order:
ALTER TABLE analytics.events
WRITE ORDERED BY user_id, event_date;
-- Все будущие записи будут автоматически сортироваться

# НЕПРАВИЛЬНО: компактируем всю таблицу каждый час
# Для таблицы 10 ТБ: каждый час читаем и перезаписываем 10 ТБ данных
# Стоимость: 10 ТБ / час × 24 ч = 240 ТБ I/O в день
# Это парализует кластер и съест весь бюджет S3 API

schedule_interval = "0 * * * *"  # каждый час

def bad_hourly_optimize():
    optimize_delta_table(spark, "s3a://bucket/huge_events/")  # всё сразу!

# ПРАВИЛЬНО: компактируем только вчерашний раздел раз в день
schedule_interval = "0 2 * * *"  # 02:00 UTC каждый день

def good_daily_optimize():
    yesterday = (datetime.utcnow() - timedelta(days=1)).strftime("%Y-%m-%d")
    optimize_delta_table(
        spark,
        "s3a://bucket/events/",
        partition_filter=f"event_date = '{yesterday}'"
    )

-- НЕПРАВИЛЬНО: удаляем файлы сразу после OPTIMIZE
-- Это ломает time travel, concurrent транзакции и streaming checkpoints
SET spark.databricks.delta.retentionDurationCheck.enabled = false;
VACUUM events RETAIN 0 HOURS;  -- НИКОГДА не делайте это в production!

-- ПРАВИЛЬНО: минимум 7 дней retention
VACUUM events RETAIN 168 HOURS;

# НЕПРАВИЛЬНО: один файл на партицию
# При большом объёме данных один Executor пишет весь раздел
# Время записи возрастает, параллелизм потерян
df.write.partitionBy("event_date").coalesce(1).parquet(path)

# ПРАВИЛЬНО: оптимальное число файлов
target = calculate_target_file_count(total_bytes, target_file_size_mb=256)
df.write.partitionBy("event_date").repartition(target).parquet(path)

# НЕПРАВИЛЬНО: компактировать в рабочее время
# OPTIMIZE требует много I/O, конкурирует с аналитическими запросами
schedule_interval = "0 9 * * 1-5"  # 09:00 в рабочие дни - плохо!

# ПРАВИЛЬНО: компактировать ночью или в выходные
schedule_interval = "0 2 * * *"   # 02:00 UTC ежедневно
# Или по выходным для исторических данных:
schedule_interval = "0 3 * * 0"   # воскресенье 03:00

# НЕПРАВИЛЬНО: только expire_snapshots без remove_orphan_files
# После expire_snapshots файлы из старых snapshot'ов становятся "сиротами"
# Они физически существуют на S3, но Iceberg их не видит
# Это создаёт скрытые расходы на хранение

# Только это:
spark.sql("CALL iceberg.system.expire_snapshots(...)")
# Забыли:
# spark.sql("CALL iceberg.system.remove_orphan_files(...)")

# ПРАВИЛЬНО: всегда запускать оба шага
iceberg_maintenance(spark, ...)  # expire_snapshots + remove_orphan_files

from pyspark.sql import SparkSession


def audit_table_health(
    spark: SparkSession,
    table_path: str,
    format: str = "delta"
) -> dict:
    """
    Собирает метрики здоровья таблицы для принятия решения о компакции.
    """
    if format == "delta":
        dt = DeltaTable.forPath(spark, table_path)
        detail = dt.detail().first()

        file_count = detail["numFiles"]
        total_bytes = detail["sizeInBytes"]
        avg_size_mb = (total_bytes / file_count / 1024**2) if file_count > 0 else 0

        # История операций (последние 10)
        history = dt.history(10).select(
            "version", "timestamp", "operation", "operationMetrics"
        )

    elif format == "parquet":
        # Для обычного Parquet - считаем файлы через Hadoop API
        sc = spark.sparkContext
        hadoop_conf = sc._jvm.org.apache.hadoop.conf.Configuration()
        fs = sc._jvm.org.apache.hadoop.fs.FileSystem.get(
            sc._jvm.java.net.URI.create(table_path), hadoop_conf
        )
        status_list = fs.listStatus(
            sc._jvm.org.apache.hadoop.fs.Path(table_path)
        )
        file_sizes = [s.getLen() for s in status_list if not s.isDirectory()]
        file_count = len(file_sizes)
        total_bytes = sum(file_sizes)
        avg_size_mb = (total_bytes / file_count / 1024**2) if file_count > 0 else 0
        history = None
    else:
        raise ValueError(f"Неизвестный формат: {format}")

    # Классификация здоровья
    if avg_size_mb < 10:
        health = "CRITICAL"
        action = "Немедленная компакция!"
    elif avg_size_mb < 64:
        health = "WARNING"
        action = "Компакция рекомендована"
    elif avg_size_mb < 512:
        health = "OK"
        action = "Компакция не нужна"
    else:
        health = "LARGE_FILES"
        action = "Файлы слишком большие для оптимального параллелизма"

    metrics = {
        "path": table_path,
        "format": format,
        "file_count": file_count,
        "total_size_gb": total_bytes / 1024**3,
        "avg_size_mb": avg_size_mb,
        "health": health,
        "action": action
    }

    # Распределение размеров файлов
    if format == "delta" and file_count > 0:
        # Через Delta log можно получить distribution
        files_df = spark.read.format("delta").load(table_path) \
            ._jdf.queryExecution().analyzed()
        # В реальном коде: dt.detail() имеет sizeDistribution в новых версиях Delta

    return metrics


def print_audit_report(metrics: dict) -> None:
    """Форматированный вывод отчёта."""
    status_icon = {
        "CRITICAL": "🔴",
        "WARNING":  "🟡",
        "OK":       "🟢",
        "LARGE_FILES": "🔵"
    }.get(metrics["health"], "⚪")

    print(f"\n{'='*50}")
    print(f"{status_icon} Таблица: {metrics['path']}")
    print(f"  Формат:       {metrics['format']}")
    print(f"  Файлов:       {metrics['file_count']:,}")
    print(f"  Объём:        {metrics['total_size_gb']:.2f} ГБ")
    print(f"  Средний файл: {metrics['avg_size_mb']:.1f} МБ")
    print(f"  Статус:       {metrics['health']}")
    print(f"  Рекомендация: {metrics['action']}")
    print(f"{'='*50}\n")

from datetime import datetime, timedelta
from pyspark.sql import SparkSession


def run_production_compaction_pipeline(
    spark: SparkSession,
    tables: list[dict]
) -> None:
    """
    Production pipeline компакции для нескольких таблиц.

    tables: список словарей с описанием таблиц:
    [
        {
            "path": "s3a://bucket/events/",
            "format": "delta",
            "partition_column": "event_date",
            "zorder_columns": ["user_id", "event_type"],
            "target_file_size_mb": 256
        },
        {
            "path": "s3a://bucket/users/",
            "format": "iceberg",
            "catalog": "iceberg",
            "database": "analytics",
            "table": "users",
            "strategy": "binpack",
            "target_file_size_mb": 128
        }
    ]
    """
    yesterday = (datetime.utcnow() - timedelta(days=1)).strftime("%Y-%m-%d")
    results = []

    for table_config in tables:
        fmt = table_config["format"]
        path = table_config["path"]

        print(f"\n{'='*60}")
        print(f"Обработка: {path} ({fmt})")

        # Шаг 1: проверяем, нужна ли компакция
        metrics = audit_table_health(spark, path, fmt)
        print_audit_report(metrics)

        if metrics["health"] in ("OK",) and metrics["file_count"] < 100:
            print("Пропускаем - компакция не нужна")
            results.append({"path": path, "action": "skipped"})
            continue

        # Шаг 2: компакция по формату
        if fmt == "delta":
            partition_col = table_config.get("partition_column", "event_date")
            partition_filter = f"{partition_col} = '{yesterday}'"
            zorder_cols = table_config.get("zorder_columns")

            optimize_delta_table(
                spark, path,
                partition_filter=partition_filter,
                zorder_columns=zorder_cols
            )
            results.append({"path": path, "action": "delta_optimize"})

        elif fmt == "iceberg":
            iceberg_compact(
                spark=spark,
                catalog=table_config["catalog"],
                database=table_config["database"],
                table=table_config["table"],
                partition_filter=f"event_date = '{yesterday}'",
                strategy=table_config.get("strategy", "binpack"),
                target_file_size_mb=table_config.get("target_file_size_mb", 256)
            )
            results.append({"path": path, "action": "iceberg_compact"})

    # Шаг 3: итоговый отчёт
    print(f"\n{'='*60}")
    print("Итоговый отчёт компакции:")
    for r in results:
        print(f"  {r['path']}: {r['action']}")


# Запуск pipeline
tables = [
    {
        "path": "s3a://bucket/events/",
        "format": "delta",
        "partition_column": "event_date",
        "zorder_columns": ["user_id", "event_type"],
        "target_file_size_mb": 256
    },
    {
        "path": "s3a://bucket/sessions/",
        "format": "delta",
        "partition_column": "session_date",
        "target_file_size_mb": 128
    }
]

run_production_compaction_pipeline(spark, tables)

# День 1: экстренная компакция всей таблицы через OPTIMIZE
# Запускаем на ночь (02:00), ждём 4 часа
spark.sql("""
    OPTIMIZE analytics.events
""")
# Результат: 4 320 000 файлов → 1 800 файлов (средний 90 МБ)
# count() теперь: 12 секунд (вместо 3 часов)

# День 2: настраиваем ежедневный OPTIMIZE только вчерашних партиций
spark.sql("""
    OPTIMIZE analytics.events
    WHERE event_date = '2024-01-14'
""")
# Время: 2 минуты для одной партиции

# Параллельно: фиксируем стриминг с coalesce
# Вместо 50 файлов на батч - 2 файла
def fixed_batch_handler(df: DataFrame, batch_id: int) -> None:
    row_count = df.count()
    # Предполагаем ~500 байт на строку; 64 МБ целевой файл
    target_files = max(1, row_count * 500 // (64 * 1024 * 1024))
    df.coalesce(target_files).write.format("delta").mode("append").save(path)

# День 3: VACUUM для физической очистки (файлы старше 7 дней)
spark.sql("VACUUM analytics.events RETAIN 168 HOURS")
# S3: 77 ГБ старых файлов удалено