\nДля этого используется механизм S3 SSE-C (Server-Side Encryption with Customer-provided keys). Trino генерирует уникальный случайный AES-ключ для каждого запроса, отправляет его в MinIO вместе с данными, а клиенту (вашему Jupyter) отдает ссылку + этот же ключ для расшифровки.
\nЕсли мы используем локальный MinIO по адресу http:\/\/minio:9000<\/a> (без SSL\/TLS), сервер MinIO видит, что ему пытаются передать секретный пароль (SSE-C ключ) по открытому незащищенному HTTP-каналу.
\nMinIO (как и настоящий AWS S3) строго запрещает это по спецификации. Он возвращает HTTP 400 Bad Request с ошибкой: “Requests specifying Server Side Encryption... must be made over a secure connection”. Поэтому тестировать лучше на реальном s3. И еще<\/p>\n
Мгновенное удаление (Сборка мусора)<\/p>\n
Главное правило Client Spooling: Trino удаляет файлы сразу же, как только они были прочитаны клиентом.
\nКак только ваш Python-скрипт или Jupyter получает ссылку на файл, скачивает его и отправляет координатору Trino HTTP-сигнал (ACK), что кусок получен, координатор дает команду немедленно удалить этот объект из S3.
\nЕсли запрос отменен или упал с ошибкой, Trino тоже моментально зачищает за собой fs.location. Вы просто не успеете их там увидеть.<\/p>\n
Данных слишком мало (Thresholds)<\/p>\n
Писать 10 строк в S3, генерировать для них Pre-signed URLs и отдавать клиенту — это дольше, чем просто плюнуть эти 10 строк текстом через координатор. Trino использует эвристику: если Result Set маленький, он отдается “инлайн” (внутри JSON-ответа самого координатора), и S3 не задействуется.<\/p>\n
В этой статье мы разберем, как передавать результаты запросов через промежуточное S3-хранилище, на примере движков Trino и Apache DataFusion.<\/p>\n
Физика проблемы и математика Spooling<\/h4>\n
В классической архитектуре все воркеры кластера отправляют вычисленные строки на главный узел (Coordinator), а тот уже отдает их по одному каналу клиенту.<\/p>\n
Если D — это объем результирующей выборки, а B c — пропускная способность сети координатора, то время выгрузки данных клиенту без спулинга равно:<\/p>\n
T classic = B \/ Dc<\/p>\n
В режиме Spooling<\/b> координатор не гоняет данные через себя. Воркеры напрямую, параллельно пишут куски результата в дешевое объектное хранилище (S3\/MinIO). Клиент получает лишь ссылки на эти файлы и скачивает их напрямую. Если у нас N файлов в S3, доступных для многопоточного скачивания с пропускной способностью клиента B client: T spooling ≈ min(N×B s3,B client)D<\/p>\n
Это позволяет ускорить выгрузку в десятки раз, так как $B_{client}$ и распределенный $B_{s3}$ обычно значительно больше ограничений одного координатора.<\/p>\n
\n
Подготовка минимальной инфраструктуры<\/h4>\n
Для демонстрации двух подходов мы убрали из нашего кластера все тяжелые клиентские среды (Jupyter, Spark) и оставили только “голое” ядро: хранилище S3, REST-каталог и SQL-движок.<\/p>\n
<\/b><\/summary><\/p>\n использует динамическое подключение к Lakekeeper REST API. Скрипт записывает файлы в формате Parquet в MinIO:<\/p>\n config.properties<\/b><\/p>\n spooling-manager.properties<\/b><\/p>\n -- 1. Подключение каталога Iceberg<\/p>\n -- 2. Создание структуры<\/p>\n -- 3. Запись данных<\/p>\n Если написать Select – должно быть как-то так<\/p>\n Trino поддерживает протокол *Client Spooling* “из коробки” — когда Python-клиент запрашивает огромный `SELECT`, Trino сам незаметно пишет куски в S3 и отдает клиенту готовые ссылки.<\/p>\n В Apache DataFusion<\/b> (который часто работает как локальный движок `datafusion-cli` или встраиваемая библиотка поверх S3) применяется более прозрачный паттерн делегирования (Explicit Spooling). Мы вручную инструктируем движок сохранить результаты агрегации в распределенное хранилище, чтобы позже забрать их в удобном формате — например, упаковав их в `JSON` и сжав алгоритмом `ZSTD`.<\/p>\n Запускаем `datafusion-cli`, передав доступы как переменные среды (для предотвращения ошибок парсинга опций):<\/p>\n Внутри консоли подключаем директорию с Parquet-файлами, сгенерированными Trino:<\/p>\n Вместо того чтобы тянуть миллионы строк на локальный терминал, мы просим DataFusion выполнить преобразование и записать итог запроса обратно в MinIO.<\/p>\n Мы выбираем построчный JSON с экстремальным сжатием:<\/p>\n Результат:<\/b><\/p>\n За миллисекунды (0.270 sec) DataFusion прочитал партиции, трансформировал бинарные столбцы в текст и сжал его.<\/p>\n Описанный паттерн выполнения команды `COPY TO` с сохранением `.json.zst` в MinIO полностью воспроизводит механику Spooling:<\/p>\n Еще немного про Fault-Tolerant Execution (FTE), нужно провести важную границу между архитектурой Trino<\/b> (готовый распределенный кластер) и архитектурой DataFusion<\/b> (ядро\/библиотека выполнения запросов).<\/p>\n В самом “голом” ядре DataFusion (которое вы запускаете в `datafusion-cli` или в Jupyter) нет встроенного механизма Task Retries<\/b>, потому что процессы выполняются на одной машине в рамках одного приложения. Если сервер падает — запрос прерывается.<\/p>\n Однако, в экосистеме DataFusion есть механизмы отказоустойчивости, которые делятся на два уровня: локальный (Spilling)<\/b> и распределенный (Apache Ballista \/ Ray)<\/b>.<\/p>\n В Trino частой причиной падения задач является нехватка памяти (Out of Memory). В DataFusion реализован мощный механизм управления памятью.<\/p>\n Если DataFusion понимает, что оперативной памяти для агрегации или JOIN’а не хватает, он не “роняет” задачу, а начинает сбрасывать промежуточные данные на диск (Spill to Disk<\/b>).<\/p>\n Trino использует архитектуру Fault-Tolerant Execution (FTE)<\/b>, при которой промежуточные результаты (Shuffle Exchange) пишутся в S3, а упавшие воркеры заменяются, и их задачи (Tasks) перезапускаются координатором.<\/p>\n В мире DataFusion эту задачу решает не само ядро, а распределенные планировщики<\/b>, построенные поверх него:<\/p>\n Ballista — это надстройка над DataFusion, превращающая его в полноценный кластер (с Coordinator и Executors), архитектурно очень похожая на Apache Spark и Trino.<\/p>\n Сейчас огромную популярность набирает запуск DataFusion поверх кластера Ray<\/b>. DuckDB завоевал огромную популярность как “SQLite для аналитики”. Это невероятно быстрый, встраиваемый, колоночный движок, который не требует отдельного сервера. Однако его мощь по-настоящему раскрывается, когда он получает доступ к данным эффективно. Просто натравить DuckDB на петабайтный дата-лейк без подготовки — это рецепт для медленных запросов и высоких затрат.<\/p>\n Как же построить мост между огромным хранилищем данных и молниеносной интерактивной аналитикой, которую обещает DuckDB?<\/p>\n В этой статье рассмотрим три фундаментальных архитектурных подхода к организации доступа к данным для DuckDB. Но прежде чем мы погрузимся в то, как *читать* данные, давайте поговорим о том, как их *готовить*.<\/p>\n Данные в вашем Lakehouse не появляются из ниоткуда. Они поступают из операционных баз данных, потоков событий (Kafka), логов и десятков других источников. Прежде чем DuckDB сможет их эффективно запросить, эти данные нужно собрать, очистить, трансформировать и, что самое важное, организовать в надежный и производительный формат.<\/p>\n Здесь на сцену выходит Trino<\/b> (ранее известный как PrestoSQL).<\/p>\n Что такое Trino?<\/b> Это мощный распределенный SQL-движок, созданный для выполнения запросов к гетерогенным источникам данных. Его суперсила — способность “на лету” объединять данные из PostgreSQL, Kafka, Hive, MySQL и многих других систем.<\/p>\n Роль Trino в Lakehouse:<\/b> В современной архитектуре Trino часто выступает в роли “фабрики данных”. Он выполняет тяжелую работу по ETL\/ELT<\/b> (Extract, Transform, Load), подготавливая данные для аналитических инструментов вроде DuckDB.<\/p>\n Типичный сценарий использования:<\/b><\/p>\n Как отмечается в одном из руководств, именно такой `INSERT INTO ... SELECT ...` является типичным способом перемещения данных в Iceberg с помощью Trino.<\/p>\n Итог:<\/b> Trino работает “глубоко в машинном отделении” вашего Lakehouse. Он берет на себя тяжелые, распределенные задачи по преобразованию данных, а DuckDB получает на вход уже чистые, структурированные и оптимизированные для чтения таблицы Iceberg.<\/p>\n Теперь, когда данные готовы, давайте рассмотрим, как их лучше всего потреблять.<\/p>\n Это самый продвинутый и рекомендуемый подход для серьезной аналитики, особенно в serverless-архитектуре.<\/p>\n Иногда вам не нужна вся мощь Iceberg, а нужно просто эффективно выполнить запрос на удаленном экземпляре DuckDB и получить результат.<\/p>\n Что, если вам не нужна вся гибкость SQL, а нужен стандартный REST или GraphQL API поверх ваших данных без строчки кода? Здесь на сцену выходит ROAPI<\/b>.<\/p>\n Еще вот тут можно почитать: https:\/\/duckdb.org\/docs\/stable\/guides\/performance\/how_to_tune_workloads<\/a><\/p>\n Выбор архитектуры зависит от вашей задачи. Каждая из них занимает свою нишу в стеке современной инженерии данных.<\/p>\ndocker-compose.yml<\/code><\/pre>version: '3.8'\n\nservices:\n minio:\n image: minio\/minio:latest\n ports:\n - "19000:9000"\n - "19001:9001"\n environment:\n MINIO_ROOT_USER: "minio-root-user"\n MINIO_ROOT_PASSWORD: "minio-root-password"\n command: server \/data --console-address ":9001"\n\n minio-setup:\n image: minio\/mc:latest\n depends_on:\n - minio\n entrypoint: >\n \/bin\/sh -c "\n sleep 5;\n mc alias set myminio http:\/\/minio:9000 minio-root-user minio-root-password;\n mc mb myminio\/warehouse || true;\n "\n\n lakekeeper:\n image: dalongrong\/lakekeeper:latest\n ports:\n - "8181:8181"\n environment:\n - S3_ENDPOINT=http:\/\/minio:9000\n - S3_REGION=us-east-1\n - S3_ACCESS_KEY_ID=minio-root-user\n - S3_SECRET_ACCESS_KEY=minio-root-password\n depends_on:\n - minio-setup\n\n trino:\n image: trinodb\/trino:latest\n ports:\n - "8080:8080"<\/code><\/pre>
\n
\nСначала мы генерируем данные в Trino. Запрос<\/p>\nCREATE CATALOG<\/code><\/pre>protocol.spooling.enabled=true\n# 256-битный ключ в формате base64. Вы можете сгенерировать свой с помощью команды `openssl rand -base64 32`\nprotocol.spooling.shared-secret-key=jxTKysfCBuMZtFqUf8UJDQ1w9ez8rynEJsJqgJf66u0=\n\ncatalog.management=dynamic<\/code><\/pre>spooling-manager.name=filesystem\n# Включаем чтение\/запись в S3 для Spooling\nfs.s3.enabled=true\n# Путь внутри MinIO (указываем через s3:\/\/)\nfs.location=s3:\/\/warehouse\/client-spooling\/\n\n# Системные настройки S3 (MinIO)\ns3.endpoint=http:\/\/minio:9000\ns3.region=us-east-1\ns3.aws-access-key=minio-root-user\ns3.aws-secret-key=minio-root-password\ns3.path-style-access=true<\/code><\/pre>CREATE CATALOG test_warehouse USING iceberg\nWITH (\n "iceberg.catalog.type" = 'rest',\n "iceberg.rest-catalog.uri" = 'http:\/\/lakekeeper:8181\/catalog\/',\n "iceberg.rest-catalog.warehouse" = '00000000-0000-0000-0000-000000000000\/test_warehouse',\n "iceberg.rest-catalog.security" = 'OAUTH2',\n "iceberg.rest-catalog.nested-namespace-enabled" = 'true',\n "iceberg.rest-catalog.vended-credentials-enabled" = 'true',\n "fs.native-s3.enabled" = 'true',\n "s3.region" = 'us-east-1',\n "s3.path-style-access" = 'true',\n "s3.endpoint" = 'http:\/\/minio:9000'\n);<\/code><\/pre>CREATE SCHEMA test_warehouse.test_schema;\n\nCREATE TABLE test_warehouse.test_schema.my_table (\n id BIGINT,\n data VARCHAR\n) WITH (format = 'PARQUET');<\/code><\/pre>INSERT INTO test_warehouse.test_schema.my_table VALUES (1, 'hello'), (2, 'world');<\/code><\/pre>
\n![\"\"](\"https:\/\/gavrilov.info\/pictures\/Snimok-ekrana-2026-04-12-v-18.25.52.png\")
\n<\/div>\nАналог Spooling в Apache DataFusion (Через экспорт)<\/h4>\n
1. Подключение к S3 и маппинг исходной таблицы<\/h5>\n
AWS_ACCESS_KEY_ID="minio-root-user" \\\nAWS_SECRET_ACCESS_KEY="minio-root-password" \\\nAWS_ENDPOINT="http:\/\/localhost:19000" \\\nAWS_REGION="us-east-1" \\\nAWS_ALLOW_HTTP="true" \\\ndatafusion-cli<\/code><\/pre>CREATE EXTERNAL TABLE my_parquet_data \nSTORED AS PARQUET \nLOCATION 's3:\/\/warehouse\/019d81a3-c2d6-7ed2-ab15-070becf62582\/my_table-13e4b91a2b4e47d98f312b1384263880\/data\/';<\/code><\/pre>2. Массовая конвертация и выгрузка (DataFusion COPY)<\/h5>\n
COPY (\n -- Тут может быть любая сложная агрегация:\n -- SELECT id, count(data) FROM my_parquet_data GROUP BY id\n SELECT * FROM my_parquet_data\n) \nTO 's3:\/\/warehouse\/019d81a3-c2d6-7ed2-ab15-070becf62582\/my_table-13e4b91a2b4e47d98f312b1384263880\/json_export\/' \nSTORED AS JSON\nOPTIONS (\n 'format.compression' 'zstd'\n);<\/code><\/pre>+-------+\n| count |\n+-------+\n| 2 |\n+-------+\n1 row(s) fetched. \nElapsed 0.270 seconds.<\/code><\/pre>В чем преимущество подхода DataFusion?<\/h4>\n
\n
\n1. Локальная отказоустойчивость (защита от OOM)<\/h4>\n
\n
2. Распределенная отказоустойчивость (Аналог Trino FTE)<\/h4>\n
А. Apache Ballista (Официальный распределенный DataFusion)<\/h5>\n
\n
Б. DataFusion on Ray (datafusion-ray)<\/h5>\n
\nRay — это супер-устойчивый распределенный фреймворк. Интеграция `datafusion-ray` позволяет разбить SQL-запрос на граф задач прямо в Ray.<\/p>\n\n
Резюме: Как получить “Trino-like” Fault Tolerance в DataFusion?<\/h4>\n
\n
![\"\"](\"https:\/\/gavrilov.info\/pictures\/Snimok-ekrana-2025-09-09-v-01.35.35.png\")
\n<\/div>\nБольшая картина: Подготовка данных с помощью Trino<\/h4>\n
\n
-- Этот запрос выполняется в Trino, а не в DuckDB!\n INSERT INTO iceberg_catalog.analytics.daily_user_activity\n SELECT\n u.user_id,\n u.country,\n e.event_timestamp,\n e.track_id,\n e.duration_ms\n FROM\n postgres_catalog.public.users u\n JOIN\n kafka_catalog.raw_data.listen_events e ON u.user_id = e.user_id\n WHERE\n e.event_date = CURRENT_DATE;<\/code><\/pre>Подход 1: Табличные форматы (Iceberg) — Читайте только то, что нужно<\/h4>\n
\n
Назначение и сценарии использования:<\/h5>\n
\n
Подход 2: RPC-стриминг (Apache Arrow Flight) — Прямой канал к данным<\/h4>\n
\n
Назначение и сценарии использования:<\/h5>\n
\n
Подход 3: “API поверх данных” (ROAPI & DataFusion) — Декларативная альтернатива<\/h4>\n
![\"\"](\"https:\/\/gavrilov.info\/pictures\/image-212.png\")
\n<\/div>\n\n
Назначение и сценарии использования:<\/h5>\n
\n
и еще немного про DuckDB<\/h4>\n
1. Читайте меньше данных (Золотое правило)<\/h5>\n
\n
2. Оптимизация SQL-запросов<\/h5>\n
\n
3. Настройка окружения DuckDB<\/h5>\n
\n
Заключение: Какой подход выбрать?<\/h4>\n
![\"\"](\"https:\/\/gavrilov.info\/pictures\/duck.png\")
\n![\"\"](\"https:\/\/gavrilov.info\/pictures\/image-140.png\")
\n![\"\"](\"https:\/\/gavrilov.info\/pictures\/image-141.png\")
\n<\/div>\n![\"\"](\"https:\/\/gavrilov.info\/pictures\/image-142.png\")
\n<\/div>\n![\"\"](\"https:\/\/gavrilov.info\/pictures\/image-143.png\")
\n<\/div>\n![\"\"](\"https:\/\/gavrilov.info\/pictures\/image-144.png\")
\n<\/div>\n![\"\"](\"https:\/\/gavrilov.info\/pictures\/image-145.png\")
\n<\/div>\n![\"\"](\"https:\/\/gavrilov.info\/pictures\/image-146.png\")
\n<\/div>\n