Executor Sizing: 4-компонентная формула памяти и правило 5 cores

# Параметры физического сервера
total_cpu = 24
total_ram_gb = 96

# Шаг 1: Резервируем ресурсы под ОС и YARN NodeManager
# Правило: 1 CPU + 1 GB для ОС, 1 CPU + 1 GB для NodeManager
os_cpu = 1
nm_cpu = 1
os_ram_gb = 1
nm_ram_gb = 1

available_cpu = total_cpu - os_cpu - nm_cpu   # = 22
available_ram_gb = total_ram_gb - os_ram_gb - nm_ram_gb  # = 94 GB

# Шаг 2: Вычисляем количество Executor'ов по правилу 5 cores
executor_cores = 5

# Число Executor'ов = floor(available_cpu / executor_cores)
num_executors_per_node = available_cpu // executor_cores  # = 22 // 5 = 4

# Шаг 3: Вычисляем RAM на Executor
# available_ram / num_executors (с небольшим запасом)
ram_per_executor_raw = available_ram_gb / num_executors_per_node  # = 94 / 4 = 23.5 GB

# Округляем вниз для безопасности
ram_per_executor_gb = 23  # GB

print(f"Сервер {total_cpu} vCPU / {total_ram_gb} GB:")
print(f"  Executor'ов на ноду: {num_executors_per_node}")
print(f"  Ядер на Executor: {executor_cores}")
print(f"  RAM на Executor: {ram_per_executor_gb} GB")

# Проверка: суммарное потребление
total_used_cpu = num_executors_per_node * executor_cores + os_cpu + nm_cpu  # 20 + 2 = 22 ✓
total_used_ram = num_executors_per_node * ram_per_executor_gb + os_ram_gb + nm_ram_gb  # 92 + 2 = 94 ✓
print(f"Утилизация: CPU={total_used_cpu}/{total_cpu}, RAM={total_used_ram}/{total_ram_gb} GB")

Узлы:             5
Executor'ов всего: 5 × 4 = 20 (один оставляем Driver'у)
Итого Executor'ов: 19 (рабочие) + 1 Driver
Общий параллелизм: 19 × 5 = 95 одновременных Task'ов

Total Container Memory =
    spark.executor.memory          (On-Heap JVM Heap)
  + spark.executor.memoryOverhead  (Off-Heap нативная память)
  + spark.memory.offHeap.size      (Tungsten Off-Heap, если включён)
  + spark.executor.pyspark.memory  (Python Workers, только PySpark)

# Пример расчёта для executor.memory = 20 GB
executor_memory_mb = 20 * 1024  # = 20480 MB

reserved_mb = 300
usable_mb = executor_memory_mb - reserved_mb  # = 20180 MB

memory_fraction = 0.6
unified_mb = usable_mb * memory_fraction  # = 12108 MB ≈ 11.8 GB
user_mb = usable_mb * (1 - memory_fraction)  # = 8072 MB ≈ 7.9 GB

print(f"executor.memory = 20 GB:")
print(f"  Reserved:       {reserved_mb} MB (фиксировано)")
print(f"  Unified Memory: {unified_mb:.0f} MB ({unified_mb/1024:.1f} GB)")
print(f"  User Memory:    {user_mb:.0f} MB ({user_mb/1024:.1f} GB)")
print()
print(f"Unified Memory делится между:")
print(f"  Execution Memory (Shuffle, Sort, Join) ↔ динамически ↔ Storage Memory (cache)")

# Формула расчёта по умолчанию:
import math

executor_memory_mb = 20 * 1024  # 20 GB = 20480 MB
overhead_fraction = 0.10        # 10%

overhead_mb = max(384, math.ceil(executor_memory_mb * overhead_fraction))
# = max(384, ceil(20480 * 0.10)) = max(384, 2048) = 2048 MB ≈ 2 GB

