Task Scheduling: FIFO vs FAIR, locality levels

Когда Spark запускает groupBy().count(), он не просто «отправляет задачи». За кадром работает двухуровневая система планирования: Cluster Manager решает, сколько ресурсов выделить приложению, а Task Scheduler - как распределить конкретные Task по уже выделенным Executor-ам. Именно Task Scheduler определяет, почему одна задача запускается раньше другой, почему некоторые Task ждут своего Executor, и что происходит когда Executor падает.

Двухуровневая иерархия планирования¶

Cluster Manager работает на уровне приложений: он выделяет контейнеры/ресурсы для Driver и Executor-ов. Он не знает о конкретных Task - только о том, сколько CPU и памяти нужно каждому Executor.

Task Scheduler работает внутри Driver. Он получает TaskSet от DAG Scheduler и решает: какую Task запустить на каком Executor, в каком порядке, и что делать при сбоях. Это чисто логическая операция - Task Scheduler не управляет ресурсами кластера.

Иерархия Job → Stage → TaskSet → Task¶

Job создаётся при каждом Action. Один df.show() = один Job, df.count() = ещё один Job.

Stage - это граф вычислений без shuffle-границ. Порядок выполнения Stage определяется зависимостями: Stage 1 начинается только после завершения Stage 0.

TaskSet - набор Task для одного Stage. TaskSetManager управляет жизненным циклом TaskSet: отслеживает выполненные и упавшие Task, применяет retry-логику.

Task - минимальная единица выполнения. Содержит:

• Сериализованный байткод пользовательского кода
• Информацию о партиции (какой файл/блок читать)
• Список preferred locations (где хранятся данные для этой партиции)

TaskSetManager: управление TaskSet¶

TaskSetManager - внутренний компонент Task Scheduler, отвечающий за один TaskSet. Он отвечает на вопросы:

• Какую Task назначить на предложенный Executor?
• Что делать если Task упала?
• Как долго ждать Task на "предпочтительном" Executor-е?

Ключевой момент: когда Executor освобождается, он не спрашивает «что мне делать?» - наоборот, он сообщает о свободном слоте через reviveOffers(), и TaskScheduler решает, какую Task туда отправить.

Режимы планирования: FIFO и FAIR¶

Task Scheduler поддерживает два режима планирования нескольких Job в одном приложении. Это важно в случаях, когда несколько потоков (threads) одновременно запускают Action на одном SparkSession - например, в многопоточном сервере или Jupyter-ноутбуке с параллельными ячейками.

FIFO (First In, First Out)¶

Принцип: первый пришёл - первый обслуживается. Job 1 получает все ресурсы кластера. Job 2 (быстрая проверка, 5 секунд работы) ждёт 30 минут, пока завершится Job 1.

Приоритеты внутри FIFO: у Job есть числовой приоритет (по умолчанию равен jobId, меньше = выше). Можно задать явно:

spark.sparkContext.setJobGroup("analytics", "Daily metrics", interruptOnCancel=True)
spark.sparkContext.setLocalProperty("spark.scheduler.pool", "analytics")

Плюсы FIFO:

• Предсказуемость - первый Job всегда завершится максимально быстро
• Нет накладных расходов на перераспределение ресурсов
• Подходит для batch pipeline без конкурирующих Job

Минусы FIFO:

• Starvation - маленькие быстрые Job ждут завершения больших
• Не подходит для интерактивных запросов рядом с batch-задачами

FAIR (Fair Scheduling)¶

Принцип: ресурсы кластера делятся между всеми активными Job через round-robin. Быстрый Job не ждёт большой - он получает свою долю немедленно.

Включение FAIR:

spark = SparkSession.builder \
    .config("spark.scheduler.mode", "FAIR") \
    .getOrCreate()

Пулы в FAIR-планировании¶

Пулы - это именованные группы Job с настраиваемыми параметрами распределения ресурсов.

Конфиг fairscheduler.xml:

<?xml version="1.0"?>
<allocations>
  <!-- Пул для production pipeline -->
  <pool name="production">
    <schedulingMode>FIFO</schedulingMode>  <!-- внутри пула - FIFO -->
    <weight>3</weight>                     <!-- получает в 3× больше ресурсов чем default -->
    <minShare>10</minShare>                <!-- минимум 10 Task всегда -->
  </pool>

  <!-- Пул для аналитических запросов -->
  <pool name="analytics">
    <schedulingMode>FAIR</schedulingMode>  <!-- внутри пула - FAIR -->
    <weight>1</weight>
    <minShare>2</minShare>
  </pool>
</allocations>

# Подключить конфиг
spark = SparkSession.builder \
    .config("spark.scheduler.mode", "FAIR") \
    .config("spark.scheduler.allocation.file", "/path/to/fairscheduler.xml") \
    .getOrCreate()

# Назначить Job в пул (thread-local property)
spark.sparkContext.setLocalProperty("spark.scheduler.pool", "production")
result = df.groupBy("city").count()  # этот Job идёт в production-пул
result.show()

# Сбросить пул (вернуться в default)
spark.sparkContext.setLocalProperty("spark.scheduler.pool", None)

