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

Интеграция с Kafka и CDC

Кратко Apache Doris изначально поддерживает загрузку данных из топиков Kafka и потоков изменений CDC тремя способами. Используйте Routine Load, если нужен потребитель Kafka, работающий внутри FE и не требующий внешнего сервиса. Используйте Flink Doris Connector вместе с Flink CDC, если нужна синхронизация всей базы данных MySQL/Postgres/Oracle с гарантией exactly-once через Stream Load 2PC. Используйте Doris Kafka Connector, если у вас уже работает Kafka Connect, а данные имеют формат Avro, Protobuf или Debezium. Все три способа записывают данные в одни и те же таблицы Unique Key Merge-on-Write, поэтому события, поступающие не по порядку, и удаления обрабатываются одинаково независимо от выбранного способа.

Интеграция Apache Doris с Kafka и CDC: загружайте данные из топиков Kafka напрямую в Doris с помощью Routine Load или передавайте поток CDC из MySQL/Postgres через Flink с гарантией exactly-once на базе Stream Load 2PC.

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

Интеграция Apache Doris с Kafka и CDC обеспечивает потоковую загрузку данных, доставку exactly-once и обработку событий, поступающих не по порядку, внутри движка, сокращая путь от топика или binlog до строки, доступной для запросов. В большинстве хранилищ потоковую загрузку приходится добавлять отдельно. Вы запускаете потребитель Kafka, заново реализуете учёт смещений, повторные попытки, сопоставление схем и удаления и надеетесь, что потребитель корректно завершит работу при отказе хранилища. Или ставите перед хранилищем Flink и платите за дополнительный кластер. Или отказываетесь от «реального времени» и отправляете пакеты раз в пять минут через пакетную загрузку.

  • Конвейер CDC повторно обрабатывает последний пакет после сбоя контрольной точки и дублирует каждую строку из своего буфера.
  • Обновление, поступившее в таблицу Unique Key не по порядку, уступает более старой версии, потому что движку не указали, какая запись новее.
  • Топик Kafka с данными Avro или Protobuf не подходит для API загрузки по HTTP, который принимает только JSON или CSV.

Doris переносит большую часть этой логики в движок, поэтому путь от топика Kafka или binlog MySQL до строки, доступной для запросов, остаётся коротким, а возможные сценарии отказов понятны.

Что представляет собой интеграция с Kafka и CDC в Apache Doris?​

Интеграция Apache Doris с Kafka и CDC представляет собой набор способов потоковой загрузки данных и загрузки CDC с общей основой: транзакционные загрузки с идентификатором Label, таблицы Unique Key Merge-on-Write для вставки с обновлением и удаления, а также столбец последовательности, определяющий приоритетную версию при поступлении событий не по порядку. Способы различаются тем, кто запускает потребитель.

Ключевые термины

  • Routine Load: длительно работающий потребитель Kafka под управлением FE. Достаточно одного оператора CREATE ROUTINE LOAD, и FE распределяет задачи между BE, которые получают пакеты данных и оформляют каждый пакет как транзакционную загрузку.
  • Stream Load 2PC: загрузка по HTTP, которая останавливается в состоянии PRECOMMITTED, пока последующий вызов не передаст txn_operation: commit или txn_operation: abort. Именно разделение на две фазы обеспечивает гарантию exactly-once для внешних sink.
  • Flink Doris Connector: внешний sink для Flink, который управляет Stream Load 2PC и привязывает фиксацию транзакций к контрольным точкам Flink. В сочетании с Flink CDC обеспечивает синхронизацию всей базы данных и передачу изменений DDL.
  • Doris Kafka Connector: sink-плагин для Kafka Connect. Использует экосистему конвертеров Kafka Connect, поэтому данные Avro, Protobuf и Debezium поддерживаются без собственного сериализатора.
  • function_column.sequence_col: свойство таблиц Unique Key MOW, указывающее столбец, значение которого определяет, какая версия строки будет выбрана. Защита от событий CDC, поступающих не по порядку.

Как работает интеграция с Kafka и CDC в Apache Doris?​

После поступления данных в FE все способы интеграции Apache Doris с Kafka и CDC следуют одному жизненному циклу: получение пакета, оформление его в транзакцию с меткой, фиксация после подтверждения и публикация на всех репликах.

  1. Потребитель получает пакет. Routine Load назначает задачу на BE, который читает данные из Kafka до достижения max_batch_rows (по умолчанию 20M), max_batch_size (по умолчанию 1 GB) или max_batch_interval (по умолчанию 60 s), в зависимости от того, какой предел будет достигнут первым. Flink CDC передаёт строки через HTTP Stream Load с передачей по частям; соединение остаётся открытым на протяжении интервала между контрольными точками.
  2. Пакет становится транзакцией. BE открывает транзакцию с Label. Routine Load генерирует метку, а Flink Doris Connector включает в неё идентификатор контрольной точки. В обоих случаях повторная отправка с той же меткой не создаёт дубликатов.
  3. Фиксация ожидает подтверждения. Routine Load фиксирует транзакцию сразу после загрузки пакета и продвигает смещение Kafka только при успехе. Flink CDC останавливается в состоянии PRECOMMITTED; барьер следующей контрольной точки вызывает txn_operation: commit, а сбой контрольной точки вызывает abort, который отбрасывает подготовленные данные.
  4. Публикация делает строки видимыми. Фоновый процесс отправляет PublishVersion каждой реплике. Когда транзакция переходит в состояние VISIBLE, новые строки становятся доступны для запросов. Данные всех tablet одной транзакции становятся видимыми атомарно.
  5. Ошибки приостанавливают задание без потери данных. Если пакет Routine Load превышает max_error_number (по умолчанию 0), задание переходит в состояние PAUSED, а FE сохраняет URL с информацией о последней ошибке. Возобновите работу после исправления исходных данных; смещения останутся прежними.

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

