HDFS vs S3: когда что выбирать - latency, consistency, стоимость

# Признак S3 throttling в логах Spark:
# WARN  RetryThrottlingHandler: Throttling exception from S3. Sleeping for...
# WARN  AmazonS3Exception: Status Code: 503, AWS Service: Amazon S3,
#   AWS Request ID: ..., Extended Request ID: ...,
#   Error Message: Slow Down, Error type: Service

# Решение: S3 автоматически увеличивает лимит при хорошей партиционной схеме
# Ключ: распределять объекты по разным prefix'ам
# НЕ делать: s3://bucket/data/events/ (все объекты в одном prefix)
# ДЕЛАТЬ:     s3://bucket/data/events/year=2024/month=01/day=15/ (разные prefix)

Spark Job 1: записывает 1000 файлов в s3://bucket/output/
Spark Job 1: завершается успешно, пишет _SUCCESS файл

Spark Job 2: немедленно запускается, делает LIST s3://bucket/output/
Spark Job 2: видит только 700 из 1000 файлов (еще 300 не "проявились")
Spark Job 2: обрабатывает неполные данные → тихая ошибка в данных

# Последствия неатомарного rename в S3 для Spark:
# Стандартный FileOutputCommitter (v1) работает так:
# 1. Spark Tasks записывают в s3://bucket/_temporary/attempt_xxx/
# 2. Driver делает rename/_temporary/ → s3://bucket/output/

# В S3 шаг 2 = тысячи PUT + тысячи DELETE операций
# Если кластер упадёт в середине этой операции:
# - Часть файлов в новом пути (s3://bucket/output/)
# - Часть файлов в старом пути (s3://bucket/_temporary/)
# - Данные неконсистентны! Нет транзакционности!

# Решение: S3A Magic Committer или Staging Committer
# (подробнее в модуле 05 о S3 хранилищах)
from pyspark.sql import SparkSession

spark = SparkSession.builder \
    # Magic Committer: файлы сразу пишутся по финальному пути
    # Нет staging директории = нет rename-проблемы!
    .config("spark.hadoop.fs.s3a.committer.name", "magic") \
    .config("spark.hadoop.fs.s3a.committer.magic.enabled", "true") \
    .config("spark.sql.sources.commitProtocolClass",
            "org.apache.spark.internal.io.cloud.PathOutputCommitProtocol") \
    .getOrCreate()

spark = SparkSession.builder \
    # S3: никогда нет NODE_LOCAL → ожидание бессмысленно
    .config("spark.locality.wait", "0") \
    .config("spark.locality.wait.node", "0") \
    .config("spark.locality.wait.process", "0") \
    # HDFS: ждём NODE_LOCAL слот
    # .config("spark.locality.wait.node", "10s")
    .getOrCreate()

# Iceberg с S3: никаких LIST запросов при чтении!
spark = SparkSession.builder \
    .config("spark.sql.extensions",
            "org.apache.iceberg.spark.extensions.IcebergSparkSessionExtensions") \
    .config("spark.sql.catalog.iceberg",
            "org.apache.iceberg.spark.SparkCatalog") \
    .config("spark.sql.catalog.iceberg.type", "hive") \
    .config("spark.sql.catalog.iceberg.warehouse", "s3a://warehouse/") \
    # S3FileIO: обходит Hadoop S3A, работает напрямую через AWS SDK
    .config("spark.sql.catalog.iceberg.io-impl",
            "org.apache.iceberg.aws.s3.S3FileIO") \
    .config("spark.sql.catalog.iceberg.s3.endpoint", "http://minio:9000") \
    .getOrCreate()

# Spark читает metadata.json → manifest list → manifest files
# Ни одного LIST запроса к S3!
df = spark.table("iceberg.warehouse.events")
df.filter("event_date = '2024-01-15'").explain()
# Physical Plan: BatchScan iceberg[event_date=2024-01-15] (точный список файлов)
# Нет: ListObjectsV2 → FilteredScan → ...

