Форматы файлов в HDFS: ORC vs Parquet vs Avro, Snappy vs ZSTD vs Gzip - когда что выбирать

[user_id=1, name="Alice", amount=500, date="2024-01-15", status="active"]
[user_id=2, name="Bob",   amount=200, date="2024-01-15", status="closed"]
[user_id=3, name="Carol", amount=800, date="2024-01-16", status="active"]

[user_id]:  1,     2,     3,     4,     5
[name]:     Alice, Bob,   Carol, Dave,  Eve
[amount]:   500,   200,   800,   150,   300
[date]:     01-15, 01-15, 01-16, 01-15, 01-16
[status]:   active,closed,active,active,closed

# Исходная схема (версия 1.0)
schema_v1 = {
    "type": "record",
    "name": "UserEvent",
    "fields": [
        {"name": "user_id", "type": "long"},
        {"name": "event_type", "type": "string"},
        {"name": "timestamp", "type": "long"}
    ]
}

# Обновлённая схема (версия 2.0) - добавили новые поля
# Старые данные в Avro-файлах за прошлые месяцы остаются читаемыми!
schema_v2 = {
    "type": "record",
    "name": "UserEvent",
    "fields": [
        {"name": "user_id", "type": "long"},
        {"name": "event_type", "type": "string"},
        {"name": "timestamp", "type": "long"},
        # Новое поле: если его нет в старых данных - подставляется default
        {"name": "session_id", "type": ["null", "string"], "default": None},
        # Ещё одно новое поле
        {"name": "platform", "type": "string", "default": "web"}
    ]
}

# При чтении старых данных с новой схемой:
# - session_id будет null (default значение)
# - platform будет "web" (default значение)
# Никаких ошибок, никакой потери данных!

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

spark = SparkSession.builder \
    .appName("avro-bronze-ingestion") \
    .config("spark.jars.packages", "org.apache.spark:spark-avro_2.12:3.5.1") \
    .getOrCreate()

# Чтение потока событий из Kafka (Avro с Schema Registry)
kafka_df = spark.readStream \
    .format("kafka") \
    .option("kafka.bootstrap.servers", "kafka:9092") \
    .option("subscribe", "user_events") \
    .load()

# Десериализация Avro (байты → DataFrame)
# Schema Registry хранит схемы, Spark получает их по schema_id
from pyspark.sql.avro.functions import from_avro

avro_schema = """
{
  "type": "record",
  "name": "UserEvent",
  "fields": [
    {"name": "user_id", "type": "long"},
    {"name": "event_type", "type": "string"},
    {"name": "amount", "type": ["null", "double"], "default": null},
    {"name": "timestamp", "type": "long"}
  ]
}
"""

events_df = kafka_df.select(
    from_avro(F.col("value"), avro_schema).alias("event")
).select("event.*")

# Записываем в Bronze слой как Avro + Snappy
# Сохраняем ОРИГИНАЛЬНЫЙ формат - это важно для Bronze!
# Bronze должен хранить данные максимально близко к источнику
query = events_df.writeStream \
    .format("avro") \
    .option("path", "hdfs://cluster/data/bronze/user_events/") \
    .option("checkpointLocation", "/tmp/checkpoints/bronze/user_events") \
    .partitionBy("event_type") \
    .option("compression", "snappy") \
    .trigger(processingTime="30 seconds") \
    .start()

spark = SparkSession.builder \
    # Vectorized Reader включён по умолчанию.
    # Отключите только если видите проблемы с NULL-обработкой или UDF
    .config("spark.sql.parquet.enableVectorizedReader", "true") \

    # Размер батча: 4096 строк - баланс между памятью и эффективностью SIMD.
    # Увеличьте до 8192 если колонок мало и они числовые (лучше SIMD).
    # Уменьшите до 2048 если много широких строк (ограничение памяти Executor).
    .config("spark.sql.parquet.columnarReaderBatchSize", "4096") \

    # Если файлы Parquet содержат глубокие вложенные структуры (ARRAY of STRUCT),
    # Vectorized Reader не работает и Spark падает на Java-ридер автоматически.
    # В этом случае можно явно отключить:
    # .config("spark.sql.parquet.enableVectorizedReader", "false") \
    .getOrCreate()

Файл 10 GB = один непрерывный gzip-поток
Нет маркеров блоков, нельзя начать чтение с середины
→ Один Spark Task читает ВСЮ таблицу
→ Параллелизм = 1, независимо от числа Executor'ов

