Erasure Coding: Reed-Solomon коды, overhead vs репликация, hot/cold данные

# Эффект EC на Spark-задачи можно увидеть в метриках:
# При нормальной работе (все DN доступны):
# - Task Duration: ~30 сек
# - Input Bytes Read: 128 MB (один блок)
# - Bytes Read From Remote: 107 MB (5 из 6 ячеек по сети)

# При деградации (1 DN недоступен, идёт reconstruction):
# - Task Duration: ~90 сек (в 3 раза дольше!)
# - Input Bytes Read: 128 MB (тот же объём данных)
# - Bytes Read From Remote: 128 MB (все 6 ячеек по сети + parity)
# - Extra CPU Time: +15 сек (RS decode)

# 1. Проверить поддержку AVX-512 на процессоре
grep -m1 "avx512" /proc/cpuinfo
# Если вывод есть - AVX-512 поддерживается

# 2. Проверить наличие ISA-L библиотеки в Hadoop
find /usr/lib/hadoop /opt/hadoop -name "libhadoop.so*" 2>/dev/null

# 3. Проверить что ISA-L скомпилирован в libhadoop.so
nm -D /usr/lib/hadoop/lib/native/libhadoop.so | grep -i "isal\|isa_l"
# Должны быть символы: erasure_code_cal_matrix, gf_gen_rs_matrix и т.д.

# 4. Проверить через Hadoop CLI что EC использует нативный codec
hadoop checknative -a
# Ожидаемый вывод:
# native-hadoop: true /usr/lib/hadoop/lib/native/libhadoop.so.1.0.0
# isa-l: true /usr/lib/hadoop/lib/native/libhadoop.so.1.0.0
# java zlib: false (не нужно для EC)

# 5. Если ISA-L не встроен - установить и пересобрать Hadoop
# Ubuntu/Debian:
sudo apt-get install libisal-dev
# CentOS/RHEL:
sudo yum install isa-l-devel

# Пересборка Hadoop с ISA-L поддержкой:
mvn package -Pdist,native -DskipTests \
    -Disal.prefix=/usr \
    -Disal.lib=/usr/lib64 2>&1 | tail -20

spark = SparkSession.builder \
    # Путь к нативным библиотекам Hadoop (включая libhadoop.so с ISA-L)
    .config("spark.executor.extraJavaOptions",
            "-Djava.library.path=/usr/lib/hadoop/lib/native") \
    .config("spark.executorEnv.LD_LIBRARY_PATH",
            "/usr/lib/hadoop/lib/native") \
    # Принудительно использовать нативный EC codec
    .config("spark.hadoop.io.erasurecode.codec.rs.rawcoder",
            "org.apache.hadoop.io.erasurecode.rawcoder.NativeRSRawErasureCoderFactory") \
    .getOrCreate()

# 1. Просмотр доступных EC политик
hdfs ec -listPolicies

# Пример вывода:
# ErasureCodingPolicy=[Name=RS-6-3-1024k, Schema=[ECSchema=[Codec=rs, numDataUnits=6,
#   numParityUnits=3]], CellSize=1048576, State=ENABLED]
# ErasureCodingPolicy=[Name=RS-3-2-1024k, Schema=[ECSchema=[Codec=rs, numDataUnits=3,
#   numParityUnits=2]], CellSize=1048576, State=ENABLED]
# ErasureCodingPolicy=[Name=RS-10-4-1024k, Schema=[ECSchema=[Codec=rs, numDataUnits=10,
#   numParityUnits=4]], CellSize=1048576, State=ENABLED]

# 2. Просмотр текущей политики директории
hdfs ec -getPolicy -path /user/spark/data/bronze/current/
# Directory: /user/spark/data/bronze/current/
# ErasureCodingPolicy=null (нет EC, использует репликацию)

hdfs ec -getPolicy -path /user/spark/data/bronze/archive_2y/
# ErasureCodingPolicy=RS-6-3-1024k

# 3. Установка EC политики на директорию
# ВАЖНО: политика применяется к НОВЫМ файлам!
# Уже существующие файлы не конвертируются автоматически!
hdfs ec -setPolicy -path /user/spark/data/bronze/archive_2y -policy RS-6-3-1024k

