Small Files Disease: причины, последствия и влияние на планировщик

# Проверить средний размер файлов в таблице
import os

def check_file_sizes(path):
    """Диагностика размера файлов через os.walk."""
    sizes = []
    for root, dirs, files in os.walk(path):
        for f in files:
            if f.endswith((".parquet", ".orc")):
                full_path = os.path.join(root, f)
                sizes.append(os.path.getsize(full_path))

    if not sizes:
        print("Файлы не найдены")
        return

    avg_mb = sum(sizes) / len(sizes) / (1024 * 1024)
    min_mb = min(sizes) / (1024 * 1024)
    max_mb = max(sizes) / (1024 * 1024)

    print(f"Файлов: {len(sizes):,}")
    print(f"Средний размер: {avg_mb:.2f} МБ")
    print(f"Минимальный: {min_mb:.3f} МБ")
    print(f"Максимальный: {max_mb:.2f} МБ")

    if avg_mb < 16:
        print("КРИТИЧНО: Small Files Disease в тяжёлой форме")
    elif avg_mb < 64:
        print("ВНИМАНИЕ: файлы меньше оптимального размера")
    else:
        print("Норма: размер файлов в допустимом диапазоне")

check_file_sizes("/tmp/events_by_date/")

from pyspark.sql import functions as F

# Демонстрация: 50 МБ данных, 200 Tasks → 200 файлов по ~250 КБ
small_df = spark.range(1, 100_001) \
    .withColumn("value", F.rand()) \
    .withColumn("dt", F.lit("2026-01-15"))

# Количество shuffle partitions по умолчанию: 200
print(spark.conf.get("spark.sql.shuffle.partitions"))  # → 200

# Запись: Spark создаст ~200 файлов
small_df.repartition(200).write \
    .mode("overwrite") \
    .partitionBy("dt") \
    .parquet("/tmp/small_files_demo/")

# Итог: 200 файлов по ~250 КБ в папке /tmp/small_files_demo/dt=2026-01-15/
# Хотя данных всего ~8 МБ!

# Streaming пишет micro-batches - каждый trigger создаёт новые файлы
# query = events_stream \
#     .writeStream \
#     .trigger(processingTime="10 seconds") \
#     .partitionBy("event_date") \
#     .format("parquet") \
#     .option("path", "s3a://data/events/") \
#     .option("checkpointLocation", "s3a://checkpoints/events/") \
#     .start()

# Через 1 час: 360 micro-batches × 3 файла = 1 080 файлов
# Через 1 день: 8 640 micro-batches × 3 файла = 25 920 файлов
# Без компакции это накапливается бесконечно

Пример: 10 миллионов мелких файлов
→ 10 000 000 × 300 байт = 3 ГБ только для метаданных
→ NameNode с 16 ГБ heap может хранить ≈53 миллиона объектов
→ Один большой кластер с плохим партиционированием легко достигает предела
→ Далее: GC паузы, деградация, падение NameNode

Пример: читаем таблицу из 10 миллионов мелких файлов

Directory Listing:
  10 000 000 файлов / 1000 per request = 10 000 LIST запросов
  Стоимость: 10 × $0.05 = $0.05

Чтение файлов:
  10 000 000 GET запросов
  Стоимость: 10 000 × $0.0004 = $4.00

Итого за ОДИН запрос к таблице: ~$4.05
При 100 запросах в день: $405 в день = $12 150 в месяц
И это только стоимость API-запросов, не считая трафик и хранение!

При тех же 1 ТБ данных, но в 1000 файлах по 1 ГБ:
  1000 GET запросов = $0.0004 - в 10 000 раз дешевле!

1 000 000 путей × ~200 байт (String объект в JVM) = 200 МБ только для путей
+ metadata объекты FileStatus: ещё ~500 байт на файл
= 700 МБ в heap Driver только для метаданных файлов

При 10 миллионах файлов: ~7 ГБ
Типичный heap Driver: 4-8 ГБ → OutOfMemoryError

INFO FileSourceScanExec: Listing leaf files and directories...
(зависает на этой строке на несколько минут)
ERROR SparkContext: Error initializing SparkContext
java.lang.OutOfMemoryError: GC overhead limit exceeded