# Демонстрация проблемы несплинтуемых кодеков

# ❌ АНТИПАТТЕРН: CSV + Gzip = не сплинтуемый!
# Один файл 10 GB = одна Spark-задача
df.write \
    .option("compression", "gzip") \
    .csv("hdfs://cluster/data/bad_idea.csv.gz")

# Проверим что получилось:
df_bad = spark.read.csv("hdfs://cluster/data/bad_idea.csv.gz")
print(df_bad.rdd.getNumPartitions())
# Вывод: 1 (одна партиция для всего файла!)

# ✅ ПРАВИЛЬНО: Parquet + Gzip = сплинтуемый!
# (Gzip применяется к каждому Page отдельно)
df.write \
    .option("compression", "gzip") \
    .parquet("hdfs://cluster/data/good.parquet")

df_good = spark.read.parquet("hdfs://cluster/data/good.parquet")
print(df_good.rdd.getNumPartitions())
# Вывод: 80 (80 Row Groups = 80 партиций)

# Snappy - дефолтный кодек для Parquet в Spark:
spark.conf.get("spark.sql.parquet.compression.codec")
# Вывод: snappy

# Это намеренное решение: Snappy даёт предсказуемую производительность
# на горячих данных Silver/Gold, которые читаются сотни раз в день.
# CPU-overhead на распаковку минимален.

# Явное указание:
df.write \
    .mode("overwrite") \
    .option("compression", "snappy") \
    .parquet("hdfs://cluster/data/silver/events/")

# Через глобальную конфигурацию:
spark.conf.set("spark.sql.parquet.compression.codec", "snappy")

spark = SparkSession.builder \
    # ZSTD с дефолтным уровнем (3) для Parquet
    .config("spark.sql.parquet.compression.codec", "zstd") \

    # ZSTD для ORC
    .config("spark.sql.orc.compression.codec", "zstd") \

    .getOrCreate()

# Запись с явным уровнем сжатия ZSTD для Parquet:
df.write \
    .mode("overwrite") \
    .option("compression", "zstd") \
    # Уровень задаётся через отдельную опцию (Spark 3.2+)
    .option("parquet.compression.codec.zstd.level", "9") \
    .parquet("hdfs://cluster/data/cold/events_2022/")

# Для горячих Silver данных (уровень 3 - оптимально):
df_silver.write \
    .mode("overwrite") \
    .option("compression", "zstd") \
    .option("parquet.compression.codec.zstd.level", "3") \
    .partitionBy("event_date") \
    .parquet("hdfs://cluster/data/silver/events/")

# Антипаттерн: CSV + Gzip = несплинтуемый!
df_bad = spark.read.csv("hdfs://cluster/data/logs.csv.gz")
# HDFS report:
# Файл: 10 GB на диске, но один блок для Spark!
# Spark создаст: 1 Task (не 80!)
# Производительность: 1/80 от максимальной

# НО: Parquet + Gzip = сплинтуемый!
df_ok = spark.read.parquet("hdfs://cluster/data/logs_parquet_gzip/")
# Каждый Row Group сжат отдельным gzip-потоком
# Spark создаст: 80 Tasks
# Производительность: максимальная

from pyspark.sql import SparkSession

# ── Сессия для Silver/Gold слоя (горячие данные) ──────────────────────
spark_hot = SparkSession.builder \
    .appName("silver-gold-etl") \

    # Parquet + Snappy: максимальная скорость чтения для интерактивной аналитики
    .config("spark.sql.parquet.compression.codec", "snappy") \

    # Vectorized Reader для колоночного чтения через SIMD
    .config("spark.sql.parquet.enableVectorizedReader", "true") \
    .config("spark.sql.parquet.columnarReaderBatchSize", "4096") \

    # Predicate Pushdown: фильтры применяются на уровне Parquet Footer
    .config("spark.sql.parquet.filterPushdown", "true") \

    # Row Group статистики: Spark использует min/max для пропуска Row Groups
    .config("spark.sql.parquet.recordLevelFilter.enabled", "true") \

    # Размер Row Group = размер блока HDFS = одна Task = оптимальная locality
    # 128 MB - дефолт Parquet, должен совпадать с dfs.blocksize
    .config("spark.hadoop.parquet.block.size", "134217728") \

    .getOrCreate()


