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

Транзакции загрузки

Кратко Apache Doris выполняет каждую загрузку данных в транзакции. Вы можете объединить несколько вставок в один блок BEGIN ... COMMIT, устранять дубликаты при повторных попытках с помощью Label загрузки и управлять двухфазной фиксацией через заголовки HTTP-запроса Stream Load. Именно последняя возможность позволяет Flink, Spark и конвейерам CDC записывать данные в Doris с семантикой exactly-once без собственной служебной таблицы для учёта операций.

Транзакции загрузки Apache Doris: выполняйте загрузки из нескольких операторов как единое атомарное целое, с дедупликацией по Label и 2PC для потоковых приёмников с семантикой exactly-once.

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

Транзакции загрузки Apache Doris переносят решение проблем с дублирующимися строками и частично выполненными записями из приложения в хранилище данных. Благодаря этому повторная обработка данных во Flink, незавершённая запись в несколько таблиц или повторная загрузка файла обрабатываются корректно. При загрузке данных в хранилище возникают типичные сбои. Задание Flink перезапускается во время создания контрольной точки и повторно обрабатывает последний пакет, дублируя строки. Конвейер CDC обновляет таблицу фактов, но связанная запись в таблицу измерений завершается с ошибкой, и данные остаются рассогласованными до следующей дозагрузки. Скрипт повторных попыток загружает тот же файл ещё раз, поскольку не может определить, завершилась ли предыдущая попытка успешно. В каждом из этих случаев приложению приходится вести сложный служебный учёт.

  • Повторная обработка данных после зависания потокового задания приводит к дублированию строк.
  • Записи в несколько таблиц, которые должны выполняться как единое целое, могут прерваться на полпути.
  • Повторная отправка одного и того же файла приводит к его повторной загрузке, если вы сами не отслеживаете такие случаи.

Apache Doris решает все три проблемы на сервере, позволяя упростить клиент загрузки.

Что такое транзакции загрузки в Apache Doris?​

Транзакции загрузки Apache Doris служат атомарной единицей записи для всех способов приёма данных. Любая запись в Doris выполняется внутри транзакции. Это относится к INSERT INTO, Stream Load, Routine Load и Broker Load. Каждая транзакция получает уникальную Label и проходит через отслеживаемые состояния жизненного цикла под управлением FE. Можно позволить Apache Doris неявно открывать и фиксировать транзакцию для каждой загрузки, управлять ею явно с помощью BEGIN/COMMIT/ROLLBACK или разделить фиксацию на две фазы через HTTP для потоковых приёмников.

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

  • Label: строка, идентифицирующая транзакцию в пределах базы данных. Повторная отправка с той же Label обеспечивает дедупликацию загрузки.
  • Implicit transaction: каждая отдельная загрузка выполняется в собственной транзакции, которая автоматически фиксируется при успехе.
  • Explicit transaction: блок BEGIN ... COMMIT; объединяет несколько записей в одной базе данных в единое атомарное целое.
  • Stream Load 2PC: загрузка Stream Load, которая остаётся в состоянии PRECOMMITTED до тех пор, пока последующий HTTP-вызов не зафиксирует или не прервёт её.
  • Lifecycle states: PREPARE → PRECOMMITTED → COMMITTED → VISIBLE (или ABORTED). Строки доступны для запросов только после перехода в VISIBLE.

Как работают транзакции загрузки в Apache Doris?​

Транзакции загрузки Apache Doris проводят каждую запись через пять состояний под управлением FE (PREPARE → PRECOMMITTED → COMMITTED → VISIBLE или ABORTED при сбое). При этом Label обеспечивает дедупликацию повторных попыток, а заключительная публикация на уровне реплик одновременно делает строки доступными для запросов.

  1. Открытие. Загрузка открывает транзакцию в состоянии PREPARE. FE назначает TxnId и связывает его с Label, предоставленной пользователем или сгенерированной автоматически. При повторной отправке с той же Label сразу возвращается результат исходной транзакции.
  2. Запись. Узлы BE получают данные, создают сегменты и сообщают об успешном завершении. Транзакция переходит в PRECOMMITTED, когда все ожидаемые операции загрузки подтвердят завершение.
  3. Фиксация. Команда COMMIT (или HTTP-вызов второй фазы с txn_operation: commit) переводит транзакцию в COMMITTED. Данные уже находятся на диске, но ещё недоступны для чтения.
  4. Публикация. Фоновый процесс отправляет задачу PublishVersion каждой реплике. Когда все реплики всех затронутых таблетов подтвердят выполнение, транзакция переходит в VISIBLE.
  5. Чтение. Запросы, выполняемые после перехода в VISIBLE, видят новые строки. Более ранние запросы видят предыдущую версию. Таблеты в одной транзакции никогда не возвращают несогласованные данные при чтении.

Если на любом этапе происходит сбой или клиент вызывает ROLLBACK (либо txn_operation: abort), транзакция переходит в ABORTED, а данные отбрасываются.

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

CREATE TABLE orders (
  id INT, amount DECIMAL(10,2)
) DUPLICATE KEY(id) DISTRIBUTED BY HASH(id) BUCKETS 4;

CREATE TABLE order_audit (
  id INT, action STRING
) DUPLICATE KEY(id) DISTRIBUTED BY HASH(id) BUCKETS 4;

BEGIN;
INSERT INTO orders       VALUES (1, 99.50), (2, 149.00);
INSERT INTO order_audit  VALUES (1, 'created'), (2, 'created');
COMMIT;

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

Query OK, 2 rows affected
{'label':'txn_insert_b55db21aad7451b','status':'VISIBLE','txnId':'10013'}

Данные в обеих таблицах становятся доступными для чтения одновременно. Если бы второй INSERT завершился с ошибкой или соединение оборвалось до COMMIT, новые строки не появились бы ни в одной из таблиц.

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

Используйте транзакции загрузки Apache Doris, когда несколько вставок должны выполняться как единое целое, потоковому приёмнику нужна семантика exactly-once или требуется дедупликация повторных загрузок по Label на стороне хранилища данных.

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

  • Приёмники Flink и Spark, использующие Stream Load 2PC для обеспечения семантики exactly-once между контрольными точками.
  • Конвейеры CDC, которым необходимо согласованно обновлять таблицы фактов и измерений.
  • Дозагрузки, при которых пакет связанных строк добавляется или удаляется как единое целое.
  • Нагрузки, для которых важна идемпотентность повторных попыток и требуется, чтобы дедупликацию по Label обеспечивало хранилище данных, а не клиент.

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

  • Частые вставки по одной строке. Каждая транзакция добавляет обмен запросом и ответом для фиксации и новую версию таблета, а при небольших загрузках этап фиксации становится узким местом. Используйте Group Commit.
  • Атомарность между двумя базами данных. Все таблицы одной транзакции должны находиться в одной базе данных.
  • Stream Load 2PC для таблиц Unique Key с Merge-on-Write в облачных развёртываниях / развёртываниях с разделением хранения и вычислений. BE отклоняет такое сочетание с ошибкой Status::NotSupported; в кластерах on-prem (shared-nothing) оно разрешено.
  • Долгосрочная дедупликация. Label удаляются примерно через 3 дня или при достижении 2,000 записей на базу данных, в зависимости от того, что наступит раньше; повторное использование Label после этого не обеспечивает дедупликацию. Отслеживайте долгоживущие ключи идемпотентности в приложении.
  • Чтение собственных записей внутри той же транзакции. Каждый оператор работает со снимком, сделанным в момент его запуска; результат предыдущего INSERT становится виден только после COMMIT.

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

Report issue