Динамическое отсечение разделов (Dynamic Partition Pruning) в Spark SQL: архитектура, алгоритмы и лучшие практики оптимизации пакетных запросов

Введение: цели, задачи и контекст оптимизации пакетных запросов в Spark SQL

Современные корпоративные аналитические ландшафты опираются на пакетные запросы, которые обрабатывают десятки и сотни терабайт данных. Главной статьей затрат в таких сценариях является ввод-вывод (I/O) и перемещение данных при соединениях больших партиционированных таблиц. Dynamic Partition Pruning (DPP, динамическое отсечение разделов)в Apache Spark SQL - это оптимизация, появившаяся в Spark 3.x, которая позволяет автоматически исключать из чтения разделы таблиц фактов, не участвующие в результате соединения. В центре подхода - перенос селективности с фильтров таблиц измерений на таблицу фактов по ключам партиционирования.

Цель статьи - дать профессиональному читателю системное понимание архитектуры DPP, объяснить алгоритмы и условия применимости, показать связь с Catalyst и AQE (Adaptive Query Execution), а также сформировать набор практических рекомендаций по промышленному внедрению оптимизации в средах с Hive Metastore, Lakehouse-форматами и облачными хранилищами.

Теоретическая база: партиционирование, селективность и стоимость чтения данных

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

Себестоимость пакетного запроса складывается из:

чтения метаданных и списков файлов разделов (catalog, файловая система);
чтения и декодирования файлов (Parquet/ORC/Avro);
передачи данных по сети при шинах shuffle/broadcast;
CPU при фильтрации, хэшировании и агрегациях.

DPP уменьшает первую и вторую составляющие, а при удачной конфигурации - еще и объем shuffle из-за радикального сужения фактического набора данных.

Архитектура Spark SQL: Catalyst, логическая оптимизация, физический план и AQE

Spark SQL использует оптимизатор Catalyst. Жизненный цикл запроса включает:

построение логического плана;
применение правил логической оптимизации (перенос предикатов, упрощение вычислений);
планирование физических операторов (сканирование, join, exchange);
оптимизацию и исполнение физического плана.

AQE (Adaptive Query Execution) вносит корректировки уже во время выполнения, используя реальные статистики, чтобы перераспределять данные, изменять стратегии join и коалесцировать партиции. DPP взаимодействует с обоими уровнями: вставляется на этапе логической оптимизации и реализуется в физическом плане, а с AQE усиливается за счет лучшей статистической осведомленности о селективности.

Что такое Dynamic Partition Pruning и зачем он нужен в Spark SQL

Dynamic Partition Pruning - это автоматическое добавление динамического предиката к сканированию партиционированной таблицы, когда в соединении присутствует значительно меньшая таблица с селективными фильтрами по ключу соединения, совпадающему с ключом партиционирования большой таблицы. Предикат формируется на лету из фактических значений, прошедших фильтры на стороне измерения, и применяется к таблице фактов до чтения данных. В результате не читаются целые разделы, заведомо не дающие строк результата.

Типичный сценарий - звездообразная схема: таблица фактов (партиционирована по date_key, geo_key и т.д.) соединяется с одной или несколькими таблицами измерений, на которых есть фильтры (например, конкретные даты, страны или продуктовые подкатегории). Чем больше несоответствующих разделов удается отбросить, тем сильнее снижение времени и стоимости запроса.

Единицы параллелизма и влияние числа разделов на эффективность обработки

В Spark единицей параллелизма является раздел (partition). В каждый момент исполнитель (executor task) обрабатывает один раздел. Избыток маленьких разделов приводит к накладным расходам на планирование и синхронизацию, а также к перегрузке драйвера на этапе подготовки метаданных. Недостаток разделов уменьшает параллелизм. DPP не повышает параллелизм напрямую, но уменьшает число разделов, которые требуется прочесть, и тем самым сокращает длительность фаз чтения и подготовки данных для join и агрегаций.

Рекомендация: поддерживать разумное количество разделов в больших фактовых таблицах (обычно десятки-сотни на «срезе»), чтобы балансировать между эффективным pruning и управляемой стоимостью метаданных.

