Rename Problem: почему S3 не POSIX и как это ломает Spark

Физическая реальность S3:
┌─────────────────────────────────────────────────────────────┐
│  Бакет: my-bucket                                           │
│  Тип:   Flat key-value store                                │
│                                                             │
│  Ключ → Значение (blob bytes):                              │
│  "data/year=2024/month=06/part-00000.parquet" → [bytes...]  │
│  "data/year=2024/month=06/part-00001.parquet" → [bytes...]  │
│  "data/year=2024/month=07/part-00000.parquet" → [bytes...]  │
│  "README.md"                                  → [bytes...]  │
└─────────────────────────────────────────────────────────────┘

Что видит клиент (иллюзия директорий):
data/
  year=2024/
    month=06/
      part-00000.parquet
      part-00001.parquet
    month=07/
      part-00000.parquet
README.md

Rename директории с 10 000 файлами по 100 MB:

ListObjectsV2:   10 000 / 1000 = 10 запросов × 50ms = 500ms
CopyObject:      10 000 × 100ms = 16 минут (при параллельности 1)
                 10 000 × 100ms / 100 потоков = ~10 секунд
                 + пропускная способность: 1 TB данных через S3
DeleteObjects:   10 000 / 1000 = 10 запросов × 20ms = 200ms

Итого: от 10 секунд до 16+ минут в зависимости от параллельности
Сравните с rename в HDFS: ~50ms (изменение метаданных NameNode)

v1: attempt_dir → task_dir → output/  (2 rename)
v2: attempt_dir → output/             (1 rename)

# Выбор алгоритма в spark-submit или SparkConf:
spark = SparkSession.builder \
    .config("spark.hadoop.mapreduce.fileoutputcommitter.algorithm.version", "2") \
    .getOrCreate()

# ❌ v1 на S3 = катастрофически медленно (двойное копирование)
# ❌ v2 на S3 = быстрее, но data corruption при падении

# Ни один из вариантов не является правильным для production S3!

Симптомы в Spark UI:
- Все 500 тасков: SUCCEEDED (100%)
- Stage status: "Commit in progress"
- Driver logs: "Committing output of task attempt_0..."
- Прошло 2 часа, статус не меняется
- Heartbeat timeout → драйвер помечается мёртвым
- Джоба помечается как FAILED

# Конфигурация Magic Committer:
spark = SparkSession.builder \
    .config("spark.hadoop.fs.s3a.committer.name", "magic") \
    .config("spark.hadoop.fs.s3a.bucket.all.committer.magic.enabled", "true") \
    .config("spark.sql.sources.commitProtocolClass",
            "org.apache.spark.internal.io.cloud.PathOutputCommitProtocol") \
    .config("spark.sql.parquet.output.committer.class",
            "org.apache.spark.internal.io.cloud.BindingParquetOutputCommitter") \
    .getOrCreate()

{
  "Rules": [{
    "ID": "abort-incomplete-multipart",
    "Status": "Enabled",
    "Filter": {"Prefix": ""},
    "AbortIncompleteMultipartUpload": {"DaysAfterInitiation": 7}
  }]
}

# Конфигурация Directory Staging Committer:
spark = SparkSession.builder \
    .config("spark.hadoop.fs.s3a.committer.name", "directory") \
    .config("spark.hadoop.fs.s3a.committer.staging.tmp.path", "/tmp/spark-staging") \
    .config("spark.sql.sources.commitProtocolClass",
            "org.apache.spark.internal.io.cloud.PathOutputCommitProtocol") \
    .getOrCreate()

# Partitioned Staging Committer (для партиционированных таблиц):
spark = SparkSession.builder \
    .config("spark.hadoop.fs.s3a.committer.name", "partitioned") \
    .config("spark.hadoop.fs.s3a.committer.staging.conflict-mode", "replace") \
    .getOrCreate()
# conflict-mode: fail | append | replace

// _delta_log/0000000000000000042.json
{
  "commitInfo": {"timestamp": 1717660800000, "operation": "WRITE"},
  "add": {
    "path": "year=2024/month=06/part-abc123.parquet",
    "size": 104857600,
    "stats": {"numRecords": 1000000}
  },
  "add": {
    "path": "year=2024/month=06/part-def456.parquet",
    "size": 98765432,
    "stats": {"numRecords": 950000}
  }
}

# Запись в Delta Lake:
df.write \
    .format("delta") \
    .mode("append") \                       # или "overwrite"
    .partitionBy("year", "month") \
    .save("s3a://my-bucket/delta/orders")

# Никаких _temporary/ директорий, никаких rename!
# Файлы идут сразу в финальный путь с UUID-именами
# Коммит = запись одного JSON в _delta_log/