# ── Сессия для Cold Archive (архивные данные) ─────────────────────────
spark_cold = SparkSession.builder \
    .appName("archive-etl") \

    # Parquet + ZSTD для максимального сжатия при хорошей скорости чтения
    .config("spark.sql.parquet.compression.codec", "zstd") \

    # ORC для архивов в Hive-экосистеме
    .config("spark.sql.orc.compression.codec", "zstd") \

    .getOrCreate()

# Уровень ZSTD задаётся при записи:
df_cold.write \
    .option("compression", "zstd") \
    .option("parquet.compression.codec.zstd.level", "9") \
    .parquet("hdfs://cluster/data/archive/2022/")


# ── Сессия для Avro (Bronze/Ingestion) ────────────────────────────────
spark_bronze = SparkSession.builder \
    .appName("kafka-bronze-ingestion") \
    .config("spark.jars.packages", "org.apache.spark:spark-avro_2.12:3.5.1") \
    .getOrCreate()

# Ключевой принцип: Row Group size = HDFS block size
# Это обеспечивает: 1 Row Group = 1 HDFS Block = 1 Spark Task = NODE_LOCAL чтение

# Если HDFS blocksize = 128 MB (стандарт):
spark.conf.set("spark.hadoop.parquet.block.size", "134217728")  # 128 MB

# Если HDFS blocksize = 256 MB (крупные файлы, меньше Tasks):
spark.conf.set("spark.hadoop.parquet.block.size", "268435456")  # 256 MB

# Для ORC:
spark.conf.set("spark.hadoop.orc.stripe.size", "67108864")  # 64 MB (ORC default)
# Или увеличить для крупных таблиц:
spark.conf.set("spark.hadoop.orc.stripe.size", "134217728")  # 128 MB

# Размер Page внутри Row Group (влияет на granularity статистики):
# Меньше Page = точнее predicate pushdown, но больше метаданных
spark.conf.set("spark.hadoop.parquet.page.size", "1048576")  # 1 MB (default)

from pyspark.sql import SparkSession, functions as F
from datetime import date

spark = SparkSession.builder \
    .appName("medallion-formats-demo") \
    .enableHiveSupport() \
    .getOrCreate()

# ── Bronze Layer: Avro + Snappy (источник - Kafka с Debezium CDC) ─────
def write_bronze_avro(df, target_path: str, date_str: str) -> None:
    """
    Bronze слой: сохраняем данные максимально близко к источнику.
    Avro + Snappy оптимален для потоковой записи из Kafka/Debezium:
    - Схема хранится в файле (самодостаточный файл)
    - Schema Evolution (добавление колонок в исходной системе = OK)
    - Snappy: минимальный CPU overhead при высокоскоростной записи
    """
    df.write \
        .format("avro") \
        .mode("append") \
        .option("compression", "snappy") \
        .partitionBy("ingestion_date") \
        .save(f"{target_path}/date={date_str}/")


# ── Silver Layer: Parquet + Snappy (горячие очищенные данные) ─────────
def write_silver_parquet(df, table_name: str, date_str: str) -> None:
    """
    Silver слой: очищенные и нормализованные данные.
    Parquet + Snappy: оптимально для Spark SQL аналитики.
    - Columnar: Column Pruning (Spark читает только нужные колонки)
    - Snappy: быстрая распаковка при интерактивных запросах
    - Vectorized Reader: SIMD-оптимизация
    - partitionBy(date): partition pruning при ежедневных запросах
    """
    spark.conf.set("spark.sql.parquet.compression.codec", "snappy")

    df.repartition(50) \
        .write \
        .mode("overwrite") \
        .format("parquet") \
        .option("compression", "snappy") \
        .partitionBy("event_date") \
        .saveAsTable(f"silver.{table_name}")


# ── Gold Layer: Parquet + Snappy или ZSTD ──────────────────────────────
def write_gold_parquet(df, table_name: str, use_zstd: bool = False) -> None:
    """
    Gold слой: агрегаты и витрины для BI.
    - Если данные часто читаются → Snappy
    - Если данные большие и читаются реже → ZSTD (экономия 30%)
    """
    codec = "zstd" if use_zstd else "snappy"

    # Gold обычно небольшой после агрегации → меньше партиций
    df.coalesce(10) \
        .write \
        .mode("overwrite") \
        .format("parquet") \
        .option("compression", codec) \
        .saveAsTable(f"gold.{table_name}")


