Перейти к содержимому
Последнее обновление

Пакетная загрузка

Кратко Пакетная загрузка Apache Doris переносит большие файлы из S3, HDFS или другого хранилища данных в Doris за один запуск. Вы отправляете оператор LOAD LABEL, FE планирует работу и распределяет её между узлами BE, а ход выполнения можно проверить с помощью SHOW LOAD. Задание выполняется асинхронно, поэтому клиент может отключиться; метка позволяет найти результат и исключить дублирование при повторной попытке. Для внешних каталогов и чтения через TVF синхронная форма INSERT INTO ... SELECT решает те же задачи и упрощает ETL.

Пакетная загрузка Apache Doris: асинхронная массовая загрузка из S3, HDFS и внешних каталогов в Doris с помощью LOAD LABEL или INSERT INTO SELECT, с дедупликацией по меткам.

Зачем использовать пакетную загрузку в Apache Doris?​

Пакетная загрузка Apache Doris подходит для случаев, когда одно задание загрузки выполняется часами и за его жизненный цикл должно отвечать хранилище данных, а не клиент. Средства загрузки в реальном времени хорошо подходят для постепенно поступающих событий, но не справляются, когда нужно перенести TB исторических данных, перестроить таблицу фактов из снимка Hive или загрузить данные за прошлый квартал из Parquet на S3. Загрузка через Stream Load по HTTP прерывается по тайм-ауту. Построчный INSERT занял бы несколько дней. Задание должно выполняться на сервере, самостоятельно повторять попытки и продолжать работу при потере соединения с клиентом.

  • В рамках ночного ETL нужно загрузить несколько сотен GB данных Parquet из S3 до утреннего обновления дашбордов.
  • При миграции из Snowflake или Hive нужно копировать целые партиции, применяя фильтры к столбцам и преобразования типов.
  • Дозагрузка данных за один пропущенный день не должна блокировать соединение, через которое её запустили.

Пакетная загрузка решает все три задачи: большая массовая загрузка становится длительным заданием, которое отслеживается по метке и на всех этапах управляется хранилищем данных.

Что такое пакетная загрузка Apache Doris?​

Пакетная загрузка Apache Doris объединяет методы асинхронной массовой загрузки в Doris. Основная точка входа — оператор LOAD LABEL ... WITH S3|HDFS|BROKER, исторически называемый Broker Load. Сегодня он поддерживает S3, HDFS и любые системы хранения, доступные через процесс Broker. Синхронный вариант INSERT INTO ... SELECT решает те же задачи загрузки, когда источником служит внешний каталог (Hive, Iceberg, JDBC) или файл, доступный через TVF, например S3() или HDFS(). Оба варианта используют ту же модель транзакций, что и остальные методы загрузки Doris: одна метка на задание, атомарная фиксация и публикация на уровне реплик.

Основные термины

  • LOAD LABEL: оператор SQL для запуска асинхронной загрузки Broker Load. Метка действует в пределах базы данных и предотвращает дублирование при повторных попытках.
  • WITH S3 / WITH HDFS / WITH BROKER: предложение, определяющее способ доступа к системе хранения. Поддержка S3 и HDFS встроена в BE; BROKER broker_name направляет обращения ко всем остальным системам через внешний процесс Broker.
  • SHOW LOAD: представление на стороне FE, которое показывает состояние, ход выполнения, URL с ошибками и разобранное содержимое EtlInfo.
  • TVF (табличная функция): S3(...), HDFS(...), LOCAL(...), iceberg_meta(...) и подобные функции. Они представляют файлы или внешние метаданные в виде таблицы, из которой можно читать с помощью SELECT, а затем передавать данные в INSERT INTO ... SELECT для синхронной загрузки.
  • max_filter_ratio: допустимая доля некорректных строк для отдельного задания. По умолчанию равна нулю, поэтому даже одна некорректная строка отменяет загрузку.

Как работает пакетная загрузка Apache Doris?​

Пакетная загрузка Apache Doris регистрирует задание с меткой на FE, распределяет сканирование файлов между узлами BE и атомарно фиксирует результат после того, как каждая реплика опубликует новую версию.

  1. Отправка и присвоение метки. Запрос LOAD LABEL my_db.daily_orders ... поступает на FE. FE регистрирует метку, открывает транзакцию, и загрузка переходит в состояние PENDING. Повторная отправка той же метки в течение срока её хранения сразу возвращает исходное задание.
  2. Планирование и распределение работы. FE оценивает размеры файлов (один BE обрабатывает объём между min_bytes_per_broker_scanner и max_bytes_per_broker_scanner, по умолчанию от 64 MB до 500 GB), разделяет работу и отправляет задачи сканирования узлам BE. Задание переходит в состояние LOADING.
  3. Параллельное чтение и запись. Каждый BE получает свою часть исходных данных (из S3, HDFS или через Broker), разбирает CSV/JSON/Parquet/ORC, применяет выражения SET, фильтр WHERE или извлечение значений с помощью COLUMNS FROM PATH и записывает сегменты в целевые tablet. Некорректные строки учитываются при проверке max_filter_ratio; выборка таких строк доступна по ссылке из поля URL.
  4. Фиксация и публикация. После получения подтверждений от всех узлов BE FE фиксирует транзакцию. Каждой реплике отправляется задача PublishVersion. Когда все реплики подтверждают публикацию, задание переходит в состояние FINISHED, и строки становятся доступны для запросов. Сбой на любом этапе переводит задание в состояние CANCELLED, а данные отбрасываются.
  5. Проверка с помощью SHOW LOAD. Команда SHOW LOAD WHERE LABEL = ... показывает State, Progress, EtlInfo (число просканированных, отфильтрованных и ошибочных строк), а также URL с ошибками, если что-то пошло не так.

Форма INSERT INTO target SELECT * FROM s3(...) использует такую же структуру плана, но выполняется синхронно. Соединение должно оставаться открытым до завершения загрузки; взамен вы сразу получаете результат и возможность объединять операции средствами SQL (соединения таблиц, преобразования типов, проталкивание предикатов в источник).

Быстрый старт​

LOAD LABEL testdb.orders_2024_q1
(
  DATA INFILE("s3://my-bucket/orders/2024-q1/*.parquet")
  INTO TABLE orders
  FORMAT AS "parquet"
  (order_id, customer_id, amount, order_date)
)
WITH S3 (
  "provider"      = "S3",
  "AWS_ENDPOINT"  = "https://s3.us-west-2.amazonaws.com",
  "AWS_REGION"    = "us-west-2",
  "AWS_ACCESS_KEY" = "<ak>",
  "AWS_SECRET_KEY" = "<sk>"
)
PROPERTIES ("timeout" = "3600", "max_filter_ratio" = "0.01");

Ожидаемый результат

+----------+--------------------+----------+------------------+--------+
| JobId    | Label              | State    | Progress         | Type   |
+----------+--------------------+----------+------------------+--------+
| 41326624 | orders_2024_q1     | FINISHED | ETL:100%;LOAD:100% | BROKER |
+----------+--------------------+----------+------------------+--------+

SHOW LOAD ORDER BY CreateTime DESC LIMIT 1 возвращает результат, когда задание находится в состоянии FINISHED. Допускается до 1% некорректных строк; превышение этого порога отменяет задание, а поле URL указывает на выборку строк, вызвавших ошибку.

Когда следует использовать пакетную загрузку Apache Doris?​

Используйте пакетную загрузку Apache Doris, когда за один запуск нужно обработать десятки GB данных или больше из объектного хранилища, внешнего хранилища данных или каталога с партиционированием Hive.

Подходящие сценарии

  • Периодическая загрузка больших файлов CSV, JSON, Parquet или ORC из S3, HDFS, OSS, COS, OBS или любого S3-совместимого хранилища.
  • Разовые миграции из Hive, Iceberg, Snowflake или BigQuery: используйте Catalog вместе с INSERT INTO ... SELECT для выбора нужных столбцов и проталкивания операций на сторону источника.
  • Дозагрузка данных объёмом от десятков до сотен GB на задание, когда клиент не может поддерживать соединение в течение нескольких часов.
  • Загрузки, требующие фильтрации на сервере через WHERE, преобразования столбцов через SET или извлечения полей партиционирования из пути к файлу с помощью COLUMNS FROM PATH AS.
  • Каталоги с партиционированием Hive, выбираемые по шаблонам (*, ?, {1..10}) в одном операторе.

Неподходящие сценарии

  • Непрерывная загрузка из Kafka. Используйте Routine Load, чтобы FE управлял потребителем и смещениями.
  • Запись каждого отдельного события через JDBC из микросервиса. Для небольших объёмов данных накладные расходы на фиксацию каждого задания неоправданны. Используйте Group Commit.
  • Загрузка локальных файлов с клиентской машины. Broker Load читает только из удалённых хранилищ; для этого сценария подходят Stream Load или Doris Streamloader.
  • Отдельные задания объёмом свыше нескольких TB. Тайм-аут по умолчанию в 4 часа может неожиданно прервать загрузку, повторные попытки обходятся дорого, а частичный сбой требует повторной обработки всего пакета. Разделите входные данные по датам или каталогам и отправьте несколько заданий.
  • Непрерывная загрузка новых файлов из S3. Ночной запуск Broker Load подходит; ежеминутный опрос — нет. Используйте непрерывную загрузку Streaming Job, чтобы FE отслеживал уже загруженные файлы.

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

  • Stream Load: синхронная загрузка по HTTP для небольших пакетов данных в реальном времени.
  • Group Commit: объединение загрузок на сервере для частых небольших пакетов.
  • Справочник по Broker Load: полный синтаксис SQL, настройка Kerberos и HDFS HA, примеры для различных сценариев.
  • INSERT INTO Select: синхронный вариант, включая загрузку через TVF и Catalog.
  • Таблица выбора метода загрузки: какой метод подходит для каждого сценария загрузки — в реальном времени, потоковой или пакетной.
  • Знакомство с Catalog и TVF: как S3(), HDFS() и подобные функции представляют файлы в виде таблиц, доступных для запросов.
  • Транзакции загрузки: метки, состояния жизненного цикла и модель атомарной фиксации, общие для пакетной загрузки и остальных методов загрузки Doris.
  • Справочник SQL по BROKER LOAD: точная грамматика оператора и все его параметры.
Report issue