Spark UI: вкладка Jobs — DAG визуализация и поиск узкого места

Job (1 Action)
  ├── Stage 0 (чтение файлов + фильтрация)
  │     ├── Task 0 (Partition 0 → один файл или его часть)
  │     ├── Task 1 (Partition 1)
  │     └── ... Task N
  │
  ├── Stage 1 (после Shuffle: groupBy aggregation)
  │     ├── Task 0 (hash partition 0)
  │     ├── Task 1 (hash partition 1)
  │     └── ... Task N
  │
  └── Stage 2 (после Shuffle: join)
        ├── Task 0
        └── ...

ID  | Description        | Submitted | Duration | Stages | Tasks
----|--------------------|-----------|----------|--------|------
0   | count at Main.py:42| 10:01:15  | 45 s     | 3/3    | 1200/1200
1   | save at Main.py:57 | 10:02:05  | 2m 30s   | 4/4    | 2400/2400
2   | show at Main.py:63 | 10:04:40  | 3 s      | 2/2    | 200/200

df = spark.read.parquet("s3://bucket/events/")     # Stage 0 begins
filtered = df.filter(df.amount > 100)              # ↑ Narrow
enriched = filtered.join(F.broadcast(products),    # BHJ: остаётся в Stage 0
                         "product_id")
grouped = enriched.groupBy("category").sum("amount")  # Wide! Stage 1 begins
grouped.write.parquet("s3://bucket/result/")

Stage 0: Scan parquet → Filter → BroadcastHashJoin → Partial Aggregation
Stage 1: Exchange (hashpartitioning) → Sort → Aggregation → Write

spark = SparkSession.builder \
    # Включает спекулятивное выполнение
    .config("spark.speculation", "true") \

    # Задача считается "медленной" если выполняется в multiplier раз
    # дольше медианы всех задач данного Stage
    .config("spark.speculation.multiplier", "1.5") \

    # Запускать спекуляцию только если завершено >= quantile задач
    .config("spark.speculation.quantile", "0.75") \

    .getOrCreate()

spark = SparkSession.builder \
    # Dynamic Allocation: Spark сам добавляет/удаляет Executor'ы
    .config("spark.dynamicAllocation.enabled", "true") \

    # Минимальное число Executor'ов (всегда доступно)
    .config("spark.dynamicAllocation.minExecutors", "2") \

    # Максимальное число Executor'ов
    .config("spark.dynamicAllocation.maxExecutors", "50") \

    # Удалять Executor если он простаивал N секунд
    # ПРОБЛЕМА: если установить слишком маленькое значение →
    # кластер постоянно убивает и создаёт Executor'ы
    # → overhead на JVM startup, потеря кешированных данных!
    .config("spark.dynamicAllocation.executorIdleTimeout", "120s") \

    # Executor с кешированными данными — дольше ждём
    .config("spark.dynamicAllocation.cachedExecutorIdleTimeout", "3600s") \

    .getOrCreate()

df = spark.read.parquet("s3://bucket/large_dataset/")  # 500 GB
filtered = df.filter(df.year == 2024)

# Кешируем дорогостоящий промежуточный результат
filtered.cache()
filtered.count()  # Job 0: материализует кеш
# ↑ Stage 0 (Scan + Filter) → выполнен, данные в памяти

# Повторное использование
result1 = filtered.groupBy("region").count()
result1.show()  # Job 1: Stage 0 → SKIPPED! читаем из кеша

result2 = filtered.join(another_df, "user_id").count()
result2.count()  # Job 2: Stage 0 → SKIPPED! снова из кеша

# Проверяем что кеш реально сохранился
df.cache()
df.count()  # материализуем

# Проверяем в каталоге:
print(spark.catalog.isCached("my_table"))  # для таблиц

# Через Storage вкладку: смотрим на RDD Blocks
# Если видим "0 Blocks Cached" → кеш не занял место