# 4. Удаление EC политики (возврат к репликации для новых файлов)
hdfs ec -unsetPolicy -path /user/spark/data/bronze/archive_2y

# 5. Конвертация СУЩЕСТВУЮЩИХ реплицированных файлов в EC
# hdfs distcp с автоматическим применением EC политики целевой директории:
hdfs distcp \
  -Ddfs.replication=1 \
  hdfs://cluster/user/spark/data/bronze/2022/ \
  hdfs://cluster/user/spark/data/cold/bronze/2022/
# Целевая директория должна иметь установленную EC политику заранее!

# 6. Проверка статуса блоков конкретного файла
hdfs fsck /user/spark/data/cold/bronze/2022/part-001.parquet \
  -files -blocks -locations
# Вывод покажет EC блоки и их расположение по DataNode

# dags/hdfs_ec_migration_dag.py
# Airflow DAG: автоматический перевод данных старше N месяцев в EC

from datetime import datetime, timedelta
from airflow import DAG
from airflow.operators.python import PythonOperator
from airflow.providers.apache.hdfs.hooks.hdfs import HDFSHook
import subprocess
import logging


logger = logging.getLogger(__name__)

DEFAULT_ARGS = {
    "owner": "data-platform",
    "depends_on_past": False,
    "retries": 2,
    "retry_delay": timedelta(minutes=30),
    "email_on_failure": True,
}

# Конфигурация: какие директории переводить в EC и когда
EC_MIGRATION_CONFIG = [
    {
        "source_base": "/user/spark/data/bronze",
        "cold_base": "/user/spark/data/cold/bronze",
        "ec_policy": "RS-6-3-1024k",
        "age_months": 24,        # данные старше 24 месяцев → EC
        "description": "Bronze raw layer archival",
    },
    {
        "source_base": "/user/spark/data/silver",
        "cold_base": "/user/spark/data/cold/silver",
        "ec_policy": "RS-6-3-1024k",
        "age_months": 18,        # Silver - переводим чуть быстрее
        "description": "Silver cleaned layer archival",
    },
]


def get_partitions_older_than(hdfs_path: str, months: int) -> list[str]:
    """
    Возвращает список партиций (поддиректорий date=YYYY-MM-DD)
    старше указанного числа месяцев.
    Предполагает партиционирование по дате вида /path/date=2022-01-15/.
    """
    cutoff_date = datetime.now() - timedelta(days=months * 30)

    result = subprocess.run(
        ["hdfs", "dfs", "-ls", f"{hdfs_path}/"],
        capture_output=True, text=True
    )

    old_partitions = []
    for line in result.stdout.splitlines():
        # Парсим строку вида: drwxr-xr-x  ... /user/spark/data/bronze/date=2022-01-15
        parts = line.split()
        if len(parts) < 8:
            continue
        path = parts[-1]
        if "date=" not in path:
            continue

        date_str = path.split("date=")[-1].rstrip("/")
        try:
            partition_date = datetime.strptime(date_str[:10], "%Y-%m-%d")
            if partition_date < cutoff_date:
                old_partitions.append(path)
        except ValueError:
            continue

    logger.info(f"Найдено {len(old_partitions)} партиций старше {months} месяцев в {hdfs_path}")
    return old_partitions


def setup_cold_directory(cold_path: str, ec_policy: str) -> None:
    """
    Создаёт целевую директорию для холодных данных и устанавливает EC политику.
    EC политика должна быть установлена ДО копирования данных!
    """
    # Создаём директорию если не существует
    subprocess.run(["hdfs", "dfs", "-mkdir", "-p", cold_path], check=True)

    # Устанавливаем EC политику
    subprocess.run(
        ["hdfs", "ec", "-setPolicy", "-path", cold_path, "-policy", ec_policy],
        check=True
    )
    logger.info(f"EC политика {ec_policy} установлена на {cold_path}")


