Сброс промежуточных результатов запроса на диск

Зачем нужен сброс на диск

Вычислительный слой Doris основан на архитектуре MPP. Все вычислительные задачи выполняются в памяти BE, а обмен данными между узлами также использует память, поэтому управление памятью напрямую влияет на стабильность запросов. По мере того как все больше пользователей переносят в Doris обработку данных ETL, обработку материализованных представлений по нескольким таблицам и сложные запросы ad-hoc, памяти одного узла зачастую не хватает для хранения всего промежуточного состояния.

Сброс на диск (spill to disk) записывает на диск промежуточное состояние, например состояние агрегации и временные данные сортировки, позволяя продолжать выполнение запросов при нехватке памяти. Он дает три преимущества:

Преимущество	Описание
Масштабируемость	Позволяет обрабатывать наборы данных, превышающие лимит памяти одного узла
Стабильность	Снижает число ошибок запросов и аварийных завершений процессов из-за нехватки памяти
Гибкость	Позволяет выполнять более сложные запросы без добавления оборудования

Сброс на диск сейчас поддерживают следующие операторы: Hash Join, Aggregation, Sort и CTE.

Примечание

Сброс на диск создает дополнительную нагрузку на дисковый ввод-вывод, поэтому время выполнения запросов может существенно увеличиться. Соответственно увеличьте значение переменной сеанса query_timeout и подключите отдельный диск или используйте SSD для каталога сброса, чтобы снизить влияние на обычную загрузку данных и выполнение запросов.

По умолчанию сброс промежуточных результатов запросов на диск отключен.

Как запускается сброс на диск

Doris использует механизм резервирования памяти (reserve memory), чтобы управлять запуском сброса на диск. Он работает следующим образом:

1. Во время выполнения Doris оценивает объем памяти, необходимый для обработки каждого Block, и запрашивает его у единого диспетчера памяти.
2. Глобальный аллокатор памяти проверяет, не приведет ли выделение запрошенной памяти к превышению лимита памяти запроса, Workload Group или процесса.
3. При превышении лимита память не выделяется. Doris приостанавливает текущий запрос и запускает сброс на диск для оператора, потребляющего больше всего памяти.
4. После завершения сброса на диск выполнение запроса продолжается.

Иерархия управления памятью

Управление памятью в Doris имеет три уровня: уровень процесса -> уровень Workload Group -> уровень запроса. Все три уровня ограничивают работу механизма сброса на диск.

Память на уровне процесса (BE)

Параметр mem_limit в be.conf задает верхний предел использования памяти всем процессом BE. Когда потребление памяти превышает этот порог, Doris отменяет запрос, который в данный момент запрашивает память, а асинхронная фоновая задача завершает некоторые запросы или освобождает кеши.

Две распространенные проблемные ситуации:

• Совместное развертывание: если BE размещается на одном хосте с FE, Kafka, HDFS или другими процессами, фактически доступный объем памяти может быть значительно меньше mem_limit. В этом случае механизм освобождения памяти не справляется и срабатывает OOM Killer операционной системы.
• Развертывание в контейнерах: в средах K8s или Cgroup Doris автоматически определяет настройки памяти контейнера, поэтому ручная корректировка не требуется.

Память Workload Group

Параметр	Описание
max_memory_percent	Максимальная доля памяти процесса, которую может использовать эта Workload Group. При ее превышении запускается сброс на диск или завершаются запросы.
min_memory_percent	Минимальная доля памяти, гарантированная этой Workload Group. При нехватке памяти система распределяет ее с учетом этого значения, чтобы другим группам хватало памяти.
memory_low_watermark	Нижний порог использования памяти. По умолчанию 50%.
memory_high_watermark	Верхний порог использования памяти. По умолчанию 80%. При превышении этого значения запросы на резервирование памяти отклоняются и запускается сброс на диск.

Ограничения: сумма значений min_memory_percent всех Workload Group не должна превышать 100%, а значение min_memory_percent отдельной группы не должно быть больше ее max_memory_percent.

Память на уровне запроса

Статическое выделение памяти

Значение exec_mem_limit задается с помощью переменной сеанса до запуска запроса и не может динамически изменяться во время его выполнения.

Примечание об обновлении

Значение exec_mem_limit по умолчанию составляло 2 GB до версии 3.1, а начиная с версии 3.1 было изменено на 100 GB; в этих версиях ограничение фактически применяется на стороне BE. Перед обновлением до версии 3.1 или более поздней явно задайте для этого параметра значение 100g, чтобы существующие запросы не отменялись из-за превышения лимита и не запускали неожиданный сброс на диск.

Динамическое выделение памяти на основе слотов

При статическом выделении пользователям часто сложно точно оценить объем памяти, необходимый одному запросу, поэтому они задают слишком высокий лимит (например, половину памяти процесса). Это мешает точно управлять распределением памяти. Механизм слотов Workload Group решает эту проблему.

Принцип работы:

• Если для Workload Group заданы max_memory_percent и max_concurrency, память BE логически делится на max_concurrency слотов, а объем памяти каждого слота = max_memory_percent x mem_limit / max_concurrency.
• По умолчанию каждый запрос занимает 1 слот. Чтобы использовать больше памяти, измените переменную сеанса query_slot_count.
• Когда запрос занимает больше слотов, число запросов, которые Workload Group может выполнять одновременно, автоматически уменьшается, а новые запросы попадают в очередь.

Возможные значения slot_memory_policy:

Значение	Описание
none	Значение по умолчанию. Механизм не включен. Запрос использует максимально возможный объем памяти, а сброс на диск запускается при достижении лимита Workload Group.
fixed	Объем памяти на запрос = workload group mem_limit x query_slot_count / max_concurrency. Фиксированное выделение памяти в зависимости от допустимого числа одновременных запросов.
dynamic	Объем памяти на запрос = workload group mem_limit x query_slot_count / sum(running query slots). Память свободных слотов динамически перераспределяется в пользу выполняющихся крупных запросов.

Оба варианта, fixed и dynamic, задают жесткие лимиты. При их превышении запускается сброс на диск или завершается запрос. Они также переопределяют статически заданный лимит exec_mem_limit. При настройке slot_memory_policy внимательно подбирайте значение max_concurrency, иначе возможна нехватка памяти.

Включение сброса промежуточных результатов запросов на диск

Шаг 1: Настройте пути сброса на диск для BE

Добавьте следующую конфигурацию в be.conf. Чтобы изменение вступило в силу, BE необходимо перезапустить:

spill_storage_root_path=/mnt/disk1/spilltest/doris/be/storage;/mnt/disk2/doris-spill;/mnt/disk3/doris-spill
spill_storage_limit=100%

Параметр	Описание
spill_storage_root_path	Путь хранения файлов сброса на диск. По умолчанию совпадает со значением storage_root_path. Рекомендуется использовать путь на отдельном диске.
spill_storage_limit	Максимальный объем дискового пространства для файлов сброса. Поддерживается абсолютное значение (например, 100G или 1T) либо процент (по умолчанию 20%). При использовании отдельного диска можно задать 100%.

Шаг 2: Настройте переменные сеанса FE

SET enable_spill = true;
SET exec_mem_limit = 10g;
SET query_timeout = 3600;

Переменная	Описание
enable_spill	Включает или отключает сброс на диск. По умолчанию false. Если включен, сброс на диск запускается автоматически при нехватке памяти.
exec_mem_limit	Максимальный объем памяти, доступный одному запросу.
query_timeout	Сброс на диск увеличивает время выполнения запроса, поэтому соответственно увеличьте время ожидания (в секундах).

Шаг 3: Настройте Workload Group (необязательно)

Скорректируйте max_memory_percent, чтобы отдельная Workload Group не исчерпывала всю память процесса:

ALTER WORKLOAD GROUP normal PROPERTIES ('max_memory_percent'='90%');

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

ALTER WORKLOAD GROUP normal PROPERTIES ('slot_memory_policy'='dynamic');

Мониторинг сброса на диск

Журнал аудита

В FE Audit Log добавлены следующие поля для регистрации объемов чтения и записи при сбросе на диск:

SpillWriteBytesToLocalStorage=503412182|SpillReadBytesFromLocalStorage=503412182

Поле	Описание
SpillWriteBytesToLocalStorage	Общий объем данных, записанных на диск при сбросе (в байтах).
SpillReadBytesFromLocalStorage	Общий объем данных, прочитанных с диска при сбросе (в байтах).

Профиль запроса (Query Profile)

После запуска сброса на диск для запроса в Profile появляются счетчики с префиксом Spill. Пример для HashJoin Build HashTable:

PARTITIONED_HASH_JOIN_SINK_OPERATOR  (id=4  ,  nereids_id=179):(ExecTime:  6sec351ms)
      -  Spilled:  true
      -  CloseTime:  528ns
      -  ExecTime:  6sec351ms
      -  InitTime:  5.751us
      -  InputRows:  6.001215M  (6001215)
      -  MemoryUsage:  0.00  
      -  MemoryUsagePeak:  554.42  MB
      -  MemoryUsageReserved:  1024.00  KB
      -  OpenTime:  2.267ms
      -  PendingFinishDependency:  0ns
      -  SpillBuildTime:  2sec437ms
      -  SpillInMemRow:  0
      -  SpillMaxRowsOfPartition:  68.569K  (68569)
      -  SpillMinRowsOfPartition:  67.455K  (67455)
      -  SpillPartitionShuffleTime:  836.302ms
      -  SpillPartitionTime:  131.839ms
      -  SpillTotalTime:  5sec563ms
      -  SpillWriteBlockBytes:  714.13  MB
      -  SpillWriteBlockCount:  1.344K  (1344)
      -  SpillWriteFileBytes:  244.40  MB
      -  SpillWriteFileTime:  350.754ms
      -  SpillWriteFileTotalCount:  32
      -  SpillWriteRows:  6.001215M  (6001215)
      -  SpillWriteSerializeBlockTime:  4sec378ms
      -  SpillWriteTaskCount:  417
      -  SpillWriteTaskWaitInQueueCount:  0
      -  SpillWriteTaskWaitInQueueTime:  8.731ms
      -  SpillWriteTime:  5sec549ms

Spilled: true означает, что оператор запустил сброс на диск.

Системная таблица backend_active_tasks

В information_schema.backend_active_tasks добавлены два столбца, позволяющие в реальном времени просматривать объемы сброса на диск для выполняющихся запросов:

Столбец	Описание
SPILL_WRITE_BYTES_TO_LOCAL_STORAGE	Объем данных, уже записанных на диск текущим запросом при сбросе (в байтах).
SPILL_READ_BYTES_FROM_LOCAL_STORAGE	Объем данных, уже прочитанных с диска текущим запросом при сбросе (в байтах).

SELECT * FROM information_schema.backend_active_tasks;

Пример вывода:

+-------+------------+-------------------+-----------------------------------+--------------+------------------+-----------+------------+----------------------+---------------------------+--------------------+-------------------+------------+------------------------------------+-------------------------------------+
| BE_ID | FE_HOST    | WORKLOAD_GROUP_ID | QUERY_ID                          | TASK_TIME_MS | TASK_CPU_TIME_MS | SCAN_ROWS | SCAN_BYTES | BE_PEAK_MEMORY_BYTES | CURRENT_USED_MEMORY_BYTES | SHUFFLE_SEND_BYTES | SHUFFLE_SEND_ROWS | QUERY_TYPE | SPILL_WRITE_BYTES_TO_LOCAL_STORAGE | SPILL_READ_BYTES_FROM_LOCAL_STORAGE |
+-------+------------+-------------------+-----------------------------------+--------------+------------------+-----------+------------+----------------------+---------------------------+--------------------+-------------------+------------+------------------------------------+-------------------------------------+
| 10009 | 10.16.10.8 |                 1 | 6f08c74afbd44fff-9af951270933842d |        13612 |            11025 |  12002430 | 1960955904 |            733243057 |                  70113260 |                  0 |                 0 | SELECT     |                          508110119 |                            26383070 |
| 10009 | 10.16.10.8 |                 1 | 871d643b87bf447b-865eb799403bec96 |            0 |                0 |         0 |          0 |                    0 |                         0 |                  0 |                 0 | SELECT     |                                  0 |                                   0 |
+-------+------------+-------------------+-----------------------------------+--------------+------------------+-----------+------------+----------------------+---------------------------+--------------------+-------------------+------------+------------------------------------+-------------------------------------+

Показатели производительности (TPC-DS 10TB)

Ниже приведены результаты теста с выполнением одного запроса за раз на серверах Alibaba Cloud. Он подтверждает, что сброс на диск позволяет завершить все 99 запросов TPC-DS даже в экстремальных условиях, когда отношение объема памяти к объему данных составляет примерно 1:52.

Тестовая среда:

• 1 FE: 16 ядер vCPU, 32 GiB памяти (ecs.c6.4xlarge)
• 3 BE: 16 ядер vCPU, 64 GiB памяти (ecs.g6.4xlarge)
• Тестовые данные: TPC-DS 10TB, подключенные через Alibaba Cloud DLF Catalog

Общее время: 28,102.386 секунды

Запрос	Время (мс)	Запрос	Время (мс)	Запрос	Время (мс)
query1	29092	query34	84055	query67	3939554
query2	130003	query35	69885	query68	183648
query3	96119	query36	148662	query69	11031
query4	1199097	query37	21598	query70	137901
query5	212719	query38	164746	query71	166454
query6	62259	query39	5874	query72	2859001
query7	209154	query40	51602	query73	92015
query8	62433	query41	563	query74	336694
query9	579371	query42	93005	query75	838989
query10	54260	query43	67769	query76	174235
query11	560169	query44	79527	query77	174525
query12	26084	query45	26575	query78	1956786
query13	228756	query46	134991	query79	162259
query14	1137097	query47	161873	query80	602088
query15	27509	query48	153657	query81	16184
query16	84806	query49	259387	query82	56292
query17	288164	query50	141421	query83	26211
query18	94770	query51	158056	query84	11906
query19	124955	query52	91392	query85	57739
query20	30970	query53	89497	query86	34350
query21	4333	query54	124118	query87	173631
query22	9890	query55	82584	query88	449003
query23	1757755	query56	152110	query89	113799
query24	399553	query57	83417	query90	30825
query25	291474	query58	259580	query91	12239
query26	79832	query59	177125	query92	26695
query27	175894	query60	161729	query93	275828
query28	647497	query61	258058	query94	56464
query29	1299597	query62	39619	query95	64932
query30	11434	query63	91258	query96	48102
query31	106665	query64	234882	query97	597371
query32	33481	query65	278610	query98	112399
query33	146101	query66	90246	query99	64472

В дальнейшем поддержка сброса на диск будет добавлена для других операторов, например Window Function и Intersect, а его производительность продолжит улучшаться.