# Программная диагностика: смотреть на число partitions при чтении
from pyspark.sql import functions as F

df = spark.read.parquet("/path/to/table/")
num_partitions = df.rdd.getNumPartitions()
print(f"Число Spark-партиций (≈ число файлов): {num_partitions}")

# Если данных 500 МБ, а партиций 50 000 - каждая по 10 КБ - проблема
approx_size_mb = 500
avg_partition_kb = (approx_size_mb * 1024) / num_partitions
print(f"Средний размер Spark-партиции: {avg_partition_kb:.1f} КБ")

if avg_partition_kb < 1024:  # меньше 1 МБ
    print("ПРОБЛЕМА: слишком мелкие партиции при чтении")

df = spark.read.parquet("/path/to/table/")
query = df.filter(F.col("dt") == "2026-01-15").select("amount", "country")
query.explain(mode="formatted")

# Ищем в FileScan:
# Location: InMemoryFileIndex(50000 paths)[...]   ← 50 000 файлов!
# numFiles: 50000
# sizeInBytes: 500.0 MiB
#
# Соотношение: 500 МБ / 50 000 файлов = 10 КБ / файл → проблема

import math

def write_optimally(df, path, target_file_mb=256, partition_col=None):
    """
    Записать DataFrame с контролируемым числом выходных файлов.
    target_file_mb: целевой размер файла (128-512 МБ - золотой стандарт).
    """
    # Оцениваем размер Parquet через статистику плана (без full scan)
    plan = df._jdf.queryExecution().optimizedPlan()
    size_bytes = plan.stats().sizeInBytes()

    # Parquet ≈ 30-50% от размера в памяти (эффект сжатия + колоночное хранение)
    parquet_mb = float(size_bytes) / (1024 * 1024) * 0.4
    num_files = max(1, math.ceil(parquet_mb / target_file_mb))

    print(f"Оценка Parquet размера: ~{parquet_mb:.0f} МБ → {num_files} файлов")

    writer = df.coalesce(num_files).write.mode("overwrite")
    if partition_col:
        writer.partitionBy(partition_col)
    writer.parquet(path)

# Применение
events_df = spark.read.parquet("s3a://raw/events/dt=2026-01-15/")
write_optimally(events_df, "s3a://processed/events/dt=2026-01-15/", target_file_mb=256)

# Правило: groupBy создаёт 200 partition → 200 файлов на partition-папку
result = events_df \
    .groupBy("event_date", "country") \
    .agg(F.sum("amount").alias("total"), F.count("*").alias("cnt"))

# Решение: repartition по partition-ключу перед записью
result \
    .repartition(2, "event_date", "country") \
    .write \
    .mode("overwrite") \
    .partitionBy("event_date", "country") \
    .parquet("s3a://output/daily_country_stats/")
# Результат: ровно 2 файла в каждой папке event_date=X/country=Y/

# Spark SQL-хинты для управления партициями
spark.sql("""
    SELECT /*+ COALESCE(4) */ event_date, sum(amount) AS total
    FROM events
    WHERE event_date = '2026-01-15'
    GROUP BY event_date
""").write.parquet("s3a://output/")

spark.conf.set("spark.sql.adaptive.enabled", "true")
spark.conf.set("spark.sql.adaptive.coalescePartitions.enabled", "true")

# Целевой размер партиции после coalesce
spark.conf.set("spark.sql.adaptive.advisoryPartitionSizeInBytes", "268435456")  # 256 МБ
# AQE будет стараться создавать партиции размером ~256 МБ

# Минимальное число партиций (не схлопывать в 1)
spark.conf.set("spark.sql.adaptive.coalescePartitions.minPartitionNum", "1")

# Демонстрация без AQE:
spark.conf.set("spark.sql.adaptive.enabled", "false")
result_no_aqe = small_df.groupBy("category").agg(F.sum("amount"))
result_no_aqe.write.parquet("/tmp/without_aqe/")
# → 200 файлов по ~50 КБ (дефолт 200 shuffle partitions для маленького датасета)