def migrate_partition_to_ec(source_path: str, cold_base: str, ec_policy: str) -> bool:
    """
    Мигрирует одну партицию из репликации в EC через hdfs distcp.

    Алгоритм:
    1. Создаём целевую директорию с EC политикой
    2. Копируем данные через distcp (distcp читает старые файлы, пишет новые с EC)
    3. Верифицируем что целевые файлы корректны (checksums)
    4. Удаляем исходные файлы
    5. Создаём symlink или обновляем Hive метастор (если используется)
    """
    partition_name = source_path.split("/")[-1]
    target_path = f"{cold_base}/{partition_name}"

    logger.info(f"Начинаем миграцию: {source_path} → {target_path}")

    # Шаг 1: настройка целевой директории
    setup_cold_directory(cold_base, ec_policy)

    # Шаг 2: копирование через distcp
    # -p: сохранять атрибуты файлов (timestamps, permissions)
    # -update: пропускать уже скопированные файлы (идемпотентность)
    # -delete: удалять файлы в целевой директории которых нет в источнике
    distcp_result = subprocess.run([
        "hadoop", "distcp",
        "-p",
        "-update",
        source_path,
        target_path,
    ], capture_output=True, text=True)

    if distcp_result.returncode != 0:
        logger.error(f"distcp завершился с ошибкой: {distcp_result.stderr}")
        return False

    # Шаг 3: верификация checksums
    verify_result = subprocess.run([
        "hdfs", "dfs", "-expunge"  # убираем из корзины
    ])

    fsck_result = subprocess.run([
        "hdfs", "fsck", target_path,
        "-files", "-blocks", "-locations"
    ], capture_output=True, text=True)

    if "Status: HEALTHY" not in fsck_result.stdout and "HEALTHY" not in fsck_result.stdout:
        logger.warning(f"fsck предупреждение для {target_path}")

    # Шаг 4: удаление исходных файлов (только после успешного копирования!)
    delete_result = subprocess.run([
        "hdfs", "dfs", "-rm", "-r", "-skipTrash", source_path
    ], capture_output=True, text=True)

    if delete_result.returncode != 0:
        logger.error(f"Ошибка удаления источника: {delete_result.stderr}")
        return False

    logger.info(f"Миграция завершена: {partition_name}")
    return True


def run_ec_migration(**context) -> None:
    """
    Основная функция DAG: сканирует директории и мигрирует старые партиции.
    """
    for config in EC_MIGRATION_CONFIG:
        logger.info(f"Обрабатываем: {config['description']}")

        old_partitions = get_partitions_older_than(
            config["source_base"],
            config["age_months"]
        )

        if not old_partitions:
            logger.info(f"Нет партиций для миграции в {config['source_base']}")
            continue

        logger.info(f"Найдено {len(old_partitions)} партиций для миграции в EC")

        success_count = 0
        error_count = 0

        for partition in old_partitions:
            try:
                success = migrate_partition_to_ec(
                    partition,
                    config["cold_base"],
                    config["ec_policy"]
                )
                if success:
                    success_count += 1
                else:
                    error_count += 1
            except Exception as e:
                logger.error(f"Ошибка при миграции {partition}: {e}")
                error_count += 1

        logger.info(
            f"Итого: {success_count} успешно, {error_count} ошибок "
            f"из {len(old_partitions)} партиций"
        )


with DAG(
    dag_id="hdfs_ec_cold_migration",
    default_args=DEFAULT_ARGS,
    description="Автоматический перевод архивных данных HDFS в Erasure Coding",
    schedule_interval="0 2 1 * *",  # Каждый первый день месяца в 2:00
    start_date=datetime(2024, 1, 1),
    catchup=False,
    tags=["hdfs", "ec", "archival"],
) as dag:

    migrate_task = PythonOperator(
        task_id="migrate_cold_partitions_to_ec",
        python_callable=run_ec_migration,
        doc_md="""
        Сканирует Bronze и Silver директории HDFS.
        Переводит партиции старше N месяцев в Erasure Coding.
        Экономия: ~50% дискового пространства на архивных данных.
        """,
    )