Параметры пула: weight и minShare¶

Алгоритм выбора пула в FAIR: Spark сравнивает «дефицит» каждого пула - насколько текущее потребление ресурсов ниже его weight-доли. Пул с наибольшим дефицитом получает следующий свободный слот. Если пул ниже minShare - он имеет абсолютный приоритет.

Data Locality: данные к коду, а не код к данным¶

Философия Spark: дешевле переместить код к данным, чем данные к коду. Task - это несколько килобайт сериализованного байткода. Партиция данных - это сотни мегабайт. Перемещение данных по сети на порядки дороже.

Поэтому Task Scheduler старается запустить Task там, где физически хранятся данные для её партиции.

Иерархия уровней локальности¶

PROCESS_LOCAL - данные находятся в памяти именно того Executor-а, на котором будет запущена Task. Это происходит когда:

• DataFrame закэширован (.cache() / .persist()) на этом Executor
• Broadcast-переменная уже доставлена на этот узел
• Предыдущая Task на этом же Executor создала данные в памяти

NODE_LOCAL - данные на том же физическом сервере, но в другом процессе. Например:

• HDFS DataNode хранит реплику блока на этом хосте
• ОС-кэш (page cache) после предыдущего чтения файла
• Другой Executor на том же хосте уже читал эти данные

RACK_LOCAL - данные на другом сервере, но в той же сетевой стойке. Сеть внутри rack значительно быстрее (обычно 10 Gbit/s против 1 Gbit/s между rack).

NO_PREF - данным безразлично, где выполняться. Это типично для данных, созданных в памяти без привязки к конкретному узлу (например, spark.range(1000)).

ANY - данные могут быть на любом узле, и у Task нет предпочтений. Возникает когда:

• Данные из внешней системы (S3, JDBC), где нет блочного размещения
• Все реплики HDFS-блока находятся на узлах без свободных Executor-ов

Стоимость каждого уровня¶

Delay Scheduling: как Spark ждёт лучшей локальности¶

Если идеальный Executor (PROCESS_LOCAL) занят - Task Scheduler не сразу опускается к следующему уровню. Он ждёт:

Параметры Delay Scheduling¶

Когда менять Delay Scheduling¶

Увеличить locality.wait (до 10–30 сек) если:

• Данные закэшированы - PROCESS_LOCAL критически важна
• Сеть медленная - ANY очень дорого
• Кластер большой и Executor-ы быстро освобождаются

Уменьшить или поставить 0 если:

• Данные из S3/JDBC - всё равно ANY, локальность не имеет смысла
• Cluster под нагрузкой - ожидание только тормозит Job
• Много маленьких Task - overhead от ожидания больше выгоды от локальности

# Отключить ожидание для S3-данных (всё равно ANY)
spark.conf.set("spark.locality.wait", "0")

# Долгое ожидание для кэшированных данных
spark.conf.set("spark.locality.wait.process", "30s")
spark.conf.set("spark.locality.wait.node", "10s")

Жизненный цикл Task: от создания до результата¶

Результат Task передаётся двумя путями:

• DirectTaskResult - если результат мал (< spark.task.maxDirectResultSize, default 1 MB): напрямую через Driver
• IndirectTaskResult - если результат велик: сохраняется в BlockManager Executor-а, Driver получает ссылку и забирает при необходимости

FetchFailed - особый тип ошибки. Возникает когда Task в Stage N не может скачать shuffle-данные у Executor из Stage N-1 (Executor упал после завершения своей Task). В этом случае TaskScheduler помечает Stage N-1 для повторного выполнения - только потерянные shuffle-файлы, не весь Stage.

Retry-механизм и отказоустойчивость¶

Speculative Execution: борьба со Straggler-ами¶

Straggler (отстающий) - Task, которая работает значительно дольше медианы по Stage. Причины:

• Медленный диск на узле
• GC-пауза в JVM Executor-а
• Перегрузка сети на конкретном узле
• Data Skew - одна Task получила непропорционально большую партицию

Одна Task из 200 может задержать весь Stage. Именно для этого существует Speculative Execution:

Параметры Speculative Execution:

Когда НЕ использовать Speculative Execution:

• Запись в базы данных или внешние системы - спекулятивная Task запишет данные дважды
• Не-идемпотентные операции - дублирование вызовет проблемы
• Data Skew - спекуляция не поможет, если задача медленная из-за реально большого объёма данных

spark.conf.set("spark.speculation", "true")
spark.conf.set("spark.speculation.multiplier", "2.0")
spark.conf.set("spark.speculation.quantile", "0.9")

Dynamic Allocation: автомасштабирование Executor-ов¶

Static allocation: задаёте --num-executors 20 и Spark держит 20 Executor-ов всё время Job - даже когда они простаивают между Stage.

Dynamic Allocation: Spark автоматически запрашивает новые Executor-ы когда задач много, и возвращает простаивающие кластер-менеджеру.

Параметры Dynamic Allocation:

spark = SparkSession.builder \
    .config("spark.dynamicAllocation.enabled", "true") \
    .config("spark.dynamicAllocation.minExecutors", "2") \
    .config("spark.dynamicAllocation.maxExecutors", "50") \
    .config("spark.dynamicAllocation.initialExecutors", "5") \
    .config("spark.dynamicAllocation.executorIdleTimeout", "60s") \
    .config("spark.dynamicAllocation.schedulerBacklogTimeout", "1s") \
    .getOrCreate()

Требование: Dynamic Allocation требует External Shuffle Service на YARN NodeManager-ах или аналога на K8s. Это отдельный процесс, который хранит shuffle-файлы после остановки Executor-а. Без него возвращённый Executor унесёт с собой свои shuffle-файлы, и следующий Stage не сможет их прочитать.

Анализ Task Scheduling в Spark UI¶

Вкладка Stages → задача по задаче¶

В Spark UI, кликнув на Stage → Task Metrics, видна таблица всех Task с колонками:

Что ищем:

• Много ANY → данные из S3 или отсутствие локальности. Рассмотреть кэширование или выбор правильного partitioner
• Большой Scheduler Delay → Executor-ы перегружены или cluster manager медленно отвечает
• Большой GC Time (> 10% Duration) → нехватка памяти Executor, объекты слишком часто создаются
• Резкий разброс Duration → Data Skew: одна Task обрабатывает намного больше данных

Вкладка Executors¶

Показывает распределение задач по Executor-ам. Если один Executor выполнил значительно больше Task чем другие - возможна проблема с locality (все данные на одном узле) или Speculative Execution не включён.

Практика: сравниваем FIFO и FAIR¶

import threading
from pyspark.sql import SparkSession
from pyspark.sql.functions import count

# Переключаемся на FAIR
spark = SparkSession.builder \
    .master("local[4]") \
    .config("spark.scheduler.mode", "FAIR") \
    .config("spark.sql.shuffle.partitions", "8") \
    .getOrCreate()

big_df   = spark.range(10_000_000).toDF("id")
small_df = spark.range(100).toDF("id")

results = {}

def run_big():
    spark.sparkContext.setLocalProperty("spark.scheduler.pool", "batch")
    results["big"] = big_df.groupBy((big_df.id % 1000).alias("bucket")).count().count()

def run_small():
    spark.sparkContext.setLocalProperty("spark.scheduler.pool", "interactive")
    results["small"] = small_df.count()

# Запустить оба Job одновременно
t1 = threading.Thread(target=run_big)
t2 = threading.Thread(target=run_small)

t1.start()
t2.start()
t1.join()
t2.join()

print(results)
# При FAIR: small завершится намного раньше big, хотя стартовал позже
# При FIFO: small ждёт big

spark.stop()

Откройте http://localhost:4040/jobs/ и посмотрите на временные метки обоих Job - наглядно видна разница между FIFO и FAIR.

Best practices для production¶

Выбор режима планирования:

• FIFO - batch pipeline с одним потоком выполнения. Один Job за раз, предсказуемость
• FAIR - shared кластер с несколькими командами, интерактивные запросы рядом с batch

Настройка Data Locality:

# Для S3/GCS - отключить ожидание, всё равно ANY
spark.conf.set("spark.locality.wait", "0")

# Для HDFS с кэшированием - долгое ожидание
spark.conf.set("spark.locality.wait.process", "10s")
spark.conf.set("spark.locality.wait.node", "5s")

Speculative Execution в production:

# Включить только если операции идемпотентны
spark.conf.set("spark.speculation", "true")
spark.conf.set("spark.speculation.quantile", "0.9")   # после 90% задач
spark.conf.set("spark.speculation.multiplier", "3.0") # только если в 3× медленнее

Dynamic Allocation:

# Production YARN
spark.conf.set("spark.dynamicAllocation.enabled", "true")
spark.conf.set("spark.dynamicAllocation.minExecutors", "5")   # никогда не ниже
spark.conf.set("spark.dynamicAllocation.maxExecutors", "100") # потолок
spark.conf.set("spark.shuffle.service.enabled", "true")       # обязательно с DA

Итог¶

Task Scheduler - это мозг оперативного планирования внутри Spark-приложения. Два ключевых решения, которые он принимает для каждой Task:

Когда запускать (режим планирования):

• FIFO - первый пришёл, первый обслужен. Один большой Job занимает весь кластер
• FAIR - round-robin между Job. Маленькие запросы не голодают рядом с большими

Где запускать (Data Locality + Delay Scheduling):

• Стратегия: PROCESS_LOCAL → NODE_LOCAL → RACK_LOCAL → ANY
• Spark ждёт locality.wait секунд на каждом уровне прежде чем понизить требования
• Правило: для S3 ставить locality.wait=0, для HDFS+кэша увеличивать

Поверх этого работают:

• Speculative Execution - дублирование Straggler-ов для устранения "хвостовой задержки"
• Dynamic Allocation - автомасштабирование Executor-ов по реальной нагрузке

Понимание этих механизмов объясняет: почему одни Task запускаются с задержкой, почему кэширование ускоряет повторные Action, и почему locality.wait=0 стоит ставить при работе с объектными хранилищами.