# Типичные причины "кеш не работает":
# 1. Слишком большой DataFrame, не помещается в Storage Memory
# 2. Executor был убит, кеш потерян
# 3. .unpersist() вызван раньше времени
# 4. .cache() вызван но Action не запущен

# Антипаттерн: кешируем всё подряд
df1 = spark.read.parquet("s3://bucket/orders/").cache()
df2 = spark.read.parquet("s3://bucket/customers/").cache()
df3 = df1.join(df2, "customer_id").cache()  # уже кешированы df1 и df2 → тройное занятие памяти!
df4 = df3.groupBy("region").count().cache()  # и это тоже...

# Последствия:
# 1. Storage Memory переполнена
# 2. Execution Memory вытеснена → больше Spill на диск
# 3. GC давление растёт из-за большого heap
# 4. В Spark UI → Storage вкладка: видим что % кеша в памяти уменьшается

# Правильный подход: кешировать только то что используется 2+ раз
# и занимает дорогостоящие вычисления для пересчёта

# В Spark UI: Jobs → сортируем по Duration (клик на заголовок колонки)
# Или программно через History Server API:
import requests

def find_slowest_jobs(app_id: str, history_server_url: str) -> list[dict]:
    """
    Находит 5 самых медленных Job'ов в приложении.
    Полезно для автоматического мониторинга SLA.
    """
    url = f"{history_server_url}/api/v1/applications/{app_id}/jobs"
    jobs = requests.get(url, timeout=30).json()

    completed_jobs = [j for j in jobs if j.get("status") == "SUCCEEDED"]
    sorted_jobs = sorted(
        completed_jobs,
        key=lambda j: j.get("completionTime", 0) - j.get("submissionTime", 0),
        reverse=True
    )

    return [{
        "jobId": j["jobId"],
        "name": j.get("name", "")[:60],
        "duration_ms": j.get("completionTime", 0) - j.get("submissionTime", 0),
        "stages": j.get("numCompletedStages", 0),
        "tasks": j.get("numCompletedTasks", 0),
    } for j in sorted_jobs[:5]]

Пример DAG:
Stage 0: Scan facts (10 сек) ─────────────────────┐
Stage 1: Scan dim_1 (2 сек) → Join (5 сек) ────── join ─→ Stage 4: Write (3 сек)
Stage 2: Scan dim_2 (1 сек) → Join (3 сек) ───────┘     (критический путь: 10+5+3=18 сек)

Ускорение Stage 2 (1+3=4 сек) до 0 → итог: 18 сек (без изменений!)
Ускорение Stage 0 (10 сек) до 3 сек → итог: 11 сек (значительное улучшение!)

Симптом → Вероятная причина → Следующий шаг

"Stage с 200 Tasks, но только 2 быстро, остальные висят"
→ Data Skew или ресурсов недостаточно
→ Смотреть Task distribution в Stage (медиана vs максимум)

"Jobs выполняются 10+ секунд с 0 Tasks"
→ Overhead планирования: HMS запросы, Python init, Driver computation
→ Смотреть Event Timeline, логи Driver'а

"Много Skipped Stages, но Job всё равно медленный"
→ Кеш неэффективен: данные не в Storage Memory (на диске или evicted)
→ Смотреть Storage вкладку: Fraction Cached, Storage Level

"Job Failed с Exit Code 137"
→ OOM Killer убил контейнер (превышен memory limit)
→ Увеличить executor.memoryOverhead или уменьшить executor.memory

"Stage с огромным Shuffle Read (> 100 GB)"
→ Слишком мало партиций → каждая партиция огромная
→ Увеличить spark.sql.shuffle.partitions или включить AQE