# Включение детальных логов чтения HDFS
spark-submit \
  --conf "spark.driver.extraJavaOptions=-Dlog4j.logger.org.apache.hadoop.hdfs=DEBUG" \
  --conf "spark.executor.extraJavaOptions=-Dlog4j.logger.org.apache.hadoop.hdfs=DEBUG" \
  your_job.py

# Лог при чтении EC-файла (DFSStripedInputStream):
DEBUG DFSStripedInputStream: opening striped input stream for file
  /user/spark/data/cold/bronze/2022/part-001.parquet
DEBUG StripedBlockUtil: file is striped with policy RS-6-3-1024k:
  numDataUnits=6, numParityUnits=3, cellSize=1048576
DEBUG StripedBlockUtil: stripe 0: block BP-123.../blk_001@DN1, blk_002@DN2...

# Лог при чтении обычного реплицированного файла (DFSInputStream):
DEBUG DFSInputStream: opening input stream for file
  /user/spark/data/bronze/current/part-001.parquet
DEBUG BlockReaderLocal: creating short-circuit reader for block blk_001
  (NODE_LOCAL)

# ec_monitoring.py
# Мониторинг состояния Erasure Coding на HDFS кластере

import requests
import json
from typing import Optional


def get_ec_stats(namenode_host: str, port: int = 9870) -> dict:
    """
    Получает статистику EC операций с NameNode через JMX.

    Ключевые метрики:
    - ECReconstructionBytesRead: байт прочитано для EC реконструкции
    - ECReconstructionWritten: байт записано при EC реконструкции
    - ECReconstructionBlocks: число блоков на реконструкции
    - ECDecodingBlocksCount: текущее число блоков в процессе декодирования
    """
    url = f"http://{namenode_host}:{port}/jmx?qry=Hadoop:service=NameNode,name=NameNodeInfo"

    try:
        response = requests.get(url, timeout=10)
        data = response.json()

        for bean in data.get("beans", []):
            if "NameNodeInfo" not in bean.get("name", ""):
                continue

            return {
                "ec_policy_stats": bean.get("ErasureCodingPolicies", ""),
                "under_replicated_blocks": bean.get("UnderReplicatedBlocks", 0),
                "pending_deletions": bean.get("PendingDeletionBlocks", 0),
            }
    except Exception as e:
        return {"error": str(e)}


def get_datanode_ec_metrics(datanode_host: str, port: int = 9864) -> dict:
    """
    Получает EC-специфичные метрики DataNode:
    - ECReconstructionBytesRead: байт прочитано для восстановления EC блоков
    - ECReconstructionWritten: байт записано в ходе EC реконструкции
    - ECDecodingMillis: время затраченное на EC декодирование
    """
    url = (f"http://{datanode_host}:{port}/jmx"
           f"?qry=Hadoop:service=DataNode,name=DataNodeActivity")

    try:
        response = requests.get(url, timeout=5)
        data = response.json()

        for bean in data.get("beans", []):
            if "DataNodeActivity" not in bean.get("name", ""):
                continue

            ec_bytes_read = bean.get("ECReconstructionBytesRead", 0)
            ec_bytes_written = bean.get("ECReconstructionWritten", 0)
            total_bytes_read = bean.get("BytesRead", 1)  # избегаем деление на 0

            ec_reconstruction_pct = (
                ec_bytes_read / total_bytes_read * 100
                if total_bytes_read > 0 else 0
            )

            return {
                "hostname": datanode_host,
                "ec_reconstruction_bytes_read": ec_bytes_read,
                "ec_reconstruction_bytes_written": ec_bytes_written,
                "ec_reconstruction_pct_of_total_reads": ec_reconstruction_pct,
                "ec_decoding_millis": bean.get("ECDecodingMillis", 0),
            }
    except Exception as e:
        return {"hostname": datanode_host, "error": str(e)}


