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

В этом документе описаны основные методы и стратегии оптимизации запросов к данным в озере данных (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 поддерживается прогрев кеша, позволяющий заранее заполнять кеш данных для повышения производительности запросов.

Распределение фрагментов внешних файлов

В сборках для разработки, поддерживающих распределение фрагментов внешних файлов, повторные запросы к одному и тому же фрагменту удалённого файла могут распределяться между подходящими BE. Имеет смысл включить эту возможность, если такие запросы полностью загружают один BE, а у остальных подходящих BE остаются свободные ресурсы.

-- Enable automatic spreading for the current session.
SET external_scan_consistent_hash_spread_num = 0;

Значение по умолчанию 1 сохраняет исходную схему планирования. При распределении не учитывается текущая загрузка CPU на BE и не гарантируется кратный рост пропускной способности; при этом могут увеличиться число копий в кеше и объём чтения при прогреве. Сравнивайте пропускную способность при одинаковых запросе, состоянии кеша и ресурсах. Ограничения на число кандидатов, правила локальности и команды сброса описаны в разделе external_scan_consistent_hash_spread_num.

Оптимизация чтения из 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	true	Включать ли фильтрацию по выражениям с помощью ZoneMap. File Scanner V2 всегда выполняет безопасную фильтрацию по выражениям с помощью ZoneMap

Оптимизация 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). Таким образом, хотя Merge IO сокращает количество операций 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. Эта возможность интегрирована с существующей в Doris инфраструктурой Page Cache и существенно повышает производительность запросов за счёт кеширования распакованных (или сжатых) страниц данных в памяти.

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

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 показывает количество промахов кеша.