print(f"Memory Overhead = max(384MB, 10% × executor.memory)")
print(f"= max(384, {executor_memory_mb * 0.10:.0f}) = {overhead_mb} MB = {overhead_mb/1024:.1f} GB")

# Правило: base overhead × multiplier
def calculate_overhead(executor_memory_gb: float, workload_type: str) -> float:
    """
    Рекомендуемый overhead для разных типов нагрузки.
    """
    base_overhead_gb = max(0.375, executor_memory_gb * 0.10)

    multipliers = {
        "standard_scala":  1.0,  # дефолт, стандартная Scala/Java ETL
        "pyspark_basic":   1.5,  # PySpark без тяжёлых UDF
        "pyspark_pandas":  2.5,  # PySpark с Pandas UDF / Apache Arrow
        "heavy_shuffle":   1.5,  # большой Shuffle, интенсивная сеть
        "compression_ops": 2.0,  # работа с zip/bzip2/lz4 нативно
        "ml_tensorflow":   3.0,  # ML фреймворки с нативными библиотеками
    }

    multiplier = multipliers.get(workload_type, 1.0)
    return base_overhead_gb * multiplier

# Примеры:
for workload in ["standard_scala", "pyspark_pandas", "ml_tensorflow"]:
    oh = calculate_overhead(20, workload)
    print(f"  {workload}: overhead = {oh:.1f} GB")

spark = SparkSession.builder \
    # Включить Off-Heap Tungsten память
    .config("spark.memory.offHeap.enabled", "true") \

    # Размер Off-Heap пула на Executor
    # Рекомендация: 10-30% от executor.memory для начала
    # Увеличьте если видите тяжёлый GC при сортировках/агрегациях
    .config("spark.memory.offHeap.size", "4g") \

    .getOrCreate()

spark = SparkSession.builder \

    # spark.executor.pyspark.memory: сколько памяти разрешено КАЖДОМУ Python Worker'у
    # Это НЕ суммарная память всех Python Workers, а лимит на один Worker-процесс!
    # При 5 cores → потенциально 5 одновременных Python Workers
    # Суммарная Python память = pyspark.memory × cores ≈ 2 GB × 5 = 10 GB

    # Правило расчёта:
    # 1. Оцените максимальный размер pandas batch: batch_size × avg_row_size
    # 2. Добавьте overhead библиотек: ~500 MB для стандартного стека
    # 3. Умножьте на 2 (промежуточные вычисления удваивают потребление)

    .config("spark.executor.pyspark.memory", "2g") \

    # Суммарная память контейнера теперь:
    # executor.memory (20g) + memoryOverhead (3g) + pyspark.memory (2g × 5 cores)
    # = 20 + 3 + 10 = 33 GB на контейнер

    .getOrCreate()

def calculate_pyspark_memory(
    batch_size_rows: int,      # число строк в одном Pandas batch
    avg_row_size_bytes: int,   # средний размер одной строки в байтах
    n_cores: int = 5,          # cores на Executor
    library_overhead_mb: int = 512,  # pandas + numpy + прочее
) -> dict:
    """
    Рассчитывает рекомендуемый spark.executor.pyspark.memory.

    batch_size_rows: задаётся через spark.sql.execution.arrow.maxRecordsPerBatch
    (дефолт: 10000 строк).
    """
    # Размер одного batch в MB
    batch_mb = (batch_size_rows * avg_row_size_bytes) / (1024 * 1024)

    # Промежуточные данные при обработке (удваиваем)
    processing_mb = batch_mb * 2

    # Итого на один Worker
    per_worker_mb = batch_mb + processing_mb + library_overhead_mb

    # Рекомендуем с запасом 20%
    recommended_mb = int(per_worker_mb * 1.2)

    # Суммарная Python память на весь Executor
    total_python_mb = recommended_mb * n_cores

    return {
        "batch_size_mb": batch_mb,
        "per_worker_recommended_mb": recommended_mb,
        "total_python_memory_mb": total_python_mb,
        "suggested_config": f"{recommended_mb // 1024}g" if recommended_mb >= 1024
                           else f"{recommended_mb}m"
    }


