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

Group Commit

Кратко Apache Doris Group Commit объединяет множество небольших операторов INSERT VALUES или загрузок Stream Load в одну транзакцию на стороне сервера. Благодаря этому клиенты с частыми записями перестают перегружать FE разбором и планированием запросов, а BE мелкими rowset. Выберите sync_mode, если загрузка должна ждать, пока данные станут видимыми, и async_mode, если результат нужно вернуть сразу после надёжной записи строки в журнал упреждающей записи. С Group Commit та же нагрузка, которая приводила к ошибкам -235 из-за перегрузки, достигает скорости более 100,000 строк в секунду на одном BE.

Apache Doris Group Commit: объединение множества небольших операций INSERT и загрузок Stream Load в одну транзакцию на стороне сервера с выбором синхронного или асинхронного режима в зависимости от требований к задержке и пропускной способности.

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

Apache Doris Group Commit устраняет накладные расходы на каждую запись, которыми клиенты с частыми запросами нагружают планировщик FE и движок хранения BE. Построчные INSERT INTO ... VALUES и загрузки Stream Load крошечными пакетами используются для записи в Doris во многих реальных системах. Микросервис вызывает INSERT через JDBC каждый раз, когда пользователь на что-нибудь нажимает. Потребитель Kafka каждые несколько сотен миллисекунд отправляет всё, что успел накопить. Сборщик данных IoT передаёт каждое событие сразу после его поступления. Ни один из этих клиентов не может формировать более крупные пакеты, потому что следующих событий ещё нет. А без объединения в пакеты каждая загрузка открывает собственную транзакцию, отдельно проходит через планировщик FE и создаёт новую версию tablet. Именно такое сочетание создаёт проблемы для движка хранения Doris.

  • Частые INSERT VALUES исчерпывают ресурсы синтаксического анализатора и кеша планов FE раньше, чем ресурсы CPU на BE.
  • Небольшие загрузки Stream Load из потоковых конвейеров накапливают версии tablet, пока фоновая компакция не перестаёт справляться. Типичный симптом: ошибка -235 «слишком много сегментов».
  • Каждая загрузка становится отдельной транзакцией, которая тратит на согласование фиксации столько же времени, сколько на обработку самих данных.

Group Commit объединяет множество таких небольших записей в одну транзакцию на сервере, причём клиенту не нужно ничего об этом знать или учитывать.

Что такое Group Commit в Apache Doris?​

Apache Doris Group Commit представляет собой серверный буфер для небольших записей. BE собирает строки из множества параллельных загрузок и фиксирует их в одной транзакции каждые несколько секунд или при заполнении буфера. Group Commit работает поверх INSERT INTO ... VALUES, Stream Load и подготовленных операторов JDBC, не меняя их публичный API. Он включается одним параметром: переменной сеанса для INSERT, HTTP-заголовком для Stream Load или свойством таблицы, задающим режим по умолчанию. Выбранный режим определяет момент возврата результата загрузки.

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

  • group_commit: переменная сеанса (для INSERT) и HTTP-заголовок (для Stream Load), определяющие режим. Значения: off_mode, sync_mode, async_mode.
  • sync_mode: строки объединяются с данными параллельных загрузок на BE, а запрос блокируется, пока объединённая транзакция не будет зафиксирована и данные не станут видимыми.
  • async_mode: строки записываются в журнал упреждающей записи на BE, запрос немедленно возвращает результат со статусом PREPARE, а объединённая транзакция фиксируется позже.
  • WAL: отдельный для каждого BE журнал упреждающей записи, используемый в async_mode. Имеет одну реплику и хранится на принимающем BE.
  • group_commit_interval_ms и group_commit_data_bytes: условия сброса буфера, задаваемые для каждой таблицы. Значения по умолчанию: 10s и 128 MB.

Как работает Group Commit в Apache Doris?​