def audit_ec_health(datanodes: list[str], threshold_pct: float = 20.0) -> None:
    """
    Аудит EC health: предупреждает если доля EC реконструкции
    превышает порог (признак частых отказов DataNode).

    При высоком EC reconstruction:
    - DataNode часто недоступны (проблемы с железом или сетью)
    - Spark-задачи тратят лишние CPU и сетевые ресурсы на декодирование
    - Нужно проверить состояние кластера
    """
    print("\n" + "=" * 60)
    print("АУДИТ ERASURE CODING МЕТРИК")
    print("=" * 60)

    problem_nodes = []

    for dn in datanodes:
        metrics = get_datanode_ec_metrics(dn)

        if "error" in metrics:
            print(f"  {dn}: ОШИБКА - {metrics['error']}")
            continue

        pct = metrics["ec_reconstruction_pct_of_total_reads"]
        status = "✅" if pct < threshold_pct else "⚠️ "

        print(f"  {status} {dn}: EC reconstruction = {pct:.1f}% "
              f"({metrics['ec_reconstruction_bytes_read'] / 1024**3:.1f} GB)")

        if pct >= threshold_pct:
            problem_nodes.append(dn)

    if problem_nodes:
        print(f"\n⚠️  Высокая EC reconstruction нагрузка на {len(problem_nodes)} нодах!")
        print("Возможные причины: нестабильные DataNode, сетевые проблемы, ")
        print("высокая конкуренция за ресурсы при Spark чтении EC данных.")
    else:
        print("\n✅ EC метрики в норме")

# Шаг 1: Создать директории для эксперимента
hdfs dfs -mkdir -p /benchmark/replication-3x
hdfs dfs -mkdir -p /benchmark/ec-rs-6-3

# Шаг 2: Проверить текущие политики (пока обе используют репликацию)
hdfs ec -getPolicy -path /benchmark/replication-3x
# ErasureCodingPolicy=null (Default: 3x replication)

hdfs ec -getPolicy -path /benchmark/ec-rs-6-3
# ErasureCodingPolicy=null

# Шаг 3: Установить EC политику на вторую директорию
# ВАЖНО: для RS-6-3 нужно минимум 9 DataNode!
# Проверяем количество доступных DataNode:
hdfs dfsadmin -report | grep "Live datanodes"
# Live datanodes (9)  ← нужно минимум 9 для RS-6-3

hdfs ec -setPolicy -path /benchmark/ec-rs-6-3 -policy RS-6-3-1024k

# Шаг 4: Верифицируем
hdfs ec -getPolicy -path /benchmark/ec-rs-6-3
# ErasureCodingPolicy=RS-6-3-1024k

# Шаг 5: Запомним свободное место ДО записи
hdfs dfs -df -h /benchmark/

# benchmark_ec_vs_replication.py
# Сравнение производительности Spark чтения для EC vs репликация данных

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


@dataclass
class BenchmarkResult:
    storage_type: str      # "replication-3x" или "ec-rs-6-3"
    write_time_sec: float  # время записи датасета
    disk_usage_gb: float   # занятое место после записи
    read_time_sec: float   # время аналитического чтения
    agg_time_sec: float    # время агрегации


def create_spark() -> SparkSession:
    return (
        SparkSession.builder
        .master("yarn")
        .appName("ec-vs-replication-benchmark")
        .config("spark.executor.instances", "9")   # по числу DataNode
        .config("spark.executor.cores", "4")
        .config("spark.executor.memory", "8g")
        # Нативные библиотеки для ISA-L EC decode
        .config("spark.executor.extraJavaOptions",
                "-Djava.library.path=/usr/lib/hadoop/lib/native")
        .config("spark.executorEnv.LD_LIBRARY_PATH",
                "/usr/lib/hadoop/lib/native")
        # Для EC данных locality.wait = 0 - NODE_LOCAL всё равно недостижим
        .config("spark.locality.wait", "0")
        .config("spark.sql.adaptive.enabled", "true")
        .getOrCreate()
    )