1 PB × $0.023/GB/месяц = $23,552/месяц = $282,624/год

1 PB × $0.004/GB/месяц = $4,096/месяц = $49,152/год

# Расчёт стоимости S3 API операций для типичного Spark ETL

# Допущения:
# - 100 Spark Job в день
# - Каждый Job читает 10,000 файлов
# - 5 LIST запросов на директорию (paginated)
# - replication: нет (один экземпляр данных в S3)

daily_jobs = 100
files_per_job = 10_000
list_requests_per_job = files_per_job // 1000 * 5  # S3 returns 1000 obj/request

# Стоимость операций (AWS S3, us-east-1, 2024):
s3_get_price_per_1000 = 0.0004   # $0.0004 за 1000 GET запросов
s3_list_price_per_1000 = 0.005   # $0.005 за 1000 LIST запросов (в 12.5 раз дороже!)

daily_get_requests = daily_jobs * files_per_job
daily_list_requests = daily_jobs * list_requests_per_job

daily_api_cost = (
    daily_get_requests / 1000 * s3_get_price_per_1000 +
    daily_list_requests / 1000 * s3_list_price_per_1000
)

print(f"GET запросов в день:  {daily_get_requests:,}")
print(f"LIST запросов в день: {daily_list_requests:,}")
print(f"Стоимость API в день: ${daily_api_cost:.2f}")
print(f"Стоимость API в год:  ${daily_api_cost * 365:.2f}")

# Пример вывода:
# GET запросов в день:  1,000,000
# LIST запросов в день: 5,000
# Стоимость API в день: $0.43
# Стоимость API в год:  $156.95

# Это разумно. НО при Small Files (миллион мелких файлов):
daily_jobs_bad = 100
files_per_job_bad = 1_000_000   # Small Files!
list_requests_bad = files_per_job_bad // 1000 * 5

daily_get_bad = daily_jobs_bad * files_per_job_bad
daily_list_bad = daily_jobs_bad * list_requests_bad

daily_api_cost_bad = (
    daily_get_bad / 1000 * s3_get_price_per_1000 +
    daily_list_bad / 1000 * s3_list_price_per_1000
)

print(f"\nПри Small Files (1M файлов на Job):")
print(f"Стоимость API в день: ${daily_api_cost_bad:.2f}")
print(f"Стоимость API в год:  ${daily_api_cost_bad * 365:,.0f}")

# При Small Files:
# Стоимость API в день: $42.50
# Стоимость API в год:  $15,512  ← просто за мелкие файлы!

ВЫБИРАЙТЕ HDFS если:
□ Уже есть зрелый Hadoop-кластер с опытной командой Ops
□ Данные не могут покидать корпоративный периметр (банки, госсектор)
□ Нагрузка - тяжёлый Shuffle или итеративные ML-алгоритмы
□ Данные используются 24/7 без простоев (нет смысла в Scale-to-Zero)
□ Требуется POSIX-семантика (rename, hardlinks, директории)
□ Compliance/суверенитет данных: полный физический контроль
□ Объём данных: 1-100 PB, стабильный рост

ВЫБИРАЙТЕ S3/Object Storage если:
□ Новый проект без legacy Hadoop-инфраструктуры
□ Нужен Scale-to-Zero (вычисления нужны только N часов в день)
□ Данные неизменяемы (write-once, read-many)
□ Используется Iceberg/Delta Lake (нивелируют недостатки S3)
□ Важна скорость развёртывания (облачный managed service)
□ Объём хранения непредсказуем или растёт быстро
□ Cold данные занимают > 50% общего объёма (Glacier tiering)
□ Команда небольшая, нет ресурсов на Hadoop Ops

# Параметры:
monthly_volume_tb = 5_000      # 5 PB/месяц
retention_years = 3
access_frequency = "quarterly"  # редко
data_sovereignty = "Russia"     # no-cloud

