Зачем это нужно
In-memory платформа для обработки колоночных данных tech.ml.dataset (TMD) остаётся основным инструментом функциональной работы с данными. Когда данные не помещаются в память, с TMD можно продолжать работать на выборках или фильтровать нужные подмножества под ограничения окружения. Для сохранения данных, как небольших, так и крупных, подходят nippy, arrow или parquet.
Но когда объём данных вырастает до, например, набора .csv-файлов на ~100 ГБ с реляционными связями, привычные инструменты становятся неудобными. Возникает соблазн заняться настройкой Spark-кластера со всеми его нефункциональными сложностями. При этом сохранить транзакционность и простую модель дискового IO всё ещё крайне желательно — локальные диски достаточно большие, а процессоры достаточно быстрые, чтобы не спешить с радикальными решениями.
Реляционные базы данных хорошо подходят для хранения данных, не помещающихся в память, и быстрых реляционных запросов. Но как использовать это преимущество, не отказываясь от функционального программирования и колоночной модели обработки TMD? JDBC вместе с Postgres дают неплохой первый ответ на этот вопрос, но конвертация из строк в столбцы через неэффективный небатчевый API, чтобы прогнать данные через JDBC в TMD, вызывает раздражение.
Появляется новый претендент
DuckDB всплыл в issue на github в мае 2021 года, а к декабрю того же года tmducken получил минимальную интеграцию через их C-биндинги. В той версии все результаты запроса возвращались целиком, и им нужно было помещаться в память. Кроме того, в ранних версиях DuckDB не существовало специализированной высокопроизводительной системы для append или insert, поэтому IO ограничивал потенциальную производительность, и Postgres продолжал служить вспомогательной системой обработки для TMD. С тех пор многое изменилось.
За последние два года DuckDB заметно улучшился. Важно, что C-интерфейс теперь предоставляет батчевую систему как для вставки, так и для запросов, что позволяет обрабатывать очень крупные join — об этом дальше. Эти возможности теперь доступны из Clojure через TMD, открывая доступ к современному векторизованному SQL-движку DuckDB — и результат впечатляет.
Практика
Продолжая тему предыдущего материала, возьмём .csv-файл на 50 гигабайт с тремя годами транзакционных данных — всего 400 000 000 строк:
$ ll -h data.csv
-rw-rw-r-- 1 harold harold 50G Aug 8 09:49 data.csv
Загрузить его в DuckDB удивительно просто — хотя придётся подождать 2 минуты:
$ time duckdb data.ddb 'CREATE TABLE data AS FROM "data.csv";'
100% ▕████████████████████████████████████████████████████████████▏
real 1m50.091s
user 21m42.693s
sys 0m57.887s
$ ll -h data.ddb
-rw-rw-r-- 1 harold harold 18G Sep 6 10:57 data.ddb
Файл сократился до 18 ГБ, включая все индексы (!), созданные DuckDB автоматически.
Данные на месте:
$ duckdb data.ddb
v0.8.1 6536a77232
Enter ".help" for usage hints.
D SELECT COUNT(*) AS n FROM data;
┌───────────┐
│ n │
│ int64 │
├───────────┤
│ 400000000 │
└───────────┘
D DESCRIBE TABLE data;
┌────────────────┬─────────────┬─────────┬─────────┬─────────┬─────────┐
│ column_name │ column_type │ null │ key │ default │ extra │
│ varchar │ varchar │ varchar │ varchar │ varchar │ varchar │
├────────────────┼─────────────┼─────────┼─────────┼─────────┼─────────┤
│ customer-id │ VARCHAR │ YES │ │ │ │
│ day │ BIGINT │ YES │ │ │ │
│ inst │ TIMESTAMP │ YES │ │ │ │
│ month │ BIGINT │ YES │ │ │ │
│ brand │ VARCHAR │ YES │ │ │ │
│ style │ VARCHAR │ YES │ │ │ │
│ sku │ VARCHAR │ YES │ │ │ │
│ year │ BIGINT │ YES │ │ │ │
│ transaction-id │ VARCHAR │ YES │ │ │ │
│ quantity │ BIGINT │ YES │ │ │ │
│ price │ DOUBLE │ YES │ │ │ │
├────────────────┴─────────────┴─────────┴─────────┴─────────┴─────────┤
│ 11 rows 6 columns │
└──────────────────────────────────────────────────────────────────────┘
Обратиться к этой базе из Clojure через TMD не сложнее:
user> (require '[tmducken.duckdb :as duckdb])
nil
user> (require '[tech.v3.dataset :as ds])
nil
user> (duckdb/initialize!)
Sep 06, 2023 11:00:12 AM clojure.tools.logging$eval7454$fn__7457 invoke
INFO: Attempting to load duckdb from "./binaries/libduckdb.so"
true
user> (def db (duckdb/open-db "data.ddb"))
#'user/db
user> (def conn (duckdb/connect db))
#'user/conn
user> (time (duckdb/sql->dataset conn "SELECT COUNT(*) AS n FROM data"))
"Elapsed time: 10.305756 msecs"
:_unnamed [1 1]:
| n |
|----------:|
| 400000000 |
Допустим, менеджмент сообщает, что есть ещё один набор данных — про то, в каких цветах выпускается каждый sku. Его тоже нужно загрузить в базу:
user> (-> (let [colors ["red" "green" "blue" "yellow" "purple" "black" "white"]]
(->> (for [brand (range 100)
style (range 10)
item (range 10)]
(let [sku (format "sku-%s-%s-%s" brand style item)
n (rand-int 8)]
(for [color (take n (shuffle colors))]
{"sku" sku
"color" color})))
(apply concat)))
(ds/->dataset {:dataset-name "colors"}))
colors [35179 2]:
| sku | color |
|------------|--------|
| sku-0-0-0 | red |
| sku-0-0-0 | blue |
| sku-0-0-0 | white |
| sku-0-0-0 | yellow |
| sku-0-0-0 | black |
| sku-0-0-0 | green |
| sku-0-0-1 | black |
| sku-0-0-1 | yellow |
| sku-0-0-1 | blue |
| sku-0-0-1 | purple |
| ... | ... |
| sku-99-9-8 | yellow |
| sku-99-9-8 | purple |
| sku-99-9-8 | black |
| sku-99-9-8 | red |
| sku-99-9-8 | white |
| sku-99-9-8 | blue |
| sku-99-9-8 | green |
| sku-99-9-9 | purple |
| sku-99-9-9 | blue |
| sku-99-9-9 | black |
| sku-99-9-9 | green |
user> (duckdb/create-table! conn *1)
"colors"
user> (duckdb/insert-dataset! conn *2)
35179
Дальше понятно, к чему всё идёт: нужно объединить 400 миллионов транзакций, у каждой из которых есть sku, с таблицей, где на каждый sku приходится в среднем 3,51 цвета.
Первый запрос, к счастью, простой: «Сколько товаров каждого цвета продано в марте 2021?» и, конечно, «Когда узнаем ответ?»
;; First, because you can, join 1.4 billion rows on your laptop in 2.5s...
user> (time (duckdb/sql->dataset conn "SELECT COUNT(*) FROM data INNER JOIN colors ON data.sku = colors.sku;"))
"Elapsed time: 2486.620275 msecs"
:_unnamed [1 1]:
| count_star() |
|-------------:|
| 1416737859 |
;; Then, answer their question...
user> (time (duckdb/sql->dataset conn "SELECT color, COUNT(*) FROM data INNER JOIN colors ON data.sku = colors.sku WHERE data.year='2021' AND data.month='3' GROUP BY color;"))
"Elapsed time: 1077.723309 msecs"
:_unnamed [7 2]:
| color | count_star() |
|--------|-------------:|
| red | 5714223 |
| yellow | 5652010 |
| black | 5720753 |
| blue | 5750846 |
| white | 5689916 |
| green | 5816652 |
| purple | 5671959 |
Ответ известен уже через секунду.
Следующий вопрос может быть не так удобен для SQL, и обработка в Clojure через TMD подойдёт лучше. Пример ниже сворачивает каждую транзакцию по любимому sku начальства, отсортированную по времени (за 1 секунду):
user> (time
(reduce (fn [eax ds]
(conj eax (ds/row-count ds)))
[]
(duckdb/sql->datasets conn "SELECT * FROM data WHERE sku='sku-50-5-5' ORDER BY inst")))
"Elapsed time: 1067.480751 msecs"
[2048 576 2048 576 2048 576 2048 576 2048 576 2048 576 2048 576 2048 576 2048 576 2048 576 2048 576 2048 576 2048 576 2048 576 2048 576 2048 576 733]
Сама функция свёртки тут тривиальна, но принцип понятен: реализованные наборы данных доступны для произвольной обработки через reduce — механизмом, который никогда не исчерпает память.
DuckDB также поддерживает путь запроса без копирования данных (zero copy). Если ни один чанк результата запроса не должен «сбежать» за пределы функции свёртки, машине можно выполнить меньше работы. В примере ниже эта возможность включается опцией {:reduce-type :zero-copy-imm}.
Когда обработку можно выразить в таком виде, это теоретически самый экономный по памяти путь исполнения.
user> (time (let [sql "SELECT * FROM data WHERE sku='sku-50-5-5' ORDER BY inst"
options {:reduce-type :zero-copy-imm}]
(reduce (fn [eax zc-ds]
(conj eax (ds/row-count zc-ds)))
[]
(duckdb/sql->datasets conn sql options))))
"Elapsed time: 1037.480113 msecs"
[2048 531 2048 531 2048 531 2048 531 2048 531 2048 531 2048 531 2048 531 2048 531 2048 531 2048 531 2048 531 2048 531 2048 531 2048 531 1408]
Эти примеры дают представление о мощности, которая сейчас доступна в этой связке инструментов.
Несколько интересных фактов о DuckDB
DuckDB автоматически хранит все числовые данные в minmax-индексах, также известных как BRIN-индексы. Они почти не увеличивают исходный размер данных, но заметно ускоряют выполнение запросов. Для уникальных столбцов и первичных ключей автоматически создаются ART-индексы. Кроме того, пользователи могут по желанию создавать индексы для категориальных столбцов, но за это приходится платить увеличением размера на диске и потенциально более медленными транзакциями.
DuckDB написан на стандартном C++11 и потому достаточно портируем — сборка под Mac M1 появилась быстро, и для любой другой платформы, где потребовалась бы своя специфика, компиляцию и модификацию базы можно было бы выполнить без больших сложностей. Директория src в их кодовой базе на момент написания (сентябрь 2023) насчитывает около 100 000 строк C++ кода:
(base) chrisn@chrisn-lp2:~/dev/cnuernber/duckdb$ cloc src
1612 text files.
1612 unique files.
0 files ignored.
github.com/AlDanial/cloc v 1.90 T=0.83 s (1936.0 files/s, 203773.8 lines/s)
-------------------------------------------------------------------------------
Language files blank comment code
-------------------------------------------------------------------------------
C++ 831 13749 8912 99454
C/C++ Header 679 7754 9903 28287
CMake 101 49 0 1541
Markdown 1 7 0 15
-------------------------------------------------------------------------------
SUM: 1612 21559 18815 129297
-------------------------------------------------------------------------------
DuckDB распространяется по лицензии MIT, разработка ведётся открыто на github, а команда быстро отвечает на вопросы сообщества. Создавать настолько качественный мощный инструмент в такой открытой манере — достойный подход.
В итоге
Попробовать интеграцию с duckdb можно самостоятельно, либо обратиться за помощью к команде TechAscent. DuckDB хорошо дополняет TMD и значительно расширяет возможности небольшой команды по эффективному управлению и обработке крупных наборов данных без необходимости прибегать к дорогим распределённым решениям. Эта интеграция подтверждает ценность качественных и эффективных инструментов вычислений для настоящей демократизации обработки данных: функциональные решения теперь работают на ноутбуках там, где другие, с менее острыми инструментами, потянулись бы за кластером.