Apache Spark — DE Interview PrepПодготовка к DE-интервью

The mental model in one screenМентальная модель на одном экране

One-liner: "Prefer DataFrame/Dataset so Catalyst can optimize; drop to RDD only when you need control the optimizer can't give you."

В одну строку: «Предпочитай DataFrame/Dataset, чтобы Catalyst мог оптимизировать; спускайся к RDD только когда нужен контроль, который оптимизатор дать не может.»

What happens on an action (the end-to-end walk)Что происходит при action (полный путь)

# force broadcast (PySpark)
from pyspark.sql.functions import broadcast
big.join(broadcast(small), "key")
-- SQL hint
SELECT /*+ BROADCAST(dim) */ ... FROM fact JOIN dim ...

# принудительный broadcast (PySpark)
from pyspark.sql.functions import broadcast
big.join(broadcast(small), "key")
-- SQL hint
SELECT /*+ BROADCAST(dim) */ ... FROM fact JOIN dim ...

• cache() = persist(MEMORY_AND_DISK) for DataFrames (RDD cache = MEMORY_ONLY).
• Levels: MEMORY_ONLY · MEMORY_AND_DISK · *_SER · DISK_ONLY · *_2 (replicated) · off-heap.
• Cache when a DF is reused across multiple actions / iterations (ML loops). Don't cache a once-through pipeline.

• cache() = persist(MEMORY_AND_DISK) для DataFrames (кеш RDD = MEMORY_ONLY).
• Уровни: MEMORY_ONLY · MEMORY_AND_DISK · *_SER · DISK_ONLY · *_2 (реплицированный) · off-heap.
• Кешировать, когда DF переиспользуется через несколько actions / итераций (ML-циклы). Не кешировать одноразовый пайплайн.

AQE — Adaptive Query Execution (on by default 3.2+)

AQE — Adaptive Query Execution (по умолчанию с 3.2+)

Re-optimizes the physical plan at runtime from real shuffle stats:

Переоптимизирует физический план в runtime из реальной статистики shuffle:

• Coalesce shuffle partitions — collapse tiny post-shuffle partitions (kills the "200 default, mostly empty" problem)
• Switch join strategy — SMJ → BHJ when a side is small after filters
• Optimize skew joins — split skewed partitions into sub-partitions

• Слияние shuffle-партиций — сворачивает крошечные партиции после shuffle (убивает проблему «200 по умолчанию, в основном пустых»)
• Переключение стратегии джойна — SMJ → BHJ, когда сторона маленькая после фильтров
• Оптимизация перекоса в джойнах — разбивает перекошенные партиции на подпартиции

Skew toolkit (concrete)

Инструментарий для перекоса (конкретно)

1. AQE skew join — first line of defense.
2. Salt the hot key: suffix 0..N on both sides (replicate small side across salts), join on salted key, aggregate back.
3. Broadcast the small side → remove the shuffle entirely.
4. Isolate hot keys: process top-N separately (often broadcast), union the rest.
5. Pre-aggregate (map-side combine) before the shuffle.
6. Handle nulls: null join keys all hash to one partition — filter/coalesce them.

1. AQE skew join — первая линия защиты.
2. Соление (salt) горячего ключа: добавить суффикс 0..N с обеих сторон (реплицировать маленькую сторону по солям), джойнить по солёному ключу, агрегировать обратно.
3. Broadcast маленькой стороны → убрать shuffle полностью.
4. Изолировать горячие ключи: обрабатывать топ-N отдельно (часто broadcast), объединить остальное.
5. Предагрегация (комбинация на стороне map) перед shuffle.
6. Обработать null: null join-ключи все хешируются в одну партицию — фильтровать/сливать их.

• Execution can evict Storage beyond a floor; the reverse is bounded (asymmetric).
• Spill: execution memory exhausted → sort/agg buffers go to disk (slower, avoids OOM).
• OOM causes: huge broadcast · skewed partition too big · collect() to driver · too few partitions.