s3://bucket/table/
├── metadata/
│   ├── v1.metadata.json      # Snapshot 1
│   ├── v2.metadata.json      # Snapshot 2
│   ├── snap-123.avro         # Manifest list
│   └── manifest-456.avro     # Manifest file
└── data/
    ├── part-abc.parquet
    └── part-def.parquet

# Запись в Iceberg:
df.write \
    .format("iceberg") \
    .mode("append") \
    .save("catalog.schema.orders")

# Или через SQL:
spark.sql("""
    INSERT INTO catalog.schema.orders
    SELECT * FROM staging_orders
""")

# Исторический workaround (больше не нужен для AWS S3):
# s3guard - DynamoDB как consistent metadata store
spark.conf.set("fs.s3a.metadatastore.impl",
               "org.apache.hadoop.fs.s3a.s3guard.DynamoDBMetadataStore")
# После декабря 2020: AWS официально убрала s3guard как deprecated

# ❌ Проблема: все файлы в одном префиксе
df.write.parquet("s3a://bucket/events/year=2024/month=06/")
# Все PUT идут на один S3 partition key → throttling

# ✅ Решение 1: рандомизировать структуру префиксов
# Добавить хеш-бакет как первый уровень
df.withColumn("hash_bucket",
    F.pmod(F.hash(F.col("user_id")), F.lit(16)).cast("string")
).write \
    .partitionBy("hash_bucket", "year", "month") \
    .parquet("s3a://bucket/events/")
# Теперь: events/hash_bucket=0/year=2024/month=06/
#          events/hash_bucket=1/year=2024/month=06/
# 16 разных префиксов → нагрузка распределена

# ✅ Решение 2: Iceberg/Delta автоматически используют UUID-файлы
# UUID: a1b2c3d4-e5f6-...
# Разные UUID начинаются с разных символов → автоматически разные S3 partitions

from pyspark.sql import SparkSession

def create_spark_for_s3(
    app_name: str,
    s3_endpoint: str = None,       # None для AWS, URL для MinIO
    committer: str = "magic"       # magic | directory | partitioned
) -> SparkSession:
    """
    Создать SparkSession с оптимальной конфигурацией для S3.
    """
    builder = SparkSession.builder.appName(app_name)

    # --- Базовая аутентификация и endpoint ---
    if s3_endpoint:
        # MinIO или другой S3-совместимый сервис
        builder = builder \
            .config("spark.hadoop.fs.s3a.endpoint", s3_endpoint) \
            .config("spark.hadoop.fs.s3a.path.style.access", "true")

    # --- S3A Committer (вместо FileOutputCommitter) ---
    builder = builder \
        .config("spark.hadoop.fs.s3a.committer.name", committer) \
        .config("spark.sql.sources.commitProtocolClass",
                "org.apache.spark.internal.io.cloud.PathOutputCommitProtocol") \
        .config("spark.sql.parquet.output.committer.class",
                "org.apache.spark.internal.io.cloud.BindingParquetOutputCommitter")

    if committer == "magic":
        builder = builder \
            .config("spark.hadoop.fs.s3a.bucket.all.committer.magic.enabled", "true")

    if committer in ("directory", "partitioned"):
        builder = builder \
            .config("spark.hadoop.fs.s3a.committer.staging.tmp.path",
                    "/tmp/spark-staging") \
            .config("spark.hadoop.fs.s3a.committer.staging.conflict-mode",
                    "replace")  # fail | append | replace

    # --- Производительность I/O ---
    builder = builder \
        .config("spark.hadoop.fs.s3a.connection.maximum", "200") \
        .config("spark.hadoop.fs.s3a.threads.max", "64") \
        .config("spark.hadoop.fs.s3a.multipart.size", "134217728") \       # 128 MB
        .config("spark.hadoop.fs.s3a.fast.upload", "true") \
        .config("spark.hadoop.fs.s3a.fast.upload.buffer", "array") \
        .config("spark.hadoop.fs.s3a.block.size", "134217728") \           # 128 MB
        .config("spark.hadoop.fs.s3a.readahead.range", "2097152")          # 2 MB

    # --- Retry и throttling ---
    builder = builder \
        .config("spark.hadoop.fs.s3a.retry.limit", "7") \
        .config("spark.hadoop.fs.s3a.retry.interval", "500ms") \
        .config("spark.hadoop.fs.s3a.attempts.maximum", "20") \
        .config("spark.hadoop.fs.s3a.connection.timeout", "200000") \
        .config("spark.hadoop.fs.s3a.socket.recv.buffer", "65536")

    # --- Отключить Legacy FileOutputCommitter ---
    builder = builder \
        .config("spark.hadoop.mapreduce.fileoutputcommitter.algorithm.version", "2") \
        .config("spark.hadoop.mapreduce.fileoutputcommitter.cleanup-failures.ignored",
                "true")

    return builder.getOrCreate()