# Демонстрация с AQE:
spark.conf.set("spark.sql.adaptive.enabled", "true")
result_with_aqe = small_df.groupBy("category").agg(F.sum("amount"))
result_with_aqe.write.parquet("/tmp/with_aqe/")
# AQE видит что данных мало → схлопывает 200 партиций в 1-3 файла оптимального размера

# По умолчанию: каждый файл = минимум 1 Task
# С maxPartitionBytes Spark может объединять файлы в один сплит

spark.conf.set("spark.sql.files.maxPartitionBytes", str(128 * 1024 * 1024))  # 128 МБ
# Если файлы по 10 МБ → Spark объединит ~13 файлов в один сплит → один Task

spark.conf.set("spark.sql.files.openCostInBytes", str(4 * 1024 * 1024))  # 4 МБ
# openCostInBytes - "цена" открытия файла в байтах данных.
# При 4 МБ: Spark считает открытие файла = 4 МБ накладных расходов.
# Эффект: файлы меньше 4 МБ объединяются агрессивнее - они "дорогие" для открытия.

# Практический эффект:
# До: 10 000 файлов по 10 МБ → 10 000 Tasks
# После: 10 000 × 10 МБ / 128 МБ ≈ 780 Tasks (в 12.8 раз меньше!)

def compact_partition(table_path, partition_value, target_files=4):
    """Компакция конкретной партиции: читаем мелкие, пишем крупные."""
    partition_path = f"{table_path}/event_date={partition_value}/"

    df = spark.read.parquet(partition_path)
    actual_files = df.rdd.getNumPartitions()

    if actual_files <= target_files * 1.5:
        print(f"Партиция {partition_value}: {actual_files} файлов - компакция не нужна")
        return

    print(f"Компакция: {partition_value}: {actual_files} → {target_files} файлов")

    tmp_path = f"{partition_path}_compact_tmp/"

    df.coalesce(target_files) \
      .write \
      .mode("overwrite") \
      .parquet(tmp_path)

    # Атомарная замена файлов
    import shutil, os
    shutil.rmtree(partition_path)
    os.rename(tmp_path, partition_path)
    print(f"Партиция {partition_value}: компакция завершена")

# Компакция последних 7 дней
from datetime import date, timedelta
today = date.today()
for days_ago in range(1, 8):
    d = (today - timedelta(days=days_ago)).isoformat()
    compact_partition("s3a://data/events", d, target_files=4)

# Оптимизировать всю таблицу
spark.sql("OPTIMIZE delta.events")

# Оптимизировать только конкретную партицию (эффективнее для больших таблиц)
spark.sql("""
    OPTIMIZE delta.events
    WHERE event_date = '2026-01-15'
""")

# OPTIMIZE + ZORDER: объединить И упорядочить данные по колонке
# Ускоряет последующие запросы с фильтрами по user_id
spark.sql("""
    OPTIMIZE delta.events
    ZORDER BY (user_id, country)
""")

# Посмотреть историю операций
spark.sql("DESCRIBE HISTORY delta.events").show(5, truncate=False)

# Удалить устаревшие файлы (минимум 7 дней retention для time-travel)
spark.sql("VACUUM delta.events RETAIN 168 HOURS")

# Базовая компакция: объединить файлы меньше 64 МБ в файлы по 256 МБ
spark.sql("""
    CALL iceberg.system.rewrite_data_files(
        table => 'iceberg.events',
        options => map(
            'target-file-size-bytes', '268435456',
            'min-file-size-bytes', '67108864'
        )
    )
""")

# Компакция с фильтром по партиции
spark.sql("""
    CALL iceberg.system.rewrite_data_files(
        table => 'iceberg.events',
        where => 'event_date = ''2026-01-15''',
        options => map('target-file-size-bytes', '268435456')
    )
""")

# Компакция + сортировка (аналог ZORDER в Delta)
spark.sql("""
    CALL iceberg.system.rewrite_data_files(
        table => 'iceberg.events',
        strategy => 'sort',
        sort_order => 'user_id ASC NULLS LAST, event_date ASC',
        options => map('target-file-size-bytes', '268435456')
    )
""")