• Execution может вытеснить Storage сверх минимума; обратное ограничено (асимметрично).
• Spill: execution-память исчерпана → буферы сортировки/агрегации уходят на диск (медленнее, избегает OOM).
• Причины OOM: огромный broadcast · перекошенная партиция слишком большая · collect() на драйвер · слишком мало партиций.

• Model: stream = unbounded table; each micro-batch appends rows; same DataFrame API, incremental execution. (Continuous mode = low-latency, experimental.)
• Triggers: default (ASAP) · fixed interval · availableNow/once (batch backfill) · continuous
• Output modes: append (new rows; needs watermark for aggs) · update · complete (whole result, aggs only)
• Watermark: withWatermark("ts","10 minutes") bounds state for late data; later events dropped
• State + checkpoint: stateful ops keep a state store, checkpointed → exactly-once with replayable source (Kafka offsets) + idempotent sink

• Модель: поток = неограниченная таблица; каждый микро-батч добавляет строки; тот же API DataFrame, инкрементальное выполнение. (Continuous mode = низкая задержка, экспериментально.)
• Триггеры: по умолчанию (ASAP) · фиксированный интервал · availableNow/once (батч-backfill) · continuous
• Output modes: append (новые строки; нужен watermark для агрегатов) · update · complete (весь результат, только агрегаты)
• Watermark: withWatermark("ts","10 minutes") ограничивает state для поздних данных; более поздние события отбрасываются
• State + checkpoint: stateful-операции держат state store, чекпоинтятся → exactly-once с перечитываемым источником (Kafka offsets) + идемпотентным приёмником

The signature DE question: "this stage hangs at 199/200"

Фирменный DE-вопрос: «эта стадия зависла на 199/200»

1. Spark UI first — stage timeline + task metrics. 199/200 = classic skew.
2. Confirm skew: max vs median shuffle-read / duration (50x = skew).
3. Rule out: skewed key (null/hot) · too few partitions · spill · GC · straggler node · small-files explosion.
4. Fix by cause: AQE skew / salt / broadcast · filter nulls · raise partitions · more memory · speculative execution.
5. Verify in the UI: compare shuffle-read distribution & stage time.

1. Spark UI сначала — timeline стадии + метрики задач. 199/200 = классический перекос.
2. Подтвердить перекос: макс. vs медиана shuffle-read / продолжительность (50x = перекос).
3. Исключить: перекошенный ключ (null/горячий) · слишком мало партиций · spill · GC · отстающий узел · взрыв мелких файлов.
4. Исправить по причине: AQE skew / salt / broadcast · фильтровать null · увеличить партиции · больше памяти · speculative execution.
5. Проверить в UI: сравнить распределение shuffle-read и время стадии.

General tuning checklist

Общий чек-лист тюнинга

• Read less: column/predicate/partition pruning · columnar Parquet/ORC · compact small files
• Shuffle less: broadcast small joins · pre-aggregate · bucket on join key · avoid needless repartition
• Right-size parallelism: partitions ≈ 2–4× cores · let AQE coalesce
• Fix skew · cache only reused data
• Executor sizing: ~5 cores/executor heuristic (fat executors → GC/HDFS throughput pain) · memoryOverhead · dynamic allocation
• Avoid Python UDFs (serialization boundary) → built-in SQL fns or Arrow/pandas vectorized UDFs
• Measure before/after — never tune blind

• Читать меньше: обрезка колонок/предикатов/партиций · колоночный Parquet/ORC · компактировать мелкие файлы
• Shuffle меньше: broadcast маленьких джойнов · предагрегация · bucket по ключу джойна · избегать ненужных repartition
• Подобрать параллелизм: партиции ≈ 2–4× ядра · пусть AQE сливает
• Исправить перекос · кешировать только переиспользуемые данные
• Размер Executor: эвристика ~5 ядер/executor (жирные executors → боль GC/пропускная способность HDFS) · memoryOverhead · dynamic allocation
• Избегать Python UDFs (граница сериализации) → встроенные SQL-функции или Arrow/pandas векторизованные UDFs
• Измерять до/после — никогда не тюнить вслепую

Tap a card to flip. Answer out loud first.

Нажми на карточку, чтобы перевернуть. Сначала ответь вслух.