CREATE TABLE events (
  user_id BIGINT, event_type STRING, ts DATETIME
)
DUPLICATE KEY(user_id)
DISTRIBUTED BY HASH(user_id) BUCKETS 8;

CREATE ROUTINE LOAD events_kafka ON events
COLUMNS(user_id, event_type, ts)
PROPERTIES (
  "format" = "json",
  "jsonpaths" = "[\"$.user_id\",\"$.event_type\",\"$.ts\"]"
)
FROM KAFKA (
  "kafka_broker_list" = "localhost:9092",
  "kafka_topic" = "events",
  "property.kafka_default_offsets" = "OFFSET_BEGINNING"
);

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

+--------+--------------+---------+----------------------+
| Id     | Name         | State   | Statistic            |
+--------+--------------+---------+----------------------+
| 100123 | events_kafka | RUNNING | {"loadedRows":4821,..|
+--------+--------------+---------+----------------------+

SHOW ROUTINE LOAD FOR events_kafka показывает задание в состоянии RUNNING, последнее прочитанное смещение в каждой партиции и количество загруженных строк. PAUSE, RESUME и STOP работают так же, как для любого другого задания Doris.

Для CDC замените потребитель на Flink Doris Connector с Flink CDC (одно задание Flink, источник MySQL, sink Doris, sink.enable-2pc = true). Схема и столбец последовательности задаются на стороне Doris:

CREATE TABLE orders (
  order_id BIGINT, status STRING, updated DATETIME
)
UNIQUE KEY(order_id) DISTRIBUTED BY HASH(order_id) BUCKETS 8
PROPERTIES ("function_column.sequence_col" = "updated");

Строка с sequence_col защищает от ситуации, когда задержавшееся событие CDC приходит после более свежего: версия с более старым значением updated уступает версии с более новым, и результат не зависит от порядка поступления событий.

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

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

Подходит

  • Топик Kafka с данными CSV или JSON, которые нужно загрузить в Doris без дополнительного сервиса. Routine Load обеспечивает кратчайший путь.
  • Передача CDC из MySQL, Postgres или Oracle в Doris с гарантией exactly-once и передачей изменений DDL. Используйте Flink Doris Connector с Flink CDC.
  • В вашей инфраструктуре уже работает Kafka Connect, а данные имеют формат Avro, Protobuf или Debezium. Doris Kafka Connector устраняет проблему несовместимости форматов.
  • Панели мониторинга в реальном времени на базе таблиц Unique Key, которым нужна корректная обработка записей, поступающих не по порядку. Используйте любой из описанных выше способов вместе с function_column.sequence_col.

Не подходит

  • Передача Avro из Kafka через Routine Load. Routine Load читает только CSV и JSON. Преобразуйте данные до отправки или используйте Doris Kafka Connector с соответствующим конвертером.
  • Отдельный INSERT INTO ... VALUES для каждого события из микросервиса. Эта задача не связана с Kafka или CDC; используйте Group Commit, чтобы BE объединял небольшие записи в пакеты на стороне сервера.
  • Routine Load для топика с коротким сроком хранения данных в Kafka. Если срок хранения составляет 24 h, а задание приостановлено на два дня, сохранённое смещение оказывается раньше самого раннего доступного смещения, и Kafka отклоняет возобновление с ошибкой "out of range". Увеличьте срок хранения или используйте коннектор с буферизацией.
  • Мониторинг отставания по смещениям только через FE. Сейчас FE не предоставляет метрику отставания (открытая задача). Собирайте метрики из экспортера consumer-lag на стороне Kafka или следите за скоростью обработки на стороне BE.
  • Stream Load 2PC для таблиц Unique Key с Merge-on-Write в облачных развёртываниях / развёртываниях с разделением хранения и вычислений: BE отклоняет такое сочетание, поэтому в этих средах механизм exactly-once во Flink Connector переходит на дедупликацию на основе Label вместе со столбцом последовательности. В локально развёрнутых кластерах для MoW сохраняется полноценный механизм Stream Load 2PC.

Производительность в цифрах​

Из публикации «Как запись данных из Flink в Apache Doris в реальном времени остаётся быстрой и обеспечивает exactly-once» (блог к выпуску за июль 2022 года): при 30 параллельных загрузках Stream Load сквозная задержка оставалась ниже одной секунды; 20 задач Flink обеспечивали стабильную обработку входящего потока со скоростью около 100,000 событий в секунду; при большом числе параллельных операций нагрузка на CPU BE снизилась примерно на 25% по сравнению с предыдущим способом загрузки. Источник: блог Apache Doris.

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

Report issue