План выполнения, Catalyst/Tungsten и типичные ошибки

Скрипт работает, цифры правильные, а ночной пайплайн «Гонца» с каждой неделей идёт дольше. Где искать? Не в коде Python, а в плане, который Spark из него построил. План говорит правду: сколько данных читается, где идёт shuffle, где включается Python. Сегодня прочитаем план целиком и соберём ошибки, которые в нём видны сразу.

Как читать план

Физический план печатает df.explain(). Это дерево, и читать его надо снизу вверх: внизу чтение данных, наверху результат. Каждая строка это оператор, а сдвиг вправо показывает, кто чей вход.

Главные операторы:

Оператор	Что делает	На что смотреть
FileScan	читает файлы	ReadSchema: какие колонки, PushedFilters и PartitionFilters: что отфильтровано у источника
Filter, Project	фильтр и выбор колонок	условия, которые не дошли до скана
HashAggregate	группировка	частичная (partial_) до shuffle и окончательная после
Exchange	shuffle	каждый Exchange это граница стадий и передача данных по сети
BroadcastHashJoin, SortMergeJoin	соединение	какая стратегия выбрана
ArrowEvalPython	Python UDF	пересылка строк в Python

AdaptiveSparkPlan isFinalPlan=false наверху значит, что план ещё будет уточнён во время выполнения: AQE может слить партиции или поменять стратегию соединения. После действия explain() покажет итоговый план.

У explain есть режимы. explain(True) печатает ещё и логические планы: какой запрос вы написали и как его переписал Catalyst. explain("formatted") печатает короткое дерево с номерами и подробности по каждому оператору отдельно. Так удобнее читать длинные планы.

Типичные ошибки

collect на драйвер. df.collect() забирает все строки в память драйвера. На миллиарде строк драйвер упадёт. Для просмотра берите show() или limit(n), а результат пишите в файл или таблицу.

Лишний shuffle. repartition(200) перед группировкой по другому ключу не помогает, а добавляет ещё один полный обмен: в плане два Exchange подряд.

Фильтр, который не дошёл до источника. Условие F.date_format("created_at", "yyyy-MM-dd") == "2026-06-30" для Parquet непонятно: колонку сначала надо превратить в строку. Spark поднимет все строки и отфильтрует их потом. Условие по самой колонке, created_at >= '2026-06-30' AND created_at < '2026-07-01', уйдёт в PushedFilters, и Parquet пропустит лишние блоки.

UDF вместо встроенной функции. ArrowEvalPython в плане, как в уроке про UDF.

Повторные действия без кэша. Три count() по одному DataFrame это три задания, каждое с самого чтения файлов.

Чтение всех колонок. Если в ReadSchema двадцать колонок, а нужны три, где-то в коде select стоит слишком поздно или вовсе потерян.