# ── Cold Archive: Parquet + ZSTD Level 9 ──────────────────────────────
def archive_to_cold_storage(source_table: str, target_path: str,
                             year: int, month: int) -> int:
    """
    Архивирование исторических данных в Cold Storage.
    Parquet + ZSTD Level 9:
    - Максимальное сжатие (читаются раз в квартал)
    - Скорость чтения почти не страдает (ZSTD decode быстрый)
    - Экономия 40-50% vs Snappy на архивных числовых данных
    """
    df = spark.table(source_table) \
        .filter(F.year("event_date") == year) \
        .filter(F.month("event_date") == month)

    row_count = df.count()

    # Максимальный ZSTD: пишется медленно, читается быстро
    df.repartition(20) \
        .write \
        .mode("overwrite") \
        .format("parquet") \
        .option("compression", "zstd") \
        .option("parquet.compression.codec.zstd.level", "9") \
        .save(f"{target_path}/year={year}/month={month:02d}/")

    print(f"Архивировано {row_count:,} строк за {year}-{month:02d}")
    print(f"  Codec: ZSTD Level 9")
    return row_count

# Сравниваем физический размер файлов в разных форматах/кодеках
# После записи одного датасета в четырёх вариантах:

hdfs dfs -du -h /data/benchmark/
# Вывод (пример для 100M строк транзакционных данных):
# 3.2 G   9.6 G   /data/benchmark/json_no_compression    # JSON без сжатия
# 1.8 G   5.4 G   /data/benchmark/csv_gzip               # CSV + Gzip
# 1.5 G   4.5 G   /data/benchmark/parquet_snappy         # Parquet + Snappy
# 1.1 G   3.3 G   /data/benchmark/parquet_zstd_3         # Parquet + ZSTD L3
# 0.9 G   2.7 G   /data/benchmark/parquet_zstd_9         # Parquet + ZSTD L9
# 0.8 G   2.4 G   /data/benchmark/orc_zstd               # ORC + ZSTD

# Второе число = размер × replication factor (3x)
# Для экономии дисков смотрим на первое число

# lab_file_formats_benchmark.py
# Сравниваем Avro, Parquet, ORC с разными кодеками

import time
import subprocess
from dataclasses import dataclass
from pyspark.sql import SparkSession, functions as F


@dataclass
class BenchmarkResult:
    name: str
    format: str
    codec: str
    write_time_sec: float
    disk_size_mb: float
    read_full_time_sec: float
    read_filtered_time_sec: float
    compression_ratio: float


def create_spark() -> SparkSession:
    return (
        SparkSession.builder
        .master("yarn")
        .appName("file-formats-benchmark")
        .config("spark.jars.packages",
                "org.apache.spark:spark-avro_2.12:3.5.1")
        .config("spark.executor.instances", "10")
        .config("spark.executor.cores", "4")
        .config("spark.executor.memory", "8g")
        .getOrCreate()
    )


def generate_dataset(spark: SparkSession, n_rows: int = 10_000_000):
    """Генерирует тестовый датасет транзакций."""
    return spark.range(n_rows).select(
        F.col("id").alias("txn_id"),
        (F.rand() * 1_000_000).cast("long").alias("user_id"),
        (F.rand() * 50_000).cast("long").alias("merchant_id"),
        F.array(
            F.lit("purchase"), F.lit("refund"),
            F.lit("transfer"), F.lit("fee")
        ).getItem((F.rand() * 4).cast("int")).alias("txn_type"),
        (F.rand() * 10_000).alias("amount"),
        F.array(F.lit("RUB"), F.lit("USD"), F.lit("EUR")).getItem(
            (F.rand() * 3).cast("int")
        ).alias("currency"),
        F.date_add(F.lit("2024-01-01"), (F.rand() * 365).cast("int")).alias("txn_date"),
        F.array(
            F.lit("web"), F.lit("mobile"), F.lit("pos"), F.lit("api")
        ).getItem((F.rand() * 4).cast("int")).alias("channel"),
    )


def get_hdfs_size_mb(path: str) -> float:
    """Получает реальный размер директории на HDFS."""
    result = subprocess.run(
        ["hdfs", "dfs", "-du", "-s", path],
        capture_output=True, text=True
    )
    parts = result.stdout.split()
    if len(parts) >= 1:
        return int(parts[0]) / 1024 / 1024
    return 0.0


