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

Stream Load

Кратко Apache Doris Stream Load позволяет клиенту с помощью curl (или любой HTTP-библиотеки) отправить файл или поток байтов в таблицу одним запросом HTTP PUT. Узел FE перенаправляет запрос на BE, выбранный BE выступает координатором, разбирает CSV/JSON/Parquet/ORC, записывает данные в целевые таблеты и возвращает тело ответа в формате JSON, которое сообщает клиенту, зафиксирован ли весь пакет. Одна метка исключает дублирование при повторных попытках, одна транзакция обеспечивает атомарность пакета, а ответ JSON содержит количество строк и время выполнения каждого этапа.

Apache Doris Stream Load: синхронная загрузка CSV, JSON, Parquet и ORC через HTTP PUT с дедупликацией по меткам и атомарной фиксацией каждого пакета.

Зачем использовать Apache Doris Stream Load?​

Apache Doris Stream Load подходит клиентам, которые уже работают по HTTP и хотят получать синхронный ответ для каждого пакета, с гарантией безопасных повторных попыток по метке и без дополнительного менеджера транзакций. Запись в хранилище данных в реальном времени часто начинается со скрипта, который отправляет файл. Такой скрипт работает внутри приёмника Flink, потребителя Kafka, задания cron с rsync или микросервиса, который только что сформировал несколько тысяч событий. Автор скрипта хочет отправить байты, получить ответ в том же соединении и быть уверенным в безопасности повторной попытки.

  • Заданию Flink нужно сбросить данные текущей контрольной точки в Apache Doris и узнать, успешно ли выполнена фиксация, прежде чем продвигать смещения.
  • У ETL-скрипта есть CSV размером 200 MB из вчерашней выгрузки, и ему нужно, чтобы строки стали доступны до запуска утреннего отчёта.
  • Вспомогательный сервис отправляет пакет JSON каждые несколько секунд и нуждается в семантике at-most-once без подключения менеджера транзакций.

Apache Doris Stream Load напрямую поддерживает такой способ записи. Один HTTP PUT, одна метка, одна транзакция, один синхронный ответ через тот же сокет.

Что такое Apache Doris Stream Load?​

Apache Doris Stream Load — это синхронный HTTP-эндпоинт загрузки, доступный на каждом узле кластера. Клиент отправляет HTTP PUT на FE или BE; FE перенаправляет запрос на BE по алгоритму round-robin, а выбранный BE становится координатором этой загрузки. Координатор разбирает тело запроса, распределяет строки между узлами BE, которым принадлежат соответствующие таблеты, открывает одну транзакцию через FE, фиксирует её после получения подтверждений от всех BE и возвращает тело ответа в формате JSON через тот же сокет. Apache Doris с самого начала поддерживает CSV, а теперь также принимает JSON, Parquet, ORC и варианты CSV с заголовком.

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

  • BE-координатор: узел backend, который получает тело HTTP-запроса, разбирает его и управляет загрузкой. Он либо выбирается при перенаправлении с FE, либо клиент обращается к нему напрямую.
  • label: строка, переданная в запросе (или сгенерированная сервером), которая задаёт имя транзакции. Повторное использование метки успешной загрузки приводит к отклонению дубликата; метку неудачной загрузки можно использовать повторно.
  • /api/{db}/{table}/_stream_load: основной эндпоинт. Вариант для 2PC — _stream_load_2pc.
  • Expect: 100-continue: обязательный заголовок. Клиент сначала отправляет заголовки, даёт серверу проверить аутентификацию и авторизацию и начать транзакцию, а затем отправляет тело запроса.
  • max_filter_ratio: допустимая доля некорректных строк для отдельной загрузки. По умолчанию равна 0, поэтому одна некорректная строка приводит к сбою всего пакета.

Как работает Apache Doris Stream Load?​

Apache Doris Stream Load выполняет всю загрузку в рамках одного HTTP-запроса: FE перенаправляет запрос на BE-координатор, BE разбирает тело, открывает транзакцию, распределяет строки между владельцами таблетов и отвечает через тот же сокет.

  1. Клиент отправляет PUT на FE (или BE). Заголовки содержат метку, формат, сопоставление столбцов, фильтры и тайм-ауты. Тело запроса представляет собой файл или поток. Expect: 100-continue позволяет серверу отклонить запрос до передачи тела.
  2. FE выбирает BE-координатор и перенаправляет запрос. FE возвращает HTTP 307 с перенаправлением на BE, выбранный по алгоритму round-robin. Клиент, которому уже известен работоспособный BE, может пропустить этот шаг и отправить PUT напрямую на BE.
  3. BE-координатор открывает транзакцию и разбирает данные. Он обращается к FE, чтобы начать транзакцию с указанной меткой, а затем пропускает поток тела запроса через модуль чтения соответствующего формата (CSV, JSON, Parquet, ORC). Строки, которые не удалось разобрать или которые не прошли проверку схемы, учитываются при проверке max_filter_ratio.
  4. Распределение, запись, фиксация. Координатор направляет каждую строку на узлы BE, содержащие нужные таблеты, эти BE сбрасывают memtables в сегменты, координатор запрашивает у FE фиксацию транзакции, а FE отправляет PublishVersion каждой реплике.
  5. Возврат JSON. Координатор отвечает на исходный PUT телом в формате JSON: Status, Label, TxnId, NumberLoadedRows, NumberFilteredRows, ErrorURL, если какие-либо данные не прошли проверку качества, и время выполнения каждого этапа. Соединение закрывается.