# Для Lakehouse (Delta/Iceberg) - ещё проще, коммитер не нужен:
def create_spark_for_iceberg_s3(app_name: str) -> SparkSession:
    return SparkSession.builder \
        .appName(app_name) \
        .config("spark.sql.extensions",
                "org.apache.iceberg.spark.extensions.IcebergSparkSessionExtensions") \
        .config("spark.sql.catalog.spark_catalog",
                "org.apache.iceberg.spark.SparkSessionCatalog") \
        .config("spark.sql.catalog.spark_catalog.type", "hive") \
        .config("spark.sql.catalog.local", "org.apache.iceberg.spark.SparkCatalog") \
        .config("spark.sql.catalog.local.type", "hadoop") \
        .config("spark.sql.catalog.local.warehouse", "s3a://my-bucket/warehouse") \
        .getOrCreate()

# ❌ Классическая ошибка новичков:
df.coalesce(1) \
  .write \
  .mode("overwrite") \
  .csv("s3a://bucket/output/result.csv")

# Что происходит:
# 1. Весь датафрейм собирается на один экзекутор (OOM risk!)
# 2. Записывается один файл в _temporary/
# 3. Rename: CopyObject (весь файл!) → boto3 таймаут при файлах > 5GB
# 4. Пользователь видит "result.csv/" директорию, а не файл

# ❌ Использование Python OS API на S3 путях:
import os
os.rename("s3a://bucket/temp/output", "s3a://bucket/final/output")
# → TypeError: не работает с s3:// протоколом
# Или хуже: использование boto3 напрямую без понимания, что это copy+delete

import boto3
s3 = boto3.client("s3")
s3.copy_object(  # "Переименование" папки - нужно делать для каждого объекта!
    CopySource={"Bucket": "my-bucket", "Key": "temp/output/file.parquet"},
    Bucket="my-bucket",
    Key="final/output/file.parquet"
)
s3.delete_object(Bucket="my-bucket", Key="temp/output/file.parquet")
# ↑ Нужно делать для каждого файла в папке - легко ошибиться

# ❌ Опасный overwrite без транзакционного лога:
df.write \
    .mode("overwrite") \     # Сначала удаляет все файлы, потом пишет новые
    .parquet("s3a://bucket/data/orders/")

# Проблема: между удалением старых файлов и появлением новых
# читатели видят пустую таблицу!
# При падении пайплайна → потеря всех данных

# ✅ Безопасный overwrite с Delta Lake:
df.write \
    .format("delta") \
    .mode("overwrite") \     # Транзакционный overwrite: старое видно до коммита
    .save("s3a://bucket/delta/orders/")
# Атомарно: читатели видят старые данные до завершения коммита,
# после коммита - новые данные, никакого окна "пустой таблицы"

# ❌ Запись тысяч мелких файлов:
streaming_df.writeStream \
    .trigger(processingTime="1 second") \  # каждую секунду новые файлы
    .format("parquet") \
    .option("path", "s3a://bucket/events/") \
    .start()

# Результат через сутки:
# 86400 файлов × средний размер 1MB = 86GB данных в 86400 файлах
# Проблема: каждый файл = отдельный HTTP GET запрос при чтении
# Spark читает 86400 файлов → 86400 × 50ms = 72 минуты только на открытие файлов!

# ✅ Решение: компакция через Delta/Iceberg
spark.sql("OPTIMIZE delta.`s3a://bucket/events`")
# Объединяет мелкие файлы в крупные (target: 1 GB)

# Исходный код (legacy):
from pyspark.sql import SparkSession, functions as F

spark = SparkSession.builder \
    .appName("events-etl-legacy") \
    .config("spark.hadoop.mapreduce.fileoutputcommitter.algorithm.version", "2") \
    .getOrCreate()

def process_batch_legacy(batch_df, batch_id):
    """Исходная обработка - прямая запись Parquet."""

    enriched = batch_df \
        .withColumn("event_date", F.to_date("event_timestamp")) \
        .withColumn("event_hour", F.hour("event_timestamp"))

    # ❌ Проблемы:
    # 1. mode("overwrite") удаляет partitions до записи новых
    # 2. FileOutputCommitter v2 → неатомарный commit
    # 3. При падении → data loss или corruption
    enriched.write \
        .mode("overwrite") \
        .partitionBy("event_date", "event_hour") \
        .parquet("s3a://data-lake/raw/events/")

# Логи показывают:
# 2024-06-01 03:42:15 INFO Committing output of task attempt_0_000_m_000000_0
# 2024-06-01 03:42:15 INFO Committing output of task attempt_0_000_m_000001_0
# ... (повторяется 500 раз)
# 2024-06-01 05:18:33 ERROR Heartbeat timeout for driver
# Job FAILED after 96 minutes

# Анализ: 500 тасков × 2 файла × (CopyObject 100ms + DeleteObject 20ms) = ~2 минуты
# Но с сетевой нестабильностью и throttling → 96 минут