Механизм работы DPP: перенос фильтров, широковещательная фильтрация и пропуск нерелевантных разделов

На логическом уровне DPP извлекает из правой (или левой) стороны соединения фильтры по ключу join и строит подзапрос, который будет транслирован (broadcast) и превращен в динамический список значений. На физическом уровне результат трансляции используется как IN-фильтрдля партиционного столбца фактовой таблицы. Сканер источника данных получает сокращенный список разделов к чтению, и нерелевантные разделы пропускаются.

Именно поэтому DPP особенно эффективен в паре с источниками, поддерживающими ускоренное разбиение и фильтрацию по метаданным (Hive Metastore, Delta Lake, Iceberg, Hudi, каталоги DataSource V2).

Декомпозиция технических компонентов и их взаимодействие в DPP

Механизм включает:

правило логической оптимизации PartitionPruning;
выражение DynamicPruning (динамический предикат вида IN (подзапрос));
планировщик физических фильтров PlanDynamicPruningFilters;
правило очистки CleanupDynamicPruningFilters;
физические операторы BroadcastExchangeExec, BroadcastHashJoinExec (BHJ), BroadcastNestedLoopJoinExec (BNLJ);
источники данных, умеющие применять partition filters к списку разделов до чтения.

Связка компонент обеспечивает перенос фильтрованного множества ключей из небольшой стороны соединения в фазу сканирования крупной партиционированной таблицы.

Правило PartitionPruning: критерии применимости и вставка динамического предиката

PartitionPruning анализирует дерево плана и:

Определяет, можно ли отфильтровать сканирование таблицы по конкретному атрибуту партиционирования (getFilterableTableScan).
Проверяет тип соединения на применимость pruning для левой стороны (canPruneLeft). Для левой стороны поддержаны, как минимум, Inner и LeftSemi, а также RightOuter, поскольку левая сторона не является «сохраняемой» в смысле внешней семантики результата.
Проверяет наличие селективного предиката на другой стороне соединения, который затрагивает ключ соединения (hasPartitionPruningFilter).
Если условия для левой стороны не выполняются, такие же проверки применяются к правой стороне (canPruneRight).

При удовлетворении условий вставляется предикат через insertPredicate. Перед вставкой вызывается оценка выгод/затрат методом pruningHasBenefit (см. раздел 9). Если выгода положительна или разрешено принудительное повторное использование широковещания, динамический предикат добавляется в план.

Алгоритм pruningHasBenefit: модель затрат-выгод, статистики и fallbackFilterRatio

Оценка выгод/затрат ориентировочно сравнивает:

ожидаемую экономию от DPP = sizeInBytes(prunable side) × filterRatio;
стоимость поддержки DPP = sizeInBytes(other side), которая отражает накладные расходы на подготовку широковещательного фильтра и логической вставки.

Ключевой параметр - filterRatio:

если доступны статистики столбцов (NDV, number of distinct values), оценивается на их основе;
если статистик нет или они отключены, используется конфигурация spark.sql.optimizer.dynamicPartitionPruning.fallbackFilterRatio (часто 0.5 по умолчанию).

Влияние статистик регулируется флагом spark.sql.optimizer.dynamicPartitionPruning.useStats. Чем точнее статистики (ANALYZE TABLE), тем более корректно решение о целесообразности DPP.

CleanupDynamicPruningFilters и PlanDynamicPruningFilters: устранение дубликатов и повторное использование BroadcastExchange

Вставка предиката на логическом уровне может привести к появлению дублирующегося подзапроса. Далее:

PlanDynamicPruningFiltersпытается заменить дубликат повторно используемым BroadcastExchange, если план содержит совместимый BroadcastHashJoinExec и разрешено повторное использование (см. конфигурации ниже).
Если повторное использование невозможно, но расчетная выгода все еще выше, чем исполнение плана без DPP, подзапрос сохраняется.
В противном случае дублирующийся подзапрос удаляется (CleanupDynamicPruningFilters), чтобы избежать лишних накладных расходов.