Apache Doris Group Commit направляет параллельные небольшие загрузки в общую очередь на BE для каждой таблицы, сбрасывает очередь по таймеру или при достижении порогового размера и фиксирует объединённый пакет как одну транзакцию FE.

  1. Создание очереди или присоединение к ней. Запрос поступает на BE. Вместо запуска новой транзакции BE помещает строки в LoadBlockQueue, определяемую таблицей и текущей открытой транзакцией Group Commit.
  2. Запись в WAL (только в асинхронном режиме). В async_mode строки также сохраняются в локальном WAL, поэтому запрос может вернуть результат сразу после их надёжной записи. В sync_mode этот шаг пропускается.
  3. Ожидание условия сброса. BE сбрасывает очередь, когда истекает интервал group_commit_interval_ms или в очереди накапливается объём данных group_commit_data_bytes. Оба свойства задаются для каждой таблицы.
  4. Фиксация одной транзакции для всего пакета. BE передаёт объединённый пакет через движок хранения, а FE фиксирует ровно одну транзакцию. FE назначает одну метку Label с префиксом group_commit_, общую для всех объединённых загрузок.
  5. Возврат результата. Запросы sync_mode до этого момента оставались заблокированными и теперь возвращают результат. Запросы async_mode уже вернули результат ранее; на этом этапе данные становятся доступны для запросов.

Если запрос соответствует условию перехода на обычный путь загрузки (явная транзакция, заданная пользователем метка Label, обновление части столбцов, выражения в VALUES, нехватка места на диске для WAL в async_mode), Doris автоматически использует обычный путь для этого запроса. Решение определяется только формой запроса.

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

CREATE TABLE events (
  id BIGINT, ts DATETIME, payload STRING
)
DUPLICATE KEY(id)
DISTRIBUTED BY HASH(id) BUCKETS 4
PROPERTIES ("group_commit_interval_ms" = "1000");

SET group_commit = async_mode;
INSERT INTO events VALUES (1, NOW(), 'a');
INSERT INTO events VALUES (2, NOW(), 'b');
-- the second insert merges into the first one's transaction

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

Query OK, 1 row affected (0.01 sec)
{'label':'group_commit_a145ce07f1c972fc-bd2c54597052a9ad', 'status':'PREPARE', 'txnId':'181508'}

Обе вставки возвращают одинаковые label и txnId, что указывает на их объединение. Строки не видны, пока BE не сбросит буфер (в этом примере через секунду, поскольку для таблицы задано свойство group_commit_interval_ms со значением 1000). Для Stream Load замените переменную сеанса заголовком: -H "group_commit:async_mode".

Когда стоит использовать Group Commit в Apache Doris?​

Используйте Apache Doris Group Commit, когда множество параллельных клиентов записывают небольшие пакеты и увеличить их размер на стороне клиента невозможно.

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

  • Микросервисы, выполняющие INSERT INTO ... VALUES для каждого события через JDBC с частотой в десятки TPS и выше.
  • Приёмники Kafka или Flink, которые могут формировать только пакеты размером меньше мегабайта и сталкиваются с ошибками -235 на BE.
  • Конвейеры IoT или потоков кликов, в которых каждое событие должно поступить в течение нескольких секунд, но объединить клиентов невозможно.
  • Таблицы, в которых частое появление новых версий tablet из-за отдельных загрузок перегружает фоновую компакцию.

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

  • Загрузки, которые должны выполняться строго в одной транзакции с учётными операциями вызывающего кода. Group Commit автоматически переходит на обычный путь загрузки при использовании явного BEGIN ... COMMIT. Для транзакционных приёмников используйте непосредственно транзакции загрузки.
  • Stream Load 2PC и записи с собственной меткой Label. Оба варианта обходят Group Commit; если нужна семантика exactly-once с контрольными точками, продолжайте использовать Stream Load 2PC.
  • Обновления части столбцов. Group Commit не поддерживает partial_columns: true и переходит на обычный путь загрузки.
  • Таблицы Unique Key, которым требуется строгое соблюдение порядка фиксации. Group Commit не сохраняет порядок между объединёнными загрузками; задайте function_column.sequence_col, чтобы актуальная версия определялась по временной метке, предоставленной пользователем.
  • Загрузки, пакеты которых уже достаточно велики (сотни MB и больше). По сравнению с обычной загрузкой координация Group Commit создаёт лишнюю работу. Просто отправьте загрузку напрямую.

Производительность / показатели​

Из примечаний к выпуску Apache Doris 2.1.0:

  • JDBC, 1 FE + 1 BE, TPC-H SF10 Lineitem, параллелизм 20, менее 100 строк на вставку: 106,900 строк/с в async_mode.
  • Stream Load, 3 BE, параллелизм 10, пакеты меньше 1 MB: 810,000 строк/с, 104 MB/с. Без Group Commit та же нагрузка приводит к ошибкам -235 и остановке загрузки.
  • Stream Load, 3 BE, пакеты по 1 MB при параллелизме 30: 2,077,723 строк/с.

Источник: примечания к выпуску Apache Doris 2.1.0.

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

Report issue