# Delta Lake: Auto Optimize при записи из стрима
events_stream \
    .writeStream \
    .trigger(processingTime="1 minute") \
    .format("delta") \
    .option("path", "s3a://data/events_delta/") \
    .option("checkpointLocation", "s3a://checkpoints/events/") \
    .option("delta.autoOptimize.optimizeWrite", "true") \
    # optimizeWrite: Delta сама решает сколько файлов создать в батче
    # вместо N файлов (= числу Tasks) - минимальное нужное число
    .option("delta.autoOptimize.autoCompact", "true") \
    # autoCompact: после каждого батча Delta проверяет нужна ли компакция
    # и запускает её в фоне если мелких файлов накопилось достаточно
    .start()

# Альтернатива для plain Parquet: foreachBatch с явным контролем файлов
def write_batch_optimally(batch_df, batch_id):
    """Записать микробатч с контролем числа файлов + периодическая компакция."""
    batch_df \
        .repartition(4, "event_date") \
        .write \
        .mode("append") \
        .partitionBy("event_date") \
        .parquet("s3a://data/events/")

    # Компакция каждые 100 батчей (≈ 1000 секунд при интервале 10с)
    if batch_id % 100 == 0:
        print(f"Batch {batch_id}: запуск плановой компакции...")
        spark.read.parquet("s3a://data/events/") \
            .coalesce(20) \
            .write \
            .mode("overwrite") \
            .partitionBy("event_date") \
            .parquet("s3a://data/events/")

events_stream \
    .writeStream \
    .trigger(processingTime="10 seconds") \
    .foreachBatch(write_batch_optimally) \
    .option("checkpointLocation", "s3a://checkpoints/events/") \
    .start()

def check_table_health(path, warn_threshold=10_000, error_threshold=100_000):
    """Проверить здоровье таблицы по числу файлов."""
    import os

    file_count = 0
    total_size = 0
    for root, dirs, files in os.walk(path):
        for f in files:
            if f.endswith((".parquet", ".orc")):
                fp = os.path.join(root, f)
                file_count += 1
                total_size += os.path.getsize(fp)

    avg_size_mb = (total_size / file_count / (1024 * 1024)) if file_count > 0 else 0

    status = "OK"
    if file_count > warn_threshold:
        status = "WARNING"
    if file_count > error_threshold:
        status = "CRITICAL"

    print(f"[{status}] Файлов: {file_count:,} | Средний размер: {avg_size_mb:.1f} МБ")
    if avg_size_mb < 32:
        print("  Рекомендуется немедленная компакция")
    elif avg_size_mb < 64:
        print("  Рекомендуется плановая компакция")

    return {"file_count": file_count, "avg_size_mb": avg_size_mb, "status": status}

# Запускать ежедневно для всех ключевых таблиц
check_table_health("s3a://data/events/")
check_table_health("s3a://data/orders/")

INFO FileSourceScanExec: Listing leaf files and directories...

# 1. Сколько файлов в таблице?
df = spark.read.parquet("/path/to/table/")
print(f"Spark-партиций (≈ файлов): {df.rdd.getNumPartitions()}")

# 2. Проверить explain-план
df.filter(F.col("dt") == "2026-01-15").explain(mode="formatted")
# Ищем: Location: InMemoryFileIndex(N paths) где N > 10 000

# 3. Сколько Tasks в Spark UI?
# Stages → конкретный Stage → смотрим Total Tasks
# Если > 10 000 Tasks для < 100 ГБ данных → проблема

# 4. Средняя длительность Task?
# В Spark UI Summary Metrics: если Median Duration < 100 мс → overhead доминирует

# 5. Применить быстрое лечение:
spark.conf.set("spark.sql.files.maxPartitionBytes", str(128 * 1024 * 1024))
spark.conf.set("spark.sql.adaptive.enabled", "true")
spark.conf.set("spark.sql.adaptive.coalescePartitions.enabled", "true")
# Это помогает при ЧТЕНИИ уже существующих мелких файлов
# Для долгосрочного решения нужна компакция

api_logs_stream \
    .writeStream \
    .trigger(processingTime="30 seconds") \
    .partitionBy("request_date", "endpoint_category") \
    .format("parquet") \
    .option("path", "s3a://data/api_logs/") \
    .start()

# endpoint_category: 500 уникальных значений
# request_date: обновляется ежедневно
# Объём данных: 1 ГБ/час