def run_benchmark(
    spark: SparkSession,
    df,
    name: str,
    fmt: str,
    codec: str,
    base_path: str,
) -> BenchmarkResult:
    """Запускает один бенчмарк: пишет, измеряет размер, читает."""
    path = f"{base_path}/{name}/"
    print(f"\nБенчмарк: {name} (format={fmt}, codec={codec})")

    # Запись
    t0 = time.time()
    if fmt == "avro":
        df.write.format("avro").mode("overwrite") \
            .option("compression", codec).save(path)
    else:
        df.write.format(fmt).mode("overwrite") \
            .option("compression", codec).save(path)
    write_time = time.time() - t0
    print(f"  Запись: {write_time:.1f}с")

    # Размер на диске
    disk_mb = get_hdfs_size_mb(path)
    original_mb = get_hdfs_size_mb(f"{base_path}/parquet_snappy/")
    compression_ratio = disk_mb / original_mb if original_mb > 0 else 1.0
    print(f"  Размер: {disk_mb:.0f} MB ({compression_ratio:.2f}x vs Parquet+Snappy)")

    # Полное сканирование
    t0 = time.time()
    if fmt == "avro":
        spark.read.format("avro").load(path).agg(F.sum("amount")).collect()
    else:
        spark.read.format(fmt).load(path).agg(F.sum("amount")).collect()
    read_full = time.time() - t0
    print(f"  Полное чтение (SUM): {read_full:.1f}с")

    # Фильтрованный запрос (проверяем predicate pushdown + column pruning)
    t0 = time.time()
    if fmt == "avro":
        spark.read.format("avro").load(path) \
            .filter(F.col("txn_type") == "purchase") \
            .groupBy("currency") \
            .agg(F.sum("amount"), F.count("*")) \
            .collect()
    else:
        spark.read.format(fmt).load(path) \
            .filter(F.col("txn_type") == "purchase") \
            .groupBy("currency") \
            .agg(F.sum("amount"), F.count("*")) \
            .collect()
    read_filtered = time.time() - t0
    print(f"  Фильтрованное чтение (WHERE + GROUP BY): {read_filtered:.1f}с")

    return BenchmarkResult(
        name=name, format=fmt, codec=codec,
        write_time_sec=write_time,
        disk_size_mb=disk_mb,
        read_full_time_sec=read_full,
        read_filtered_time_sec=read_filtered,
        compression_ratio=compression_ratio,
    )


if __name__ == "__main__":
    spark = create_spark()
    BASE_PATH = "hdfs://cluster/benchmark/file-formats"
    df = generate_dataset(spark, n_rows=10_000_000)

    # Кешируем в памяти чтобы генерация не влияла на write benchmark
    df.cache()
    df.count()

    configs = [
        ("parquet_snappy",  "parquet", "snappy"),
        ("parquet_zstd_3",  "parquet", "zstd"),
        ("parquet_gzip",    "parquet", "gzip"),
        ("orc_snappy",      "orc",     "snappy"),
        ("orc_zstd",        "orc",     "zstd"),
        ("avro_snappy",     "avro",    "snappy"),
    ]

    results = []
    for name, fmt, codec in configs:
        r = run_benchmark(spark, df, name, fmt, codec, BASE_PATH)
        results.append(r)

    # Итоговая таблица
    print("\n" + "=" * 80)
    print(f"{'Формат':20s} {'Запись':>8} {'Диск MB':>9} {'Ratio':>6} {'Полн. чт':>10} {'Фильтр':>10}")
    print("-" * 80)
    for r in sorted(results, key=lambda x: x.read_filtered_time_sec):
        print(f"{r.name:20s} {r.write_time_sec:>7.1f}s {r.disk_size_mb:>8.0f} "
              f"{r.compression_ratio:>5.2f}x {r.read_full_time_sec:>9.1f}s "
              f"{r.read_filtered_time_sec:>9.1f}s")

    spark.stop()

Bronze:  Avro + Snappy    (быстрая запись из потока, schema evolution)
Silver:  Parquet + Snappy (быстрое чтение, интерактивная аналитика)
Gold:    Parquet + ZSTD   (экономия дисков, хорошее чтение)
Archive: Parquet + ZSTD 9 (максимальная компрессия, редкие чтения)