# Подсчёт объёма:
total_tb = monthly_volume_tb * 12 * retention_years
print(f"Итого данных: {total_tb} TB = {total_tb/1000:.0f} PB")
# Итого данных: 180000 TB = 180 PB

# HDFS:
hdfs_overhead = 3.0  # 3x репликация
hdfs_raw_tb = total_tb * hdfs_overhead
print(f"HDFS raw storage: {hdfs_raw_tb/1000:.0f} PB (с 3x репл.)")

# MinIO с EC RS-10-4:
ec_overhead = 1.4   # 40% overhead
minio_raw_tb = total_tb * ec_overhead
print(f"MinIO EC RS-10-4: {minio_raw_tb/1000:.0f} PB")
# Экономия: 180 PB vs 252 PB = 72 PB экономии!

spark = SparkSession.builder \
    .appName("telecom-compliance-query") \
    # MinIO endpoint
    .config("spark.hadoop.fs.s3a.endpoint", "http://minio.internal:9000") \
    .config("spark.hadoop.fs.s3a.path.style.access", "true") \
    # Magic Committer для MinIO
    .config("spark.hadoop.fs.s3a.committer.name", "magic") \
    .config("spark.hadoop.fs.s3a.committer.magic.enabled", "true") \
    # Iceberg каталог
    .config("spark.sql.extensions",
            "org.apache.iceberg.spark.extensions.IcebergSparkSessionExtensions") \
    # Для compliance запросов: большие executors, долгие таймауты
    .config("spark.executor.memory", "32g")  \
    .config("spark.sql.files.maxPartitionBytes", str(512 * 1024 * 1024)) \
    # S3 locality нет → не ждём
    .config("spark.locality.wait", "0") \
    .getOrCreate()

# Параметры:
active_data_tb = 50
batch_interval_min = 15        # каждые 15 минут новый расчёт
sla_latency_min = 10           # результат через 10 минут после события

# Паттерн доступа:
access_pattern = "hot"         # данные читаются каждые 15 минут
streaming = True               # Structured Streaming
shuffle_heavy = True           # LTV = сложные join + aggregation

# Анализ требований:
# 1. Streaming каждые 15 минут → файлы пишутся часто → Small Files!
# 2. LTV = много join'ов → тяжёлый Shuffle
# 3. Latency SLA 10 минут → критично время планирования и чтения
# 4. 50 TB активных данных → можно хранить всё на HDFS

spark = SparkSession.builder \
    .appName("retail-ltvCalculation-streaming") \
    .config("spark.sql.streaming.schemaInference", "false") \
    # HDFS HA конфигурация
    .config("spark.hadoop.fs.defaultFS", "hdfs://retail-cluster") \
    # Для тяжёлого Shuffle: локальные NVMe диски
    .config("spark.local.dir", "/nvme/spark-local/") \
    # Для streaming: не ждём локальность (микробатчи короткие)
    .config("spark.locality.wait", "0") \
    # Short-Circuit для максимальной скорости чтения
    .config("spark.hadoop.dfs.client.read.shortcircuit", "true") \
    .config("spark.hadoop.dfs.domain.socket.path",
            "/var/lib/hadoop-hdfs/dn_socket") \
    # AQE для adaptive coalescing после shuffle
    .config("spark.sql.adaptive.enabled", "true") \
    .config("spark.sql.adaptive.coalescePartitions.enabled", "true") \
    .getOrCreate()

# Ночная компакция горячего слоя
def compact_hot_layer(spark, hdfs_path: str, date_str: str) -> None:
    """
    Компактирует мелкие файлы от Streaming в оптимальные Parquet-файлы.
    Запускается в 02:00 каждую ночь через Airflow.
    """
    df = spark.read.parquet(f"{hdfs_path}/date={date_str}/")
    count = df.count()

    # Целевой размер: 256 MB на файл
    # При 50 KB/строку: ~5000 строк/файл
    target_partitions = max(1, int(count / 50_000))

    df.repartition(target_partitions) \
        .write.mode("overwrite") \
        .parquet(f"{hdfs_path}_compact/date={date_str}/")