from pyspark.sql import SparkSession, functions as F
from pyspark.sql.types import StructType, StructField, StringType, LongType, TimestampType


def create_spark_session() -> SparkSession:
    return SparkSession.builder \
        .appName("events-etl-iceberg") \
        .config("spark.sql.extensions",
                "org.apache.iceberg.spark.extensions.IcebergSparkSessionExtensions") \
        .config("spark.sql.catalog.glue",
                "org.apache.iceberg.spark.SparkCatalog") \
        .config("spark.sql.catalog.glue.catalog-impl",
                "org.apache.iceberg.aws.glue.GlueCatalog") \
        .config("spark.sql.catalog.glue.warehouse",
                "s3a://data-lake/warehouse/") \
        .config("spark.hadoop.fs.s3a.connection.maximum", "200") \
        .config("spark.hadoop.fs.s3a.multipart.size", "134217728") \
        .getOrCreate()


def ensure_iceberg_table(spark: SparkSession) -> None:
    """DDL для Iceberg таблицы - idempotent."""
    spark.sql("""
        CREATE TABLE IF NOT EXISTS glue.raw.events (
            event_id        STRING  NOT NULL,
            user_id         BIGINT,
            event_type      STRING,
            payload         STRING,
            event_timestamp TIMESTAMP NOT NULL,
            event_date      DATE      NOT NULL,
            event_hour      INT       NOT NULL
        )
        USING iceberg
        PARTITIONED BY (event_date, event_hour)
        TBLPROPERTIES (
            'write.format.default'           = 'parquet',
            'write.parquet.compression-codec'= 'zstd',
            'write.target-file-size-bytes'   = '134217728',
            'format-version'                 = '2'
        )
    """)


def process_batch_iceberg(spark: SparkSession, batch_df, batch_id: int) -> dict:
    """
    Обработка микробатча с записью в Iceberg.

    Что изменилось:
    - Нет _temporary/ директорий
    - Нет rename операций
    - Атомарный commit через snapshot metadata
    - При падении: нет data corruption, orphan files → vacuum удалит
    """
    enriched = batch_df \
        .withColumn("event_date", F.to_date("event_timestamp")) \
        .withColumn("event_hour", F.hour("event_timestamp")) \
        .dropDuplicates(["event_id"])  # идемпотентность при retry

    enriched.createOrReplaceTempView("_batch_events")

    # Iceberg MERGE для exactly-once семантики:
    spark.sql("""
        MERGE INTO glue.raw.events AS target
        USING _batch_events AS source
        ON target.event_id = source.event_id
        WHEN NOT MATCHED THEN INSERT *
    """)

    # Метрики коммита:
    stats = spark.sql(f"""
        SELECT operation, summary
        FROM glue.raw.events.history
        ORDER BY made_current_at DESC
        LIMIT 1
    """).first()

    return {
        "batch_id": batch_id,
        "operation": stats["operation"],
    }


def run_streaming_pipeline() -> None:
    spark = create_spark_session()
    ensure_iceberg_table(spark)

    kafka_df = spark.readStream \
        .format("kafka") \
        .option("kafka.bootstrap.servers", "kafka:9092") \
        .option("subscribe", "events") \
        .load()

    parsed_df = kafka_df.selectExpr(
        "CAST(value AS STRING) AS raw_json",
        "timestamp AS kafka_timestamp"
    ).select(
        F.get_json_object("raw_json", "$.event_id").alias("event_id"),
        F.get_json_object("raw_json", "$.user_id").cast("long").alias("user_id"),
        F.get_json_object("raw_json", "$.event_type").alias("event_type"),
        F.get_json_object("raw_json", "$.payload").alias("payload"),
        F.get_json_object("raw_json", "$.timestamp").cast("timestamp")
            .alias("event_timestamp"),
    )

    query = parsed_df.writeStream \
        .foreachBatch(
            lambda df, bid: process_batch_iceberg(spark, df, bid)
        ) \
        .trigger(processingTime="5 minutes") \
        .option("checkpointLocation", "s3a://data-lake/checkpoints/events/") \
        .start()

    query.awaitTermination()

Legacy FileOutputCommitter v2 + Parquet:
├── Запись данных (500 параллельных тасков): 8 минут
├── Job Commit (rename 1000 файлов × CopyObject): 96 минут (!)
├── Total pipeline: 104 минуты
├── Успешных запусков из 100: 73 (27% падений из-за timeout)
└── Инциденты data corruption за месяц: 3

Iceberg + Spark:
├── Запись данных (500 параллельных тасков): 8 минут
├── Job Commit (snapshot metadata update): 2 секунды
├── Total pipeline: 8 минут 2 секунды
├── Успешных запусков из 100: 100 (0% падений)
└── Инциденты data corruption за месяц: 0