Таким образом, переиспользование трансляции критичнодля минимизации затрат на DPP, особенно в сложных планах со множеством joins.

Физические операторы и распределение: BroadcastExchangeExec, BroadcastHashJoinExec, BroadcastNestedLoopJoinExec

BroadcastExchangeExec - оператор, который собирает результат дочернего поддерева и рассылает его всем исполнителям. Он появляется, когда физический план требует BroadcastDistribution.
BroadcastHashJoinExec - основной оператор для широковещательного соединения, строит хэш-таблицу из «маленькой» стороны и зондирует ее «большой» стороной.
BroadcastNestedLoopJoinExec - применяется, когда хэш-соединение невозможно (нет эквисоединения), но объем трансляции все еще приемлем.

DPP стремится использовать готовую трансляцию, созданную для BHJ/BNLJ, чтобы не инициировать дополнительный broadcast только ради динамической фильтрации.

Поддерживаемые типы JOIN и ограничения на сокращение разделов

Поддержка зависит от того, какую сторону допускается «сужать», не нарушая семантику join:

Для левой стороны: Inner, LeftSemi, RightOuter** - безопасны для pruning левой части.
Для правой стороны: Inner, LeftOuter и некоторые варианты полу-соединений в зависимости от версии Spark.

Типы LeftAnti/FullOuter требуют осторожности: часто они либо не поддерживаются, либо конкретные стороны недопустимы для pruning, поскольку это может изменить результат. Фактический список зависит от версии Spark 3.x, поэтому при миграциях имеет смысл проверять объяснение плана и релиз-ноты.

Источники и форматы данных: статические и динамические разделы, обнаружение новых партиций

DPP работает как со статическими, так и с динамически появляющимися (инкрементально добавленными) разделами. Ключевое требование - источник должен:

предоставлять явные partition filters (через метаданные каталога, например Hive Metastore, или через DataSource V2 интерфейсы);
корректно находить и регистрировать новые разделы (MSCK REPAIR TABLE в Hive, автообнаружение в Delta/Iceberg/Hudi, рефреш метаданных в каталоге).

В форматах Parquet и ORC DPP синергирует с predicate pushdown по статистике файла/страницы, однако именно партиционное отсечение дает наибольший выигрыш, поскольку полностью исключает чтение файлов из нерелевантных разделов.

Конфигурация и параметры Spark SQL для DPP: enabled, reuseBroadcastOnly, useStats и пр.

Ниже приведены ключевые параметры, влияющие на DPP и смежные механизмы.

Параметр	Назначение	Типичные значения
spark.sql.optimizer.dynamicPartitionPruning.enabled	Включение DPP	true (по умолчанию)
spark.sql.optimizer.dynamicPartitionPruning.reuseBroadcastOnly	Разрешать DPP только при повторном использовании BroadcastExchange	true (по умолчанию)
spark.sql.optimizer.dynamicPartitionPruning.useStats	Использовать столбцовые статистики для оценки filterRatio	true
spark.sql.optimizer.dynamicPartitionPruning.fallbackFilterRatio	Резервное значение селективности при отсутствии статистик	0.5 (часто по умолчанию)
spark.sql.autoBroadcastJoinThreshold	Порог размера для BHJ	зависит от кластера (например, 10-100 МБ)
spark.sql.execution.reuseExchange	Повторное использование Exchange	true
spark.sql.execution.reuseSubquery	Повторное использование подзапросов	true
spark.sql.adaptive.enabled	Включить AQE	true

На практике для промышленных кластеров рекомендуются включенные reuseExchange/reuseSubquery и AQE, чтобы минимизировать накладные расходы на дополнительный broadcast, вызванный DPP.

Интеграция технологических стеков и их синергия: Hive Metastore, Delta Lake/Iceberg/Hudi, Parquet/ORC, облачные хранилища

Hive Metastore обеспечивает быстрый доступ к списку разделов таблиц и хранит столбцовые статистики (ANALYZE TABLE). Это база для корректного DPP в «классическом» озере данных.
Lakehouse-форматы (Delta Lake, Apache Iceberg, Apache Hudi) предоставляют дополнительные метаданные (манифесты, метаданные снимков), что ускоряет вычисление набора разделов. DPP отлично сочетается с манифестами Iceberg и transaction log Delta, снижая стоимость листинга в облачных FS.
На облачных хранилищах (S3, ADLS, GCS) DPP сокращает количество обращений к объектному стореджу и общий объем считанных байтов, что напрямую снижает затраты.

Сопутствующие оптимизации: predicate pushdown, репартиционирование, коалесцирование, broadcast reuse и AQE

Predicate pushdown (PDP) отбрасывает строки на уровне файлов/страниц; DPP работает уровнем выше - отбрасывает целые разделы до чтения.
Репартиционирование по ключам join (repartition(cols)) до соединения уменьшает shuffle и перекосы, стабилизируя селективность DPP.
Коалесцирование выходных партиций после сильного pruning сокращает число задач и издержки на мелкие файлы.
Reuse Broadcast и AQE снижают накладные расходы, повышая отдачу от DPP.

Совместное применение этих техник дает кумулятивный эффект.

Моделирование данных под DPP: звездная схема, размеры измерений и стратегия партиционирования

При проектировании хранилищ и витрин учитывайте:

Партиционируйте факты по ключам, которые часто попадают в фильтры измерений (дата, регион, бренд), и имеют умеренный NDV, обеспечивающий высокую селективность.
Измерения должны быть существенно меньше фактов, чтобы их можно было транслировать или извлекать селективные ключи без чрезмерной стоимости.
Не дробите разделение чрезмерно: слишком мелкое партиционирование увеличивает накладные расходы на метаданныеи ослабляет стабильность плана.

Частый паттерн - партиционирование фактов по d_date и d_country с дополнительным кластерингом (sorting/bucketing) для ключей join.

Кейсы применения в реальных сценариях: BI-запросы, срезы и кубы, инкрементальные загрузки

BI и интерактивная аналитика: запросы вида «прошлая неделя по 5 рынкам и 3 категориям»; DPP исключает подавляющую долю разделов по времени и гео.
Срезы и кубы: комбинации измерений многократно переиспользуют фильтры, и DPP предотвращает чтение неиспользуемых срезов.
Инкрементальные загрузки: новые разделы автоматически учитываются каталогом, и DPP продолжает работать без ручной донастройки.

На витринах метрик с большим временным охватом DPP способен сократить сканирование в десятки раз.

Руководство по повышению эффективности: ранняя фильтрация, ANALYZE TABLE, выбор числа разделов и repartition(cols)

Эффективное применение DPP усиливается следующими приемами:

Выполняйте раннюю фильтрацию в DataFrame-логике на стороне измерений, чтобы сузить динамический фильтр.
Периодически запускайте ANALYZE TABLE [tbl] COMPUTE STATISTICS FOR COLUMNS (...), чтобы обеспечить качественные NDVи улучшить решения pruningHasBenefit.
Держите количество разделов в разумных пределах; ориентир - 10-100 на крупную таблицу для стабильной стоимости метаданных (подбирается опытно).
Используйте repartition(cols) перед join по соответствующим ключам, чтобы избежать случайных кросс-соединений и перекосов, подрывающих селективность DPP.

Диагностика и верификация: EXPLAIN-планы, метрики Spark UI и индикаторы применения DPP

Проверить срабатывание DPP можно через EXPLAIN и Spark UI.

В EXPLAIN FORMATTED/EXTENDED ищите подзапросы DynamicPruning и пометки о reuse broadcast:

== Physical Plan ==
*(1) Project ...
+- *(1) BroadcastHashJoin [fact.part_col = dim.key], Inner, BuildRight
   :- *(1) FileScan parquet fact ... PushedFilters: ..., DynamicPruning: part_col IN subquery
   +- BroadcastExchange HashedRelationBroadcastMode(keys=[key#...])
      +- *(2) Project ... (filters on dim)

В Spark UI анализируйте:

объем байтов прочитанных из источников;
число выходных строк на стадиях сканирования фактов;
наличие метрик повторного использования Broadcast (если отображается в конкретной версии);
снижение количества задач в стадиях чтения из-за отсечения разделов.

Метрики эффективности и методика бенчмаркинга: доля отброшенных разделов, сокращение байтов сканирования, время и ресурсы

Методика бенчмарка:

Зафиксировать исходную выборку запросов и входные данные.
Выполнить с DPP выключенным и собирать:
- число прочитанных разделов/файлов;
- объем прочитанных байтов;
- wall-clock время;
- затраты CPU и shuffle.
Включить DPP и сравнить те же метрики.
Повторить цикл при разных значениях fallbackFilterRatio и autoBroadcastJoinThreshold.

Ключевые KPI:

доля отброшенных разделов (PartitionPruning Rate);
сокращение байтов сканирования (I/O Reduction);
общее ускорение и экономия стоимости инфраструктуры.

Анализ рисков, уязвимостей и ограничений: неселективные фильтры, перекосы данных, неточность статистик, недопустимые типы JOIN

Основные риски:

Низкая селективность фильтров - DPP не приносит выгоды, а иногда вредит, если требуется дополнительный broadcast без reuse.
Перекосы данных по партиционному ключу - часть разделов слишком «тяжелы», что приводит к дисбалансу задач.
Неточные или устаревшие статистики - метод pruningHasBenefit принимает неверные решения.
Типы соединений и выражения предикатов, несовместимые с безопасным отсечением разделов.
Ограничения потоковой обработки - DPP относится к пакетным оптимизациям, не применяется к Structured Streaming.

Варианты отказа и устранение неполадок: когда DPP не срабатывает и как это исправить

Наиболее частые причины отсутствия DPP:

Ключ join не совпадает со столбцом партиционирования фактической таблицы. Исправление: изменить схему партиционирования или переназначить ключи join.
Источник данных не экспонирует partition filters на уровне каталога/сканера. Исправление: использовать таблицы в каталоге (Hive Metastore, Delta/Iceberg/Hudi) или перейти на DataSource V2 коннекторы.
Конфигурация reuseBroadcastOnly=true при отсутствии пригодного broadcast. Исправление: временно установить reuseBroadcastOnly=false или увеличить autoBroadcastJoinThreshold.
Отсутствуют статистики и слишком низкий fallbackFilterRatio. Исправление: ANALYZE TABLE и тюнинг fallbackFilterRatio.
Тип JOIN не поддерживает pruning требуемой стороны. Исправление: перестроить запрос (например, заменить на эквисоединение, изменить порядок соединений).

Возможности применения в экономических секторах: финансы, ритейл, телеком, производство, здравоохранение, медиа и госсектор

Финансы: анализ транзакций по окнам дат, странам и продуктам; DPP отсекает подавляющее большинство исторических разделов.
Ритейл и e-commerce: отчеты по акциям/категориям/магазинам; селективные фильтры на измерениях сильно сужают факты.
Телеком: KPI по сотам/регионам и периодам; высокая эффективность при партиционировании по времени и географии.
Производство и IoT: телеметрия по станкам и сменам; отбор по оборудованию и периодам.
Здравоохранение: когорты пациентов и эпизоды лечения; фильтры по периодам и учреждениям.
Медиа: просмотры/клики по каналам и кампаниям.
Госсектор: реестры и статистика по периодам и регионам.

Конкурентный анализ и дифференциация: DPP vs статическое обрезание, runtime filtering (Bloom), решения Hive/Presto/Trino/Impala

Статическое партиционное отсечение использует только константные предикаты запроса (например, WHERE dt='2026-01-01'). DPP добавляет динамику, извлекая значения из другой таблицы в join.
Runtime filtering (например, на основе Bloom-фильтров в Trino/Presto/Impala) может отбрасывать строки не только на границе разделов, но и в пределах файлов; однако такие механизмы требуют надежной передачи фильтров и совместимости со сканерами. В Spark аналогичный эффект часто достигается комбинацией DPP + predicate/page pushdown.
В Hive до появления Spark 3.x динамическое отсечение поддерживалось, но реализация и область применимости отличались. Catalyst+DPP в Sparkобеспечивает более глубокую интеграцию с планировщиком и AQE, что дает преимущество в смешанных нагрузках Lakehouse.

Архитектурные рекомендации и best practices для промышленного внедрения DPP

Проектируйте партиционирование фактов по «горячим» измерениям, которые часто участвуют в фильтрах и соединениях; избегайте партиционных ключей с крайне высоким NDV при скромной селективности.
Включайте DPP и добивайтесь повторного использования BroadcastExchange (reuseExchange/reuseSubquery=true). Настраивайте autoBroadcastJoinThreshold так, чтобы типичные измерения попадали в BHJ.
Регулярно обновляйте статистики ANALYZE TABLE (включая столбцовые), чтобы принятие решений pruningHasBenefit было корректным.
Стандартизируйте порядок соединений: маленькие и селективные измерения - ближе к источнику, факты - после фильтрации.
Интегрируйте DPP с AQE, PDP, репартиционированием по ключам соединений и коалесцированием партиций на выходе.
Используйте каталоги, поддерживающие эффективный список разделов (Hive Metastore, Delta, Iceberg, Hudi), особенно в облаках.
В конвейерах CI/CD автоматизируйте EXPLAIN-валидации: проверяйте наличие DynamicPruning в ключевых запросах.

Заключение и направления дальнейших исследований

Dynamic Partition Pruning - зрелая оптимизация Spark SQL, приносящая ощутимые выигрыши в пакетной аналитике, особенно в звездообразных схемах и Lakehouse-архитектурах. Ее сила - в переносе селективности на уровень сканирования партиционированных фактов и синергии с Catalyst, AQE и современными форматами таблиц. Дальнейшее развитие ожидается в направлениях:

более точные runtime-статистики и адаптивная коррекция filterRatio;
расширение набора поддерживаемых join-паттернов и выражений предикатов;
интеграция с легковесными runtime-фильтрами (включая Bloom) и улучшенные API DataSource V2.

Компании, инвестирующие в методичную настройку DPP и сопутствующих механизмов, получают двойной эффект: ускорение SLA и снижение операционных затрат на кластеры.

Вопрос-Ответ:

Вопрос: Что такое Dynamic Partition Pruning в двух словах?
Ответ: Это динамическое отсечение нерелевантных партиций фактовой таблицы на основе фильтров измерений, вычисляемых во время выполнения join.
Вопрос: Когда DPP приносит наибольшую пользу?
Ответ: В звездообразных схемах при соединении большой партиционированной таблицы фактов с малыми селективно отфильтрованными измерениями.
Вопрос: Какие ключевые условия применимости DPP?
Ответ: Совпадение ключа join со столбцом партиционирования на «сужаемой» стороне, поддерживаемый тип join и наличие селективных фильтров на противоположной стороне.
Вопрос: Как DPP соотносится с predicate pushdown?
Ответ: DPP отсекает целые разделы до чтения, а predicate pushdown отбрасывает строки/страницы внутри файлов; вместе они дают максимальный эффект.
Вопрос: Зачем нужны статистики и ANALYZE TABLE?
Ответ: Для оценки filterRatio в pruningHasBenefit. Точные NDV позволяют принять правильное решение о вставке динамического предиката.
Вопрос: Какие параметры конфигурации критичны?
Ответ: enabled, reuseBroadcastOnly, useStats, fallbackFilterRatio, а также reuseExchange/reuseSubquery и autoBroadcastJoinThreshold.
Вопрос: Почему DPP не работает в потоковых запросах?
Ответ: Это пакетная оптимизация, опирающаяся на план и трансляции, несовместимые с непрерывной моделью Structured Streaming.
Вопрос: Как проверить, что DPP сработал?
Ответ: Через EXPLAIN (ищите DynamicPruning и reuse broadcast) и в Spark UI по резкому снижению читаемых байтов и количества сканируемых разделов.

← Предыдущая статья

Облачная архитектура Magnit F&R: высоконагруженная платформа прогнозирования и пополнения для цепочек поставок в реальном времени

Следующая статья →

Изоляция транзакций в Apache Kafka при потреблении сообщений