{ "version": "https:\/\/jsonfeed.org\/version\/1.1", "title": "Yuriy Gavrilov: posts tagged big data", "_rss_description": "Welcome to my personal place for love, peace and happiness 🤖 Yuiry Gavrilov", "_rss_language": "en", "_itunes_email": "yvgavrilov@gmail.com", "_itunes_categories_xml": "", "_itunes_image": "https:\/\/gavrilov.info\/pictures\/userpic\/userpic-square@2x.jpg?1643451008", "_itunes_explicit": "no", "home_page_url": "https:\/\/gavrilov.info\/tags\/big-data\/", "feed_url": "https:\/\/gavrilov.info\/tags\/big-data\/json\/", "icon": "https:\/\/gavrilov.info\/pictures\/userpic\/userpic@2x.jpg?1643451008", "authors": [ { "name": "Yuriy Gavrilov - B[u]g - for charity.gavrilov.eth", "url": "https:\/\/gavrilov.info\/", "avatar": "https:\/\/gavrilov.info\/pictures\/userpic\/userpic@2x.jpg?1643451008" } ], "items": [ { "id": "327", "url": "https:\/\/gavrilov.info\/all\/arhitektura-client-spooling-kak-bystro-vygruzhat-gigantskie-data\/", "title": "Архитектура Client Spooling: Как быстро выгружать гигантские датасеты в Trino и Apache DataFusion", "content_html": "

Работа с Big Data часто упирается в классическое “узкое горлышко”: кластер может обработать терабайты данных за секунды, но передача результатов (Result Set) обратно на сторону клиента (например, в Jupyter или скрипт) занимает часы. На дворе апрель 2026 года, и современные аналитические движки предлагают эффективные методы обхода этой проблемы — концепцию Spooling<\/b>.<\/p>\n

Немного душноты:<\/p>\n

Архитектура Client Spooling в Trino создавалась с параноидальным акцентом на безопасность, в S3 выкидываются куски сырых, возможно, чувствительных данных.<\/p>\n

Когда Trino решает сбросить данные в объектное хранилище, он всегда шифрует их на лету.
\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

минимальный<\/p>\n
docker-compose.yml<\/code><\/pre>
<\/b><\/summary><\/p>\n
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>
Шаг 1. Настройка каталога и генерация данных (Trino)<\/b><\/summary>
\n

\nСначала мы генерируем данные в Trino. Запрос<\/p>\n
CREATE CATALOG<\/code><\/pre>
использует динамическое подключение к Lakekeeper REST API. Скрипт записывает файлы в формате Parquet в MinIO:<\/p>\n
config.properties<\/b><\/p>\n
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.properties<\/b><\/p>\n
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>
-- 1. Подключение каталога Iceberg<\/p>\n
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>
-- 2. Создание структуры<\/p>\n
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>
-- 3. Запись данных<\/p>\n
INSERT INTO test_warehouse.test_schema.my_table VALUES (1, 'hello'), (2, 'world');<\/code><\/pre>
\n
Если написать Select – должно быть как-то так<\/p>\n
\n\"\"\n<\/div>\n
Аналог Spooling в Apache DataFusion (Через экспорт)<\/h4>\n
Trino поддерживает протокол *Client Spooling* “из коробки” — когда Python-клиент запрашивает огромный `SELECT`, Trino сам незаметно пишет куски в S3 и отдает клиенту готовые ссылки.<\/p>\n
В Apache DataFusion<\/b> (который часто работает как локальный движок `datafusion-cli` или встраиваемая библиотка поверх S3) применяется более прозрачный паттерн делегирования (Explicit Spooling). Мы вручную инструктируем движок сохранить результаты агрегации в распределенное хранилище, чтобы позже забрать их в удобном формате — например, упаковав их в `JSON` и сжав алгоритмом `ZSTD`.<\/p>\n
1. Подключение к S3 и маппинг исходной таблицы<\/h5>\n
Запускаем `datafusion-cli`, передав доступы как переменные среды (для предотвращения ошибок парсинга опций):<\/p>\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>
Внутри консоли подключаем директорию с Parquet-файлами, сгенерированными Trino:<\/p>\n
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
Вместо того чтобы тянуть миллионы строк на локальный терминал, мы просим DataFusion выполнить преобразование и записать итог запроса обратно в MinIO.<\/p>\n
Мы выбираем построчный JSON с экстремальным сжатием:<\/p>\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>
Результат:<\/b><\/p>\n
+-------+\n| count |\n+-------+\n| 2     |\n+-------+\n1 row(s) fetched. \nElapsed 0.270 seconds.<\/code><\/pre>
За миллисекунды (0.270 sec) DataFusion прочитал партиции, трансформировал бинарные столбцы в текст и сжал его.<\/p>\n
В чем преимущество подхода DataFusion?<\/h4>\n
Описанный паттерн выполнения команды `COPY TO` с сохранением `.json.zst` в MinIO полностью воспроизводит механику Spooling:<\/p>\n
    \n
  1. Отсутствие OOM (Out Of Memory):<\/b> Клиент получает только метаданные `count`, а не гигабайты сырых данных в оперативную память.<\/li>\n
  2. Параллелизм:<\/b> Если исходных файлов много, DataFusion будет писать множество потоков `part-0.json.zst`, `part-1.json.zst` в бакет параллельно.<\/li>\n
  3. Удаленное потребление:<\/b> Вы можете запустить легкий Python-скрипт (Pandas) на дешевой машине, который просто прочитает эти сжатые легковесные JSON объекты напрямую из MinIO, минуя дорогостоящие вычислительные кластеры.<\/li>\n<\/ol>\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
    \n
    1. Локальная отказоустойчивость (защита от OOM)<\/h4>\n
    В Trino частой причиной падения задач является нехватка памяти (Out of Memory). В DataFusion реализован мощный механизм управления памятью.<\/p>\n
    Если DataFusion понимает, что оперативной памяти для агрегации или JOIN’а не хватает, он не “роняет” задачу, а начинает сбрасывать промежуточные данные на диск (Spill to Disk<\/b>).<\/p>\n