def get_hdfs_disk_usage_gb(path: str) -> float:
    """Возвращает реальный объём, занятый данными на HDFS (включая parity блоки)."""
    result = subprocess.run(
        ["hdfs", "dfs", "-du", "-s", "-h", path],
        capture_output=True, text=True
    )
    # Формат вывода: "1.5 G  4.5 G  /benchmark/replication-3x"
    # Первое число = размер данных, второе = реальное занятое место
    parts = result.stdout.split()
    if len(parts) >= 3:
        size_str = parts[1]  # второе число = DiskSpaceConsumed (с репликами/parity)
        multiplier = {"K": 1e-6, "M": 1e-3, "G": 1.0, "T": 1e3}.get(size_str[-1], 1.0)
        try:
            return float(size_str[:-1]) * multiplier
        except ValueError:
            return 0.0
    return 0.0


def generate_and_write_dataset(
    spark: SparkSession, output_path: str, n_rows: int = 10_000_000
) -> float:
    """
    Генерирует и записывает тестовый датасет.
    Возвращает время записи в секундах.
    """
    print(f"Генерация датасета ({n_rows:,} строк) → {output_path}")
    start = time.time()

    df = spark.range(n_rows).select(
        F.col("id").alias("transaction_id"),
        (F.rand() * 500_000).cast("long").alias("user_id"),
        (F.rand() * 50_000).cast("long").alias("merchant_id"),
        (F.rand() * 10_000.0).alias("amount"),
        F.array(F.lit("RUB"), F.lit("USD"), F.lit("EUR")).getItem(
            (F.rand() * 3).cast("int")
        ).alias("currency"),
        F.array(
            F.lit("purchase"), F.lit("refund"), F.lit("transfer"), F.lit("fee")
        ).getItem((F.rand() * 4).cast("int")).alias("tx_type"),
        F.date_add(F.lit("2022-01-01"), (F.rand() * 730).cast("int")).alias("tx_date"),
        F.current_timestamp().alias("created_at"),
    )

    # repartition(90) = 10 файлов на DataNode, равномерное распределение
    df.repartition(90, "tx_date").write \
        .mode("overwrite") \
        .partitionBy("tx_date") \
        .parquet(output_path)

    elapsed = time.time() - start
    print(f"Записано за {elapsed:.1f} секунд")
    return elapsed


def run_analytical_query(spark: SparkSession, data_path: str) -> float:
    """
    Аналитический запрос: агрегация транзакций по дате, мерчанту и типу.
    Полное сканирование всего датасета с многоуровневой агрегацией.
    """
    start = time.time()

    df = spark.read.parquet(data_path)

    result = (
        df
        .filter(F.col("tx_type") == "purchase")
        .groupBy(
            F.date_trunc("month", "tx_date").alias("month"),
            "merchant_id",
            "currency"
        )
        .agg(
            F.count("*").alias("tx_count"),
            F.sum("amount").alias("total_amount"),
            F.avg("amount").alias("avg_amount"),
            F.countDistinct("user_id").alias("unique_users"),
        )
        .orderBy(F.desc("total_amount"))
    )

    row_count = result.count()
    elapsed = time.time() - start
    print(f"Результат: {row_count:,} строк за {elapsed:.1f} секунд")
    return elapsed