# Пример: Pandas UDF обрабатывает транзакции
result = calculate_pyspark_memory(
    batch_size_rows=10000,    # дефолт Arrow batch
    avg_row_size_bytes=200,   # 200 байт/строка (10 колонок × 20 байт)
    n_cores=5,
    library_overhead_mb=512   # pandas + numpy
)
print("Рекомендуемая pyspark.memory:")
print(f"  Per Worker: {result['per_worker_recommended_mb']} MB")
print(f"  Total (5 cores): {result['total_python_memory_mb']} MB")
print(f"  Config: spark.executor.pyspark.memory = {result['suggested_config']}")

# Вариант 1: Увеличить executor.memory
spark.conf.set("spark.executor.memory", "32g")

# Вариант 2: Увеличить число партиций (меньше данных на Task)
spark.conf.set("spark.sql.shuffle.partitions", "400")

# Вариант 3: Освободить Storage Memory (uncache неиспользуемое)
spark.catalog.uncacheTable("old_cached_table")

# Вариант 4: AQE coalesce (поднять advisoryPartitionSizeInBytes)
spark.conf.set("spark.sql.adaptive.advisoryPartitionSizeInBytes", "128MB")

from pyspark.sql import SparkSession
import json

spark = SparkSession.builder.getOrCreate()


def get_executor_stats(spark: SparkSession) -> list[dict]:
    """
    Собирает метрики Executor'ов через SparkContext StatusTracker.
    Полезно для programmatic мониторинга в production.
    """
    sc = spark.sparkContext
    status = sc.statusTracker()

    executor_infos = status.getExecutorInfos()
    stats = []

    for executor in executor_infos:
        # Получаем метрики через REST API History Server (если доступен)
        # Здесь упрощённая версия через доступные API
        stats.append({
            "executor_id": executor.executorId(),
            "host": executor.host(),
            "num_running_tasks": executor.numRunningTasks(),
            "num_failed_tasks": executor.numFailedTasks(),
            "num_completed_tasks": executor.numCompletedTasks(),
        })

    return stats


def check_gc_pressure(spark: SparkSession, gc_threshold_pct: float = 10.0) -> list[str]:
    """
    Проверяет GC pressure на всех Executor'ах.
    Возвращает список Executor'ов с проблемным GC.
    """
    # Для доступа к GC Time используем History Server REST API:
    # GET /api/v1/applications/{appId}/executors
    # Поле: "totalGCTime", "totalDuration"

    # Здесь показываем логику:
    problem_executors = []

    # Пример данных из History Server API response:
    executor_metrics = [
        {"executorId": "1", "totalGCTime": 5000, "totalDuration": 100000},
        {"executorId": "2", "totalGCTime": 25000, "totalDuration": 100000},  # 25% GC!
    ]

    for ex in executor_metrics:
        if ex["totalDuration"] > 0:
            gc_pct = ex["totalGCTime"] / ex["totalDuration"] * 100
            if gc_pct > gc_threshold_pct:
                problem_executors.append(
                    f"Executor {ex['executorId']}: GC={gc_pct:.1f}% (порог={gc_threshold_pct}%)"
                )

    return problem_executors

class ClusterCalculator:
    """
    Калькулятор конфигурации Spark Executor'ов.
    Реализует стандартную методологию Executor Sizing.
    """

    def __init__(
        self,
        nodes: int,
        cpu_per_node: int,
        ram_per_node_gb: int,
        workload_type: str = "pyspark_pandas",
    ):
        self.nodes = nodes
        self.cpu_per_node = cpu_per_node
        self.ram_per_node_gb = ram_per_node_gb
        self.workload_type = workload_type

    def calculate(self) -> dict:
        # ── Шаг 1: Резервирование под ОС и менеджер узлов ───────────────
        # ОС: 1 CPU + 1 GB RAM
        # NodeManager/Kubelet: 1 CPU + 1 GB RAM
        # Итого резерв: 2 CPU + 2 GB RAM
        os_cpu = 1
        nm_cpu = 1
        os_ram_gb = 1
        nm_ram_gb = 1

        available_cpu = self.cpu_per_node - os_cpu - nm_cpu         # 24 - 2 = 22
        available_ram_gb = self.ram_per_node_gb - os_ram_gb - nm_ram_gb  # 96 - 2 = 94 GB

        # ── Шаг 2: Правило 5 cores ────────────────────────────────────────
        executor_cores = 5
        executors_per_node = available_cpu // executor_cores        # 22 // 5 = 4
        unused_cpu = available_cpu % executor_cores                  # 22 % 5 = 2 CPU (потеря)

        # ── Шаг 3: Базовая RAM на Executor ───────────────────────────────
        raw_ram_per_ex_gb = available_ram_gb / executors_per_node   # 94 / 4 = 23.5 GB
        # Округляем вниз: 23 GB для executor.memory + overhead
        total_ram_per_ex_gb = int(raw_ram_per_ex_gb)                # 23 GB на все 4 компонента

        # ── Шаг 4: Рассчитываем Memory Overhead ──────────────────────────
        # PySpark + Pandas → overhead × 2.5
        overhead_multipliers = {
            "standard_scala":  1.0,
            "pyspark_basic":   1.5,
            "pyspark_pandas":  2.5,
            "heavy_shuffle":   1.5,
            "ml_tensorflow":   3.0,
        }
        overhead_base_gb = max(0.375, total_ram_per_ex_gb * 0.10)   # max(375MB, 2.3GB) = 2.3 GB
        overhead_multiplier = overhead_multipliers.get(self.workload_type, 1.0)
        overhead_gb = round(overhead_base_gb * overhead_multiplier, 1)  # 2.3 × 2.5 = 5.75 ≈ 5.8 GB

        # ── Шаг 5: pyspark.memory (Python Workers) ───────────────────────
        # PySpark pandas: 2 GB × cores (консервативно)
        pyspark_per_worker_gb = 2.0
        pyspark_total_gb = pyspark_per_worker_gb * executor_cores   # 2 × 5 = 10 GB

        # НО pyspark.memory - это лимит на один Worker, не суммарный!
        pyspark_config_gb = pyspark_per_worker_gb                   # 2 GB per worker

        # ── Шаг 6: executor.memory = total - overhead - pyspark × cores ──
        # Мы хотим: executor.memory + overhead + pyspark_total ≤ total_ram_per_ex_gb
        executor_memory_gb = total_ram_per_ex_gb - overhead_gb - pyspark_total_gb
        # = 23 - 5.8 - 10 = 7.2 GB
        # Маловато! Нужно пересмотреть.

        # Корректировка: уменьшаем pyspark.memory
        # Для batch_size=10K строк × 200 байт = ~2 MB batch
        # 2 MB × 3 (overhead) + 512 MB (libs) ≈ 518 MB → округляем до 1 GB
        pyspark_per_worker_gb = 1.0
        pyspark_total_gb = pyspark_per_worker_gb * executor_cores   # 1 × 5 = 5 GB
        executor_memory_gb = total_ram_per_ex_gb - overhead_gb - pyspark_total_gb
        # = 23 - 5.8 - 5 = 12.2 → берём 12 GB

        executor_memory_gb = 12

        # ── Шаг 7: Проверка ───────────────────────────────────────────────
        actual_total = executor_memory_gb + overhead_gb + pyspark_total_gb
        # = 12 + 5.8 + 5 = 22.8 GB ≤ 23 GB ✓

        # ── Шаг 8: Итоговые параметры кластера ───────────────────────────
        # Один Executor оставляем под Driver на головной ноде
        total_executors = self.nodes * executors_per_node
        working_executors = total_executors - 1  # минус Driver
        total_parallel_tasks = working_executors * executor_cores

        return {
            "nodes": self.nodes,
            "executors_per_node": executors_per_node,
            "total_executors": total_executors,
            "working_executors": working_executors,
            "executor_cores": executor_cores,
            "executor_memory_gb": executor_memory_gb,
            "memory_overhead_gb": overhead_gb,
            "pyspark_memory_per_worker_gb": pyspark_per_worker_gb,
            "total_container_memory_gb": actual_total,
            "total_parallel_tasks": total_parallel_tasks,
            "unused_cpu_per_node": unused_cpu,
            "spark_submit_params": {
                "executor-cores": executor_cores,
                "executor-memory": f"{executor_memory_gb}g",
                "num-executors": working_executors,
                "conf_memory_overhead": f"{int(overhead_gb * 1024)}m",
                "conf_pyspark_memory": f"{pyspark_per_worker_gb}g",
            }
        }

    def print_report(self) -> None:
        result = self.calculate()
        print(f"\n{'='*60}")
        print(f"Executor Sizing Report")
        print(f"Кластер: {result['nodes']} нод × {self.cpu_per_node} CPU × {self.ram_per_node_gb} GB RAM")
        print(f"Нагрузка: {self.workload_type}")
        print(f"{'='*60}")
        print(f"\nКонфигурация Executor:")
        print(f"  executor-cores:       {result['executor_cores']}")
        print(f"  executor-memory:      {result['executor_memory_gb']} GB")
        print(f"  memoryOverhead:       {result['memory_overhead_gb']} GB")
        print(f"  pyspark.memory:       {result['pyspark_memory_per_worker_gb']} GB / worker")
        print(f"  Total Container:      {result['total_container_memory_gb']:.1f} GB")
        print(f"\nРазмещение на кластере:")
        print(f"  Executors на ноду:   {result['executors_per_node']}")
        print(f"  Всего Executors:     {result['total_executors']}")
        print(f"  Рабочих Executors:   {result['working_executors']} (1 для Driver)")
        print(f"  Параллельных Tasks:  {result['total_parallel_tasks']}")
        print(f"\nspark-submit параметры:")
        params = result['spark_submit_params']
        print(f"  --executor-cores {params['executor-cores']} \\")
        print(f"  --executor-memory {params['executor-memory']} \\")
        print(f"  --num-executors {params['num-executors']} \\")
        print(f"  --conf spark.executor.memoryOverhead={params['conf_memory_overhead']} \\")
        print(f"  --conf spark.executor.pyspark.memory={params['conf_pyspark_memory']}")


# Запускаем расчёт для задания
calc = ClusterCalculator(
    nodes=5,
    cpu_per_node=24,
    ram_per_node_gb=96,
    workload_type="pyspark_pandas"
)
calc.print_report()

from pyspark.sql import SparkSession

# Параметры из калькулятора
spark = SparkSession.builder \
    .master("yarn") \
    .appName("etl-pyspark-production") \

    # ── Executor ресурсы ──────────────────────────────────────────────
    .config("spark.executor.instances", "19") \
    .config("spark.executor.cores", "5") \
    .config("spark.executor.memory", "12g") \

    # ── Memory Overhead: PySpark + Pandas требует 2.5× дефолта ────────
    .config("spark.executor.memoryOverhead", "5836m") \

    # ── Python Workers ────────────────────────────────────────────────
    .config("spark.executor.pyspark.memory", "1g") \

    # ── Driver (на головной ноде) ─────────────────────────────────────
    .config("spark.driver.memory", "12g") \
    .config("spark.driver.cores", "4") \
    .config("spark.driver.memoryOverhead", "2g") \

    # ── AQE: для динамической оптимизации ────────────────────────────
    .config("spark.sql.adaptive.enabled", "true") \
    .config("spark.sql.adaptive.coalescePartitions.enabled", "true") \
    .config("spark.sql.adaptive.advisoryPartitionSizeInBytes", "128mb") \

    # ── Arrow для Pandas UDF: оптимизация передачи данных ────────────
    .config("spark.sql.execution.arrow.pyspark.enabled", "true") \
    .config("spark.sql.execution.arrow.maxRecordsPerBatch", "10000") \

    # ── GC: G1GC для умеренных heap (< 32 GB) ────────────────────────
    .config("spark.executor.extraJavaOptions",
            "-XX:+UseG1GC -XX:G1HeapRegionSize=16m "
            "-XX:+PrintGCDetails -XX:+PrintGCDateStamps") \

    .enableHiveSupport() \
    .getOrCreate()

# Сравнительная таблица: дефолты vs оптимизированная конфигурация

configurations = {
    "Дефолтная (Fat Executor)": {
        "executor_cores": 24,
        "executor_memory_gb": 90,
        "overhead_gb": 9,       # 10%
        "pyspark_memory_gb": 0, # не задан!
        "num_executors": 5,
        "parallel_tasks": 120,
        "container_gb": 99,
        "risks": [
            "Full GC 30-90 сек при 90GB Heap",
            "Python workers неограничены → OOM",
            "HDFS 24-поточный троттлинг",
        ]
    },
    "Правильная (5-core)": {
        "executor_cores": 5,
        "executor_memory_gb": 12,
        "overhead_gb": 5.8,
        "pyspark_memory_gb": 1,  # per worker
        "num_executors": 19,
        "parallel_tasks": 95,
        "container_gb": 22.8,
        "risks": [
            "Стабильная работа G1GC",
            "Python Workers ограничены 1 GB/worker",
            "HDFS оптимальный параллелизм",
        ]
    }
}

print(f"{'Параметр':35s} {'Fat Executor':20s} {'5-core Optimal':20s}")
print("-" * 75)
print(f"{'executor.cores':35s} {'24':20s} {'5':20s}")
print(f"{'executor.memory':35s} {'90 GB':20s} {'12 GB':20s}")
print(f"{'memoryOverhead':35s} {'9 GB':20s} {'5.8 GB':20s}")
print(f"{'pyspark.memory':35s} {'0 (опасно!)':20s} {'1 GB / worker':20s}")
print(f"{'num.executors':35s} {'5':20s} {'19':20s}")
print(f"{'Параллельных Tasks':35s} {'120':20s} {'95':20s}")
print(f"{'Container Memory':35s} {'99 GB':20s} {'22.8 GB':20s}")
print(f"{'GC Stop-The-World':35s} {'30-90 сек':20s} {'< 1 сек':20s}")
print(f"{'OOM риск':35s} {'ВЫСОКИЙ':20s} {'НИЗКИЙ':20s}")
print(f"{'Fault Tolerance':35s} {'Плохая':20s} {'Хорошая':20s}")

# Проверить что контейнеры запустились с нужными параметрами
yarn application -status <APPLICATION_ID>

# Смотрим логи конкретного Executor
yarn logs -applicationId <APPLICATION_ID> -containerId <CONTAINER_ID> 2>&1 | grep -E "GC|OOM|killed"

# Мониторим в реальном времени
watch -n 5 'yarn application -status <APP_ID> | grep -E "State|Progress|Container"'

# Программная проверка через SparkContext
def verify_executor_config(spark: SparkSession) -> None:
    """Проверяет что применились нужные конфигурации."""
    sc = spark.sparkContext
    configs_to_check = [
        "spark.executor.cores",
        "spark.executor.memory",
        "spark.executor.memoryOverhead",
        "spark.executor.pyspark.memory",
        "spark.executor.instances",
    ]
    print("Активные конфигурации Executor:")
    for key in configs_to_check:
        try:
            value = spark.conf.get(key)
            print(f"  {key} = {value}")
        except Exception:
            print(f"  {key} = НЕ ЗАДАН (дефолт)")

verify_executor_config(spark)