# Параметры:
active_tb = 200        # активных данных
archive_tb = 2_000     # архивных данных
no_cloud = True        # требование безопасности
nightly_sla_hours = 4  # ночной расчёт за 4 часа
batch_type = "complex" # сложные join'ы, агрегации

# Стратегия:
# - Активные 200 TB: HDFS с репликацией 3x
#   → максимальная производительность Spark SQL
#   → data locality, Short-Circuit, низкий metadata latency
# - Архив 2 PB: Ceph RGW с S3 API + EC RS-6-3
#   → 50% экономии vs HDFS replication
#   → S3-совместимый API, но физически on-prem
#   → Iceberg таблицы для прозрачного доступа

# Гибридная архитектура:
# HDFS (горячий) → Ceph S3 (архив)
# Оба on-prem, оба внутри банковского периметра

# Конфигурация Spark для банковского кейса
spark = SparkSession.builder \
    .appName("bank-regulatory-reporting") \
    # HDFS для горячих данных
    .config("spark.hadoop.fs.defaultFS", "hdfs://bank-cluster") \
    # Ceph RGW для архивных данных (через S3A)
    .config("spark.hadoop.fs.s3a.endpoint", "http://ceph-rgw.bank.internal:7480") \
    .config("spark.hadoop.fs.s3a.path.style.access", "true") \
    .config("spark.hadoop.fs.s3a.aws.credentials.provider",
            "org.apache.hadoop.fs.s3a.SimpleAWSCredentialsProvider") \
    # Iceberg с Hive Metastore (каталог над обоими хранилищами)
    .enableHiveSupport() \
    .config("spark.sql.hive.metastore.uris", "thrift://metastore:9083") \
    .config("spark.sql.extensions",
            "org.apache.iceberg.spark.extensions.IcebergSparkSessionExtensions") \
    # Производительность: максимум для ночного batch
    .config("spark.executor.instances", "50") \
    .config("spark.executor.cores", "8") \
    .config("spark.executor.memory", "24g") \
    # HDFS short-circuit для горячих данных
    .config("spark.hadoop.dfs.client.read.shortcircuit", "true") \
    .config("spark.hadoop.dfs.domain.socket.path",
            "/var/lib/hadoop-hdfs/dn_socket") \
    # Node locality для HDFS stage'ов
    .config("spark.locality.wait.node", "15s") \
    # AQE для адаптивной оптимизации
    .config("spark.sql.adaptive.enabled", "true") \
    .getOrCreate()

# Запрос, который прозрачно читает из обоих хранилищ
def run_regulatory_report(spark, report_date: str) -> None:
    """
    Регуляторный отчёт ЦБ РФ N47-P.
    Горячие данные - из HDFS, архивные - из Ceph, всё через Iceberg.
    """
    # Текущие позиции (HDFS - горячие данные)
    positions = spark.table("iceberg.bank_dw.positions") \
        .filter(f"report_date = '{report_date}'")

    # Исторические транзакции (Ceph - архив)
    transactions_12m = spark.table("iceberg.bank_archive.transactions") \
        .filter(f"transaction_date >= date_sub('{report_date}', 365)")

    # Справочники рисков (HDFS)
    risk_weights = spark.table("hive.bank_dw.risk_weights")

    # Расчёт регуляторного норматива Н1
    result = positions \
        .join(transactions_12m, "account_id", "left") \
        .join(risk_weights, "asset_class", "inner") \
        .groupBy("legal_entity", "currency") \
        .agg({
            "risk_weighted_amount": "sum",
            "capital_requirement": "sum",
        })

    result.write \
        .mode("overwrite") \
        .saveAsTable(f"iceberg.bank_reports.n1_report_{report_date.replace('-', '_')}")