def run_full_benchmark() -> None:
    spark = create_spark()

    REPLICATION_PATH = "hdfs://mycluster/benchmark/replication-3x"
    EC_PATH = "hdfs://mycluster/benchmark/ec-rs-6-3"
    N_ROWS = 10_000_000  # ~1 GB данных

    results = {}

    for storage_type, path in [("Репликация 3x", REPLICATION_PATH),
                                ("EC RS-6-3", EC_PATH)]:
        print(f"\n{'='*60}")
        print(f"Тест: {storage_type}")
        print(f"Путь: {path}")
        print(f"{'='*60}")

        # Запись
        write_time = generate_and_write_dataset(spark, path, N_ROWS)

        # Занятое место
        disk_gb = get_hdfs_disk_usage_gb(path)
        print(f"Занято на HDFS: {disk_gb:.1f} GB")

        # Прогрев (warm-up)
        print("\nПрогрев...")
        spark.read.parquet(path).count()

        # Чтение с агрегацией (3 запуска)
        read_times = []
        for i in range(1, 4):
            t = run_analytical_query(spark, path)
            read_times.append(t)
            print(f"  Запуск {i}: {t:.1f} сек")

        avg_read = sum(read_times) / len(read_times)

        results[storage_type] = {
            "write_time_sec": write_time,
            "disk_gb": disk_gb,
            "avg_read_time_sec": avg_read,
        }

        time.sleep(10)

    # Вывод сравнения
    print("\n" + "=" * 60)
    print("ИТОГОВОЕ СРАВНЕНИЕ")
    print("=" * 60)

    rep = results["Репликация 3x"]
    ec = results["EC RS-6-3"]

    disk_savings_pct = (1 - ec["disk_gb"] / rep["disk_gb"]) * 100
    read_slowdown = ec["avg_read_time_sec"] / rep["avg_read_time_sec"]

    print(f"{'Метрика':<35} {'Репликация 3x':>15} {'EC RS-6-3':>15}")
    print("-" * 65)
    print(f"{'Время записи (сек)':<35} {rep['write_time_sec']:>14.1f}s "
          f"{ec['write_time_sec']:>14.1f}s")
    print(f"{'Занято на HDFS (GB)':<35} {rep['disk_gb']:>14.1f}  "
          f"{ec['disk_gb']:>14.1f}")
    print(f"{'Среднее время чтения (сек)':<35} {rep['avg_read_time_sec']:>14.1f}s "
          f"{ec['avg_read_time_sec']:>14.1f}s")
    print()
    print(f"Экономия дискового пространства: {disk_savings_pct:.0f}%")
    print(f"Замедление чтения: {read_slowdown:.2f}x")
    print()
    print("ВЫВОД:")
    if disk_savings_pct > 40 and read_slowdown < 3:
        print("✅ EC RS-6-3 рекомендуется для холодных/архивных данных.")
        print(f"   50% экономии дисков при {read_slowdown:.1f}x замедлении чтения —")
        print("   приемлемый компромисс для редко читаемых данных.")
    else:
        print("⚠️  Результаты зависят от конфигурации кластера.")

    spark.stop()


if __name__ == "__main__":
    run_full_benchmark()

# После записи: проверяем как распределены EC блоки по DataNode
hdfs fsck /benchmark/ec-rs-6-3/tx_date=2022-01-01/ \
  -files -blocks -locations 2>&1 | head -50

# Пример вывода для EC файла:
# /benchmark/ec-rs-6-3/tx_date=2022-01-01/part-0000.parquet:
#   ErasureCoding policy=RS-6-3-1024k
#   Total size:    134217728 B
#   Total dirs:    0
#   Total files:   1
#   Total blocks (validated): 1
#   EC block group with missing blocks:0
#
#   Block group 0 of 1:
#     Data block 0: datanode1:50010 /data/hdfs/current/blk_001 cell=1048576
#     Data block 1: datanode2:50010 /data/hdfs/current/blk_002 cell=1048576
#     Data block 2: datanode3:50010 /data/hdfs/current/blk_003 cell=1048576
#     Data block 3: datanode4:50010 /data/hdfs/current/blk_004 cell=1048576
#     Data block 4: datanode5:50010 /data/hdfs/current/blk_005 cell=1048576
#     Data block 5: datanode6:50010 /data/hdfs/current/blk_006 cell=1048576
#     Parity block 0: datanode7:50010 /data/hdfs/current/blk_007 cell=1048576
#     Parity block 1: datanode8:50010 /data/hdfs/current/blk_008 cell=1048576
#     Parity block 2: datanode9:50010 /data/hdfs/current/blk_009 cell=1048576
#
# FSCK ended at ... with status HEALTHY

# Сравниваем с репликацией:
hdfs fsck /benchmark/replication-3x/tx_date=2022-01-01/ \
  -files -blocks -locations 2>&1 | head -30

# Пример вывода для реплицированного файла:
# Block 0 of 1:
#   Replica 0: datanode1:50010 /data/hdfs/current/blk_001
#   Replica 1: datanode3:50010 /data/hdfs/current/blk_001_r1
#   Replica 2: datanode7:50010 /data/hdfs/current/blk_001_r2
# (весь блок целиком на 3 разных DataNode)