Векторные аналитические СУБД: архитектура DuckDB

В предыдущих темах мы познакомились с языком аналитического SQL и колоночным форматом Parquet. Традиционный подход к аналитической обработке данных требовал развёртывания тяжёлых распределённых кластеров (Apache Spark, Presto/Trino, ClickHouse) или клиент-серверных СУБД. Однако для подавляющего большинства аналитических задач (наборы данных от мегабайтов до сотен гигабайт) администрирование распределённого кластера создаёт неоправданные инфраструктурные затраты.

С 2019 года в индустрии произошёл фундаментальный сдвиг благодаря появлению концепции встраиваемой аналитической СУБД (In-Process OLAP), лидером которой стала система DuckDB, разработанная в Центре математики и информатики в Амстердаме (CWI) Марком Раасвельдтом и Ханнесом Мюлейзеном ¹. DuckDB часто называют «SQLite для аналитики»: система работает как легковесная библиотека прямо в адресном пространстве процесса приложения, не требует отдельного сервера и обеспечивает феноменальную скорость выполнения SQL-запросов ².

В этой теме мы разберём внутреннее устройство современных векторизованных движков на примере архитектуры DuckDB: векторную обработку в памяти, многопоточное планирование Morsel-Driven Parallelism, работу с внешними форматами без промежуточного копирования (Zero-Copy Data Virtualization) и возможности расширенного аналитического SQL.

Векторизованная модель выполнения (Vectorized Execution)

Классические реляционные СУБД (PostgreSQL, Oracle, MySQL) используют модель выполнения Volcano (итераторную модель кортежей, Tuple-at-a-Time), предложенную Гётцем Грефе ³. В этой модели каждый узел дерева плана вызывает метод next() у дочернего узла и получает ровно одну строку:

       [ HashAggregate ]
              ^
              | next() -> строка 1
       [ Filter: val > 100 ]
              ^
              | next() -> строка 1
       [ TableScan ]

При обработке миллионов строк эта модель создаёт огромные накладные расходы:

• миллионы виртуальных вызовов функций (указатели на функции);
• постоянные промахи предсказателя переходов процессора (Branch Misprediction);
• неэффективное использование кэша команд (Instruction Cache) и регистров CPU.

Векторизация: обработка массивов векторов (Vector-at-a-Time)

Векторизованная модель (предложенная в системе MonetDB/X100 Питером Бончем ⁴) кардинально меняет этот принцип: метод next() возвращает не одну строку, а вектор значений (типичный размер вектора в DuckDB — 2048 элементов, помещающихся в кэш данных L1/L2 процессора):

       [ HashAggregate ]
              ^
              | next() -> Вектор [2048 значений]
       [ Filter: val > 100 ]
              ^
              | next() -> Вектор [2048 значений]
       [ TableScan ]

Внутри узла плана операция фильтрации или вычисления превращается в плотный цикл по непрерывному массиву памяти:

// Упрощённая иллюстрация векторизованного фильтра:
for (int i = 0; i < VECTOR_SIZE; i++) {
    selection_vector[count] = i;
    count += (data[i] > 100);
}

Компилятор разворачивает такой цикл с использованием векторных инструкций процессора (SIMD: SSE4.2, AVX2, AVX-512), обрабатывая по 8–16 чисел за один машинный такт.

Параллелизм на основе порций (Morsel-Driven Parallelism)

DuckDB автоматически масштабирует выполнение запроса на все доступные ядра процессора без создания сложных распределённых блокировок. В основе планировщика лежит концепция Morsel-Driven Parallelism, предложенная исследователями Мюнхенского технического университета для СУБД HyPer ⁵.

• Вся обрабатываемая таблица или набор файлов Parquet логически разбивается на небольшие независимые «порции» (morsels) фиксированного размера (обычно около 100 000 строк).
• Создаётся пул рабочих потоков (по числу физических ядер CPU).
• Потоки динамически забирают очередную порцию из глобальной очереди задач, полностью обрабатывают её в локальной памяти ядра (включая построение локальных хеш-таблиц) и объединяют результаты на финальной стадии слияния.
• Если одно ядро процессора временно занято операционной системой, другие ядра продолжают разбирать порции, гарантируя идеальную балансировку нагрузки без простоя.

Виртуализация данных и прямое чтение файлов (Zero-Copy Engine)

Одной из самых сильных сторон DuckDB является способность выполнять полноценный аналитический SQL по внешним файлам и другим СУБД без предварительного импорта:

-- Прямой запрос к CSV и Parquet с объединением на лету:
SELECT d.district_name,
       count(t.incident_id) AS total_incidents,
       avg(p.consumption) AS avg_power
FROM read_parquet('s3://smart-city-lake/telemetry/*.parquet') p
JOIN read_csv('districts.csv') d ON d.district_id = p.district_id
LEFT JOIN read_json('incidents.json') t ON t.sensor_id = p.sensor_id
GROUP BY d.district_name;

DuckDB выступает в роли универсального аналитического процессора:

• автоматический вывод схемы (Schema Inference) для JSON и CSV;
• чтение файлов по маске (*.parquet, разбиение по каталогам year=2024/month=03/*);
• прямое подключение к базам данных SQLite и PostgreSQL через сканеры памяти без промежуточной выгрузки на диск;
• бесшовная интеграция с Python DataFrame (Pandas, Polars, PyArrow) через стандарт Apache Arrow с нулевым копированием.

Продвинутые возможности SQL для анализа телеметрии

DuckDB существенно расширяет стандартный SQL, делая работу с аналитикой лаконичной и выразительной:

1. Предложения COLUMNS() и модификаторы EXCLUDE / REPLACE:

   -- Выбрать все столбцы, исключая громоздкие описания:
   SELECT * EXCLUDE (debug_payload, raw_hex) FROM sensors;
   
   -- Применить агрегатную функцию сразу ко всем числовым столбцам:
   SELECT avg(COLUMNS(* EXCLUDE (sensor_id, recorded_at))) FROM sensors;
   

2. Оконные функции над скользящими интервалами времени:

   -- Расчёт скользящего среднего значения за 3 предшествующих часа:
   SELECT sensor_id, recorded_at, val,
          avg(val) OVER (
              PARTITION BY sensor_id
              ORDER BY recorded_at
              RANGE BETWEEN INTERVAL '3 HOURS' PRECEDING AND CURRENT ROW
          ) AS rolling_avg_3h
   FROM telemetry;
   

3. Конструкция ASOF JOIN (сопоставление по ближайшему времени): При анализе IoT-данных показания разных датчиков приходят несинхронно. ASOF JOIN позволяет соединить запись с ближайшим по времени предыдущим событием без громоздких подзапросов:

   SELECT r.timestamp, r.sensor_id, r.val, a.alert_status
   FROM readings r
   ASOF JOIN alerts a
     ON r.sensor_id = a.sensor_id
    AND r.timestamp >= a.alert_time;
   

Итоги темы

DuckDB объединила в себе удобство автономной библиотеки и архитектурные решения передовых аналитических систем: векторизованное ядро, планировщик Morsel-driven, глубокую интеграцию с колоночными форматами Parquet и Arrow. Это позволяет аналитикам и инженерам данных выполнять сложнейшие исследования сотен гигабайт телеметрии на рабочей станции за считанные секунды, не прибегая к развёртыванию тяжёлых серверных кластеров.

Литература

1. Raasveldt M., Mühleisen H. DuckDB: an Embeddable Analytical Database. — Proceedings of the 2019 ACM SIGMOD International Conference on Management of Data, 2019, С. 1981–1984 ¹.
2. DuckDB Foundation. DuckDB Documentation: In-Process Analytical SQL Engine. — 2024. https://duckdb.org/docs/ ².
3. Graefe G. Volcano — An Extensible and Efficient Query Evaluation System. — IEEE Transactions on Knowledge and Data Engineering, 1994, Vol. 6, No. 1, С. 120–135 ³.
4. Boncz P. A., Zukowski M., Nes N. MonetDB/X100: Hyper-Pipelining Query Execution. — CIDR, 2005, Vol. 5, С. 225–237 ⁴.
5. Leis V., Boncz P., Kemper A., Neumann T. Morsel-Driven Parallelism: A NUMA-Aware Query Evaluation Framework for the Many-Core Age. — Proceedings of the 2014 ACM SIGMOD International Conference on Management of Data, 2014, С. 743–754 ⁵.

Дополнительные материалы

Вопросы для самопроверки

1. Что означает термин «встраиваемая СУБД» (In-Process DBMS) и какие накладные расходы исключаются за счёт отказа от архитектуры клиент-сервер?
2. В чём заключаются недостатки итераторной модели Volcano при обработке миллионов строк?
3. Чем векторизованная модель выполнения (Vector-at-a-Time) отличается от компиляции запросов в машинный код (JIT-компиляция в стиле HyPer)?
4. Как размер вектора (например, 2048 значений) связан с объёмом кэш-памяти L1/L2 центрального процессора?
5. Опишите принцип работы планировщика Morsel-Driven Parallelism. Как решается проблема разной скорости работы процессорных ядер (stragglers)?
6. Что такое виртуализация данных (Data Virtualization) и как DuckDB читает файлы Parquet и CSV напрямую без импорта в собственную базу данных?
7. Как работает предложение ASOF JOIN и почему оно незаменимо при сопоставлении несинхронизированных временных рядов IoT?
8. В чём различие между конструкциями EXCLUDE и REPLACE в предложении SELECT диалекта DuckDB?
9. Как организовано управление оперативной памятью в DuckDB при выполнении запросов, размер данных которых превышает доступную RAM?
10. Какие преимущества даёт интеграция DuckDB с экосистемой Apache Arrow при работе из языка Python?

Глоссарий темы

Шпаргалка по аналитическим возможностям DuckDB

-- Установка ограничений памяти и числа потоков:
SET memory_limit = '4GB';
SET threads TO 4;

-- Профилирование и исследование плана выполнения:
EXPLAIN ANALYZE
SELECT district, count(*)
FROM read_parquet('data/*.parquet')
GROUP BY district;

-- Оконные агрегации с процентным рангом:
SELECT sensor_id, timestamp, val,
       percent_rank() OVER (PARTITION BY sensor_id ORDER BY val) AS pct_rank
FROM telemetry;

-- Поворот данных (Pivot) для построения сводных таблиц:
PIVOT telemetry
ON sensor_type
USING avg(val)
GROUP BY date_trunc('day', timestamp);

Типичные ошибки и подводные камни

1. Неконтролируемое чтение сотен неструктурированных файлов через glob: Выражение read_parquet('**/*.parquet') при сканировании миллиона файлов на сетевом хранилище вызовет исчерпание дескрипторов файлов или тайм-аут метаданных. Рекомендуется использовать партиционирование каталогов (Hive partitioning) и фильтровать каталоги на уровне путей.
2. Неявное переполнение оперативной памяти при агрегациях с большим числом уникальных групп: Если группировка GROUP BY производится по столбцам высокой кардинальности (например, сырые UUID или высокоточные временные метки), размер хеш-таблицы может превысить memory_limit, что вызовет деградацию скорости из-за сброса на диск.
3. Игнорирование различий между локальным временем и UTC в ASOF JOIN: Если в одной таблице метки времени сохранены в TIMESTAMP WITH TIME ZONE, а в другой — в TIMESTAMP WITHOUT TIME ZONE, сопоставление по ближайшему времени вернёт некорректные пары событий со сдвигом в несколько часов.

Литература

1. Raasveldt M., Mühleisen H. DuckDB: an Embeddable Analytical Database. — ACM SIGMOD, 2019 ¹.
2. Leis V. et al. Morsel-Driven Parallelism: A NUMA-Aware Query Evaluation Framework for the Many-Core Age. — ACM SIGMOD, 2014 ².
3. DuckDB Foundation. DuckDB SQL Documentation. — 2024 ³.

Лабораторная работа 4. Аналитика телеметрии в DuckDB: оконные функции, ASOF JOIN и виртуализация данных

• Раздел курса: тема 4 «Векторные аналитические СУБД: архитектура DuckDB»

Введение

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

В этой лабораторной работе вы освоите аналитический инструментарий СУБД DuckDB: научитесь напрямую опрашивать наборы файлов Parquet без импорта, освоите расчет скользящих статистик над временными окнами, примените соединение по ближайшему времени ASOF JOIN для связывания показаний с журналом инцидентов и экспортируете готовые витрины данных.

Цель работы

Освоить методы высокопроизводительного анализа телеметрических данных в DuckDB с использованием прямого чтения файлов Parquet, расширенного аналитического SQL (оконные функции, фильтрация скользящих интервалов) и временных соединений ASOF JOIN.

После выполнения работы вы сможете:

• выполнять аналитические запросы к наборам файлов Parquet с использованием шаблонов путей и автоматического определения партиций;
• рассчитывать скользящие средние, медианы и процентили телеметрии с помощью оконных предложений OVER (RANGE BETWEEN ...);
• соединять несинхронизированные временные ряды датчиков и аварийных сигналов с помощью оператора ASOF JOIN;
• строить компактные аналитические витрины данных с сохранением в оптимизированный колоночный формат.

Оборудование и учебная среда

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

• клиент DuckDB CLI (duckdb) либо окружение Python 3 с пакетом duckdb;
• учебный набор данных: каталог telemetry_parquet/ с данными телеметрии и файл incidents.json с журналом аварийных событий.

Порядок выполнения работы

Часть 1. Виртуализация данных и сканирование партиционированных каталогов

1. Запустите сессию DuckDB.
2. Проверьте структуру каталога телеметрии: файлы распределены по дням в структуре Hive: telemetry/date=2024-03-01/data.parquet, telemetry/date=2024-03-02/data.parquet.
3. Выполните запрос к набору файлов с помощью шаблонного сканирования:

   SELECT date, count(*) AS total_points,
          min(val) AS min_val, max(val) AS max_val
   FROM read_parquet('telemetry/*/*.parquet', hive_partitioning = true)
   GROUP BY date
   ORDER BY date;
   

4. Убедитесь с помощью EXPLAIN, что фильтрация по столбцу date = '2024-03-01' отсекает чтение файлов других дат (Partition Pruning).

Часть 2. Скользящие оконные агрегации и выявление аномалий

1. Для датчика давления с идентификатором sensor_id = 'P-101' постройте временной ряд со скользящим средним значением за 1 час (окно в 60 минут) и стандартным отклонением:

   SELECT timestamp, val,
          avg(val) OVER w AS rolling_mean,
          stddev(val) OVER w AS rolling_std
   FROM read_parquet('telemetry/*/*.parquet')
   WHERE sensor_id = 'P-101'
   WINDOW w AS (
       ORDER BY timestamp
       RANGE BETWEEN INTERVAL '30 MINUTES' PRECEDING AND INTERVAL '30 MINUTES' FOLLOWING
   )
   ORDER BY timestamp;
   

2. Рассчитайте Z-оценку (Z = \frac{\text{val} - \text{rolling\_mean}}{\text{rolling\_std}}) и отфильтруйте точки, где |Z| > 3 (аномальные скачки давления).

Часть 3. Временное сопоставление через ASOF JOIN

1. Изучите файл инцидентов incidents.json. В нём содержатся отметки времени фиксации аварийных ситуаций:

   SELECT * FROM read_json_auto('incidents.json') LIMIT 5;
   

2. Сопоставьте каждый инцидент с последним известным показанием датчика до момента аварии с помощью ASOF JOIN:

   SELECT inc.incident_id,
          inc.incident_time,
          inc.description,
          tel.sensor_id,
          tel.timestamp AS sensor_timestamp,
          tel.val AS last_known_value
   FROM read_json_auto('incidents.json') inc
   ASOF JOIN read_parquet('telemetry/*/*.parquet') tel
     ON inc.sensor_id = tel.sensor_id
    AND inc.incident_time >= tel.timestamp
   ORDER BY inc.incident_time;
   

3. Сравните этот запрос с классическим эквивалентом на подзапросах или коррелированных соединениях по объёму кода и времени выполнения.

Часть 4. Создание сводной аналитической витрины

1. Сформируйте итоговую витрину суточных показателей по всем районам города, применив оператор PIVOT:

   CREATE TABLE daily_summary AS
   PIVOT (
       SELECT district, sensor_type, avg(val) AS avg_reading
       FROM read_parquet('telemetry/*/*.parquet')
       GROUP BY district, sensor_type
   )
   ON sensor_type
   USING first(avg_reading);
   

2. Экспортируйте полученную таблицу в единый файл Parquet со сжатием Zstandard:

   COPY daily_summary TO 'daily_summary.parquet' (FORMAT PARQUET, COMPRESSION ZSTD);
   

Журнал работы

Подготовьте отчет о выполнении лабораторной работы, содержащий:

1. SQL-запрос и вывод агрегации по партиционированному набору файлов Parquet.
2. Результаты расчёта скользящего среднего и выявленные аномальные точки телеметрии (|Z| > 3).
3. Сравнение производительности и синтаксиса ASOF JOIN против традиционного соединения через подзапрос с MAX(timestamp).
4. Схему полученной аналитической витрины daily_summary.parquet.