Оптимизация запросов к озеру данных

В этом документе описаны основные методы и стратегии оптимизации запросов к данным озера (Hive, Iceberg, Paimon и др.).

Отсечение разделов

Условия по столбцам разделения в запросах позволяют отсечь ненужные разделы и сократить объём данных, которые требуется прочитать.

С помощью EXPLAIN <SQL> можно просмотреть секцию partition узла XXX_SCAN_NODE, чтобы проверить, работает ли отсечение разделов и сколько разделов потребуется просканировать при выполнении этого запроса.

Например:

0:VPAIMON_SCAN_NODE(88)
    table: paimon_ctl.db.table
    predicates: (user_id[#4] = 431304818)
    inputSplitNum=15775, totalFileSize=951754154566, scanRanges=15775
    partition=203/0

Локальный кеш данных

Data Cache ускоряет последующие запросы к тем же данным, кешируя недавно использованные файлы данных из удалённых систем хранения (HDFS или объектного хранилища) на локальном диске.

По умолчанию кеширование отключено. Инструкции по настройке и включению приведены в документации Data Cache.

Начиная с версии 4.0.2 поддерживается прогрев кеша, который позволяет заранее заполнять кеш данных для повышения производительности запросов.

Оптимизация чтения HDFS

См. раздел Оптимизация IO для HDFS в документации по HDFS.

Ограничение количества сплитов

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

1. Нагрузка на память: слишком большое количество сплитов потребляет значительный объём памяти FE
2. Ошибки OOM: чрезмерное количество сплитов может вызвать OutOfMemoryError
3. Снижение производительности: управление слишком большим количеством сплитов увеличивает накладные расходы на планирование запроса

Переменная сеанса max_file_split_num позволяет ограничить максимальное количество сплитов при одном сканировании таблицы (поддерживается начиная с версии 4.0.4):

• Тип: int
• По умолчанию: 100000
• Описание: максимальное количество сплитов при одном сканировании таблицы в непакетном режиме, позволяющее предотвратить OOM из-за слишком большого количества сплитов.

Пример использования:

-- Set maximum split count to 50000
SET max_file_split_num = 50000;

-- Disable this limit (set to 0 or negative number)
SET max_file_split_num = 0;

Если это ограничение задано, Doris динамически вычисляет минимальный размер сплита, чтобы количество сплитов не превышало указанного предела.

File Scanner V2

Поддерживается начиная с версии 4.1.4.

File Scanner V2 представляет собой новый механизм сканирования файлов внешних таблиц. Он поддерживает форматы Parquet, ORC, CSV и JSON, а также чтение таблиц в таких форматах, как Hive, Iceberg, Paimon и Hudi. По сравнению с прежним сканером он улучшает отсечение данных и параллельное разбиение, а также повышает эффективность использования памяти.

Им управляет переменная сеанса enable_file_scanner_v2:

• Тип: boolean
• По умолчанию: true (включён по умолчанию)
• Описание: если переменная включена, FileScanNode использует File Scanner V2 для сканирования в поддерживаемых сценариях выполнения запросов. В таких сценариях, как работа с JDBC Catalog и системными таблицами Iceberg, по-прежнему используется прежний путь сканирования.

Если вы подозреваете, что проблема с запросом связана с новым сканером, можно временно отключить его и сравнить результаты:

SET enable_file_scanner_v2 = false;

Поведение при отсечении данных

File Scanner V2 всегда выполняет безопасное отсечение разделов и отсечение по ZoneMap для выражений, и эти операции больше не зависят от перечисленных ниже переменных сеанса. Эти две переменные влияют только на прежний сканер, который по-прежнему учитывает их значения:

Переменная	По умолчанию	Описание
enable_runtime_filter_partition_prune	true	Включает отсечение разделов с помощью Runtime Filter. File Scanner V2 всегда выполняет безопасное отсечение разделов, поэтому значение false не влияет на его работу
enable_expr_zonemap_filter	false	Включает фильтрацию выражений по ZoneMap. File Scanner V2 всегда выполняет безопасную фильтрацию выражений по ZoneMap

Детализация сплитов на стороне BE

Поддерживается начиная с версии 4.1.4.

Когда File Scanner V2 включён, FE выполняет только крупное разбиение файлов, а BE дополнительно делит каждый сплит по Parquet Row Group или ORC Stripe и обрабатывает полученные части параллельно в рамках одного экземпляра сканирования. Соседние Row Groups / Stripes объединяются, пока их суммарный размер не приблизится к целевому.

Переменная сеанса	По умолчанию	Описание
file_split_size_on_fe	536870912 (512MB)	Целевой размер в байтах для крупных сплитов файлов, создаваемых FE при использовании детализации на стороне BE
file_split_size_on_be	67108864 (64MB)	Целевой размер в байтах для мелких сплитов файлов, создаваемых на BE

Область применения: нативные сплиты Parquet / ORC, создаваемые Hive, Iceberg, нативным Paimon RawFile и табличными функциями на основе файлов, при условии, что File Scanner V2 включён и как минимум два сканера могут обрабатывать создаваемые задачи.

В следующих сценариях этот механизм не применяется и используется прежняя стратегия разбиения на FE, которой управляют file_split_size, max_initial_file_split_size, max_file_split_size и max_file_split_num:

• Форматы, отличные от Parquet / ORC;
• File Scanner V2 отключён;
• max_file_scanners_concurrency = 1, либо BE использует только один сканер для небольшого запроса с LIMIT;
• Сканирование транзакционных таблиц Hive (Full ACID);
• Сканирование, оптимизированное с помощью COUNT по метаданным уровня таблицы;
• Файлы данных Iceberg, с которыми связаны файлы удаления;
• Сериализованные логические / JNI сплиты Paimon;
• Для file_split_size_on_fe или file_split_size_on_be задано неположительное значение.

Примечание

Row Group или Stripe является минимальной неделимой единицей, поэтому фактический размер отдельного сплита может превышать целевой размер, заданный параметром file_split_size_on_be.

Повторное использование задач сканирования внешних таблиц

Поддерживается начиная с версии 4.1.4.

Когда одна и та же внешняя таблица упоминается несколько раз в одном операторе SQL (например, при соединении таблицы с самой собой или при многократном обращении к CTE), эквивалентные узлы сканирования могут повторно использовать один и тот же набор уже созданных сплитов, чтобы FE не выполнял планирование разбиения повторно.

Этим поведением управляет переменная сеанса enable_external_scan_task_reuse:

• Тип: boolean
• По умолчанию: true
• Описание: определяет, могут ли одинаковые узлы сканирования внешней таблицы повторно использовать созданные сплиты в рамках одного оператора. Применяется к Hive, Hudi, Iceberg и Paimon.

Оптимизация с помощью Merge IO

При работе с удалёнными системами хранения, такими как HDFS и объектные хранилища, Doris оптимизирует операции IO с помощью технологии Merge IO. Эта технология объединяет несколько небольших соседних запросов IO в один крупный запрос IO, что позволяет уменьшить IOPS и повысить пропускную способность IO.

Например, если исходный запрос должен прочитать диапазоны [0, 10] и [20, 50] файла file1:

Request Range: [0, 10], [20, 50]

С помощью Merge IO они объединяются в один запрос:

Request Range: [0, 50]

В этом примере два запроса IO объединяются в один, но при этом читаются и дополнительные данные (данные в диапазоне 10-20). Поэтому, сокращая число операций IO, Merge IO может приводить к избыточному чтению данных.

Подробные сведения о работе Merge IO можно просмотреть в Query Profile:

- MergedSmallIO:
    - MergedBytes: 3.00 GB
    - MergedIO: 424
    - RequestBytes: 2.50 GB
    - RequestIO: 65.555K (65555)

Здесь RequestBytes и RequestIO показывают объём данных и количество исходных запросов. MergedBytes и MergedIO показывают объём данных и количество запросов после объединения.

Если MergedBytes значительно больше RequestBytes, это указывает на существенное избыточное чтение данных. Настроить поведение можно с помощью следующих параметров:

• merge_io_read_slice_size_bytes

  Переменная сеанса, поддерживаемая начиная с версии 3.1.3. Значение по умолчанию: 8MB. Если наблюдается значительное избыточное чтение, можно уменьшить этот параметр, например до 64KB, и проверить, улучшаются ли показатели запросов IO и задержка выполнения запроса.

Parquet Page Cache

к сведению

Поддерживается начиная с версии 4.1.0.

Parquet Page Cache представляет собой механизм кеширования файлов Parquet на уровне страниц. Он интегрирован с существующей инфраструктурой Page Cache в Doris и существенно повышает производительность запросов за счёт кеширования распакованных (или сжатых) страниц данных в памяти.

Основные возможности

1. Интеграция с общей инфраструктурой Page Cache

   • Использует ту же базовую инфраструктуру StoragePageCache, что и внутренние таблицы Doris
   • Использует общий пул памяти и общие политики вытеснения
   • Повторно использует существующую статистику кеша и RuntimeProfile для единого мониторинга производительности

2. Интеллектуальная стратегия кеширования

   • Учёт коэффициента сжатия: автоматически определяет, кешировать сжатые или распакованные данные, на основе параметра parquet_page_cache_decompress_threshold
   • Гибкий подход к хранению: кеширует распакованные данные, когда decompressed size / compressed size ≤ threshold; в противном случае решение о кешировании сжатых данных принимается на основе enable_parquet_cache_compressed_pages
   • Структура ключа кеша: использует file_path::mtime::offset в качестве ключа кеша, чтобы обеспечить согласованность кеша после изменения файлов

Параметры конфигурации

Ниже приведены параметры конфигурации BE:

• enable_parquet_page_cache

  Определяет, включён ли Parquet Page Cache. Значение по умолчанию: false.

• parquet_page_cache_decompress_threshold

  Порог, определяющий, кешировать сжатые или распакованные данные. Значение по умолчанию: 1.5. Если отношение decompressed size / compressed size меньше или равно этому порогу, кешируются распакованные данные; в противном случае решение о кешировании сжатых данных принимается на основе настройки enable_parquet_cache_compressed_pages.

• enable_parquet_cache_compressed_pages

  Определяет, кешировать ли сжатые страницы данных, когда коэффициент сжатия превышает порог. Значение по умолчанию: false.

Мониторинг производительности

Статистику использования Parquet Page Cache можно просмотреть в Query Profile:

ParquetPageCache:
    - PageCacheHitCount: 1024
    - PageCacheMissCount: 128

Здесь PageCacheHitCount показывает количество попаданий в кеш, а PageCacheMissCount показывает количество промахов кеша.