□ 1. Открыть Jobs → отсортировать по Duration
□ 2. Найти Job с максимальным Duration → нажать
□ 3. Проверить: есть ли Failed Stage'ы? (красный цвет)
□ 4. Если есть Failed → прочитать ошибку, найти Root Cause
□ 5. Посмотреть DAG: сколько Stage'ов? Есть ли широкие зависимости?
□ 6. Найти самый долгий Stage (по цвету/ширине полосы)
□ 7. Проверить: Stage Skipped? → верификация кеширования
□ 8. Нажать на проблемный Stage → перейти на вкладку Stages
□ 9. В Stage проверить: Task Distribution (Median vs Max)
□ 10. Если Max >> Median → Data Skew, применить AQE или salting
□ 11. Если равномерно медленно → Shuffle, Memory, I/O проблема
□ 12. Записать гипотезу и проверить через конкретные метрики

from pyspark.sql import SparkSession, functions as F
import time

spark = SparkSession.builder \
    .master("local[4]") \
    .appName("spark-ui-demo") \
    .config("spark.sql.shuffle.partitions", "200") \
    .config("spark.ui.enabled", "true") \
    .getOrCreate()

spark.sparkContext.setLogLevel("WARN")

print("Spark UI: http://localhost:4040")


# ── ДЕМОНСТРАЦИЯ 1: Лишние пересчёты без .cache() ──────────────────────
df = spark.range(10_000_000).select(
    F.col("id"),
    (F.rand() * 100).cast("int").alias("category"),
    (F.rand() * 1000).alias("revenue"),
)

# Job 0 (Action 1 без cache): пересчёт от нуля
count1 = df.count()  # Job 0: полный scan
# Job 1 (Action 2 без cache): ещё раз от нуля!
result1 = df.groupBy("category").sum("revenue")
result1.count()  # Job 1: полный scan + groupBy

# Смотрите в Spark UI → Jobs:
# Job 0 и Job 1 имеют почти одинаковые DAG → дублирование!

# ── ДЕМОНСТРАЦИЯ 2: С .cache() → Skipped Stage'ы ─────────────────────────
df_cached = spark.range(10_000_000).select(
    F.col("id"),
    (F.rand() * 100).cast("int").alias("category"),
    (F.rand() * 1000).alias("revenue"),
)
df_cached.cache()

count2 = df_cached.count()   # Job 2: материализация кеша
result2 = df_cached.groupBy("category").sum("revenue")
result2.count()  # Job 3: Stage 0 → SKIPPED! Намного быстрее!

# ── ДЕМОНСТРАЦИЯ 3: Data Skew → Straggler Task ───────────────────────────
from pyspark.sql.types import LongType

# Скошенные данные: 90% строк с key=0
skewed_data = spark.range(1_000_000).select(
    F.when(F.rand() < 0.9, F.lit(0)).otherwise(F.col("id") % 100).alias("join_key"),
    (F.rand() * 100).alias("value"),
)
lookup = spark.range(100).select(
    F.col("id").alias("join_key"),
    F.concat(F.lit("cat_"), F.col("id")).alias("category")
)

# Без AQE: один Task получит 90% данных → Straggler!
spark.conf.set("spark.sql.adaptive.enabled", "false")
joined = skewed_data.join(lookup, "join_key")
joined.count()  # Job 4: смотрите один медленный Task в Stages!

# С AQE: Bloom Filter + Skew Join → равномерно
spark.conf.set("spark.sql.adaptive.enabled", "true")
joined_aqe = skewed_data.join(lookup, "join_key")
joined_aqe.count()  # Job 5: намного быстрее!


print("""
ЗАДАНИЕ: Откройте Spark UI http://localhost:4040

Вкладка Jobs:
1. Сравните Duration Job 0 и Job 2 (с cache vs без)
2. Найдите Job 3 → убедитесь что есть Skipped Stage'ы
3. Сравните Job 4 (без AQE) и Job 5 (с AQE)
   - Job 4 → нажмите на Stage с Join → посмотрите Distribution
   - Должен быть один Task намного медленнее остальных

Вкладка DAG:
- Нажмите на любой Job → посмотрите граф Stage'ов
- Найдите Exchange (Shuffle) узлы на границах Stage'ов
- Найдите InMemoryTableScan в Job 3 (зелёный узел кеша)
""")

spark.stop()