С точки зрения клиента вся последовательность представляет собой один HTTP-запрос. Если после успешной фиксации клиент повторит попытку с той же меткой, FE отклонит её с ответом Label Already Exists; это и обеспечивает гарантию at-most-once.

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

Создайте целевую таблицу:

CREATE DATABASE IF NOT EXISTS testdb;
CREATE TABLE testdb.test_streamload (
  user_id BIGINT NOT NULL,
  name    VARCHAR(20),
  age     INT
) DUPLICATE KEY(user_id) DISTRIBUTED BY HASH(user_id) BUCKETS 1;

Затем создайте CSV-файл из 10 строк и отправьте его в Doris через HTTP PUT. Замените 127.0.0.1 адресом вашего FE, если Doris работает на другом хосте:

cat > streamload_example.csv <<'CSV'
1,Emily,25
2,Benjamin,35
3,Olivia,28
4,Alexander,60
5,Ava,17
6,William,69
7,Sophia,32
8,James,64
9,Emma,37
10,Liam,64
CSV

curl --location-trusted -u root: \
  -H "Expect:100-continue" \
  -H "label:orders_2024_06_01" \
  -H "column_separator:," \
  -H "columns:user_id,name,age" \
  -T streamload_example.csv \
  -XPUT http://127.0.0.1:8030/api/testdb/test_streamload/_stream_load

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

{
  "TxnId": 3,
  "Label": "orders_2024_06_01",
  "Status": "Success",
  "NumberTotalRows": 10,
  "NumberLoadedRows": 10,
  "NumberFilteredRows": 0,
  "LoadBytes": 118,
  "LoadTimeMs": 173
}

Status: Success означает, что пакет зафиксирован и строки доступны для запросов: SELECT COUNT(*) FROM testdb.test_streamload возвращает 10. Повторный запуск той же команды curl возвращает Status: Label Already Exists и ничего не записывает, поэтому повторная попытка клиента после кратковременного сбоя сети не приведёт к двойной загрузке.

Когда следует использовать Apache Doris Stream Load?​

Используйте Apache Doris Stream Load, когда один HTTP PUT на пакет, гарантия безопасных повторных попыток по метке и синхронный ответ JSON в точности соответствуют требованиям клиента.

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

  • Приёмники Flink, Spark или других систем потоковой обработки, которые уже работают по HTTP и хотят получать синхронный ответ для каждой контрольной точки. Используйте вместе с two_phase_commit:true, чтобы обеспечить семантику exactly-once между контрольными точками.
  • ETL-скрипты, которые отправляют один файл CSV, JSON, Parquet или ORC (обычно размером примерно до 10 GB) и хотят зафиксировать его одной атомарной операцией.
  • Однострочные команды Bash и встроенные HTTP-клиенты языков программирования, которые программно формируют данные и передают их через curl -T -.
  • Загрузки, которым нужна гарантия безопасных повторных попыток по метке без менеджера транзакций.

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

  • Десятки тысяч крошечных PUT-запросов в секунду от множества клиентов. Каждая загрузка открывает собственную транзакцию и создаёт собственный rowset, что перегружает планировщик FE и вызывает ошибку -235 Too many segments на BE. Используйте Group Commit, чтобы BE объединял небольшие записи в одну транзакцию, или перейдите на Routine Load для Kafka.
  • Отдельные файлы размером значительно больше 10 GB. Apache Doris Stream Load выполняется как один синхронный запрос; если соединение оборвётся, придётся начинать заново. Разделите файл или используйте Batch Load, который выполняется на стороне сервера и продолжает работать после отключения клиента.
  • Непрерывное чтение новых сообщений из топика Kafka. Stream Load ничего не знает о смещениях. Используйте Routine Load, который управляет смещениями, перезапусками и приостановкой за вас.
  • Stream Load 2PC для таблиц Merge-on-Write Unique Key в облачных развёртываниях / развёртываниях с разделением хранения и вычислений. BE отклоняет такое сочетание с ответом Status::NotSupported. Локальные кластеры (shared-nothing) допускают его, и оно покрыто набором регрессионных тестов.
  • Загрузки, для которых требуется ведение истории на стороне сервера без дополнительной настройки. По умолчанию Apache Doris не записывает историю Stream Load; включите enable_stream_load_record в be.conf, если команда SHOW STREAM LOAD должна возвращать данные.

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

Report issue