Принципы работы и рекомендации по использованию Routine Load

1. Обзор

Routine Load непрерывно потребляет данные из Kafka и записывает их в Apache Doris. Создав задание Routine Load, можно автоматически подписаться на указанный топик Kafka. Основные возможности:

• Высокая доступность: Поддерживает непрерывное потребление данных Kafka в режиме 7x24 и автоматически возобновляет работу после сбоя.
• Низкая задержка: Сообщения Kafka становятся доступны для чтения в течение нескольких секунд.
• Семантика Exactly Once: Гарантирует потребление данных Kafka без потерь и дублирования, обеспечивая обработку каждого сообщения ровно один раз.

В этом документе подробно рассматриваются принципы реализации Routine Load, рекомендации для типичных сценариев и способы устранения распространённых неполадок. Это поможет быстро начать работу и эффективно использовать Routine Load.

Рекомендуемый порядок чтения:

1. Чтобы разобраться в механизме работы, прочитайте раздел Принципы реализации.
2. Чтобы оптимизировать производительность, прочитайте раздел Рекомендации по использованию.
3. Чтобы разобраться с неполадками, прочитайте раздел Устранение распространённых неполадок.

2. Принципы реализации

Данные в Kafka представляют собой поток, который Doris потребляет микропакетами. После создания задания Routine Load система разбивает его на несколько задач для параллельного выполнения в соответствии с настроенным уровнем параллелизма. Каждая задача потребляет данные из определённых партиций топика Kafka. Каждой задаче соответствует транзакция, а после завершения её выполнения создаётся новая задача для потребления следующего пакета данных.

В следующих разделах рассматриваются три аспекта:

• Механизм планирования заданий и задач
• Реализация семантики Exactly Once
• Процесс записи из одного потока в несколько таблиц

2.1 Планирование заданий и задач

Routine Load использует двухуровневую модель планирования:

Уровень планирования	Назначение
Планирование заданий	Разбиение на задачи, восстановление после сбоев и управление жизненным циклом
Планирование задач	Передача операций получения, преобразования и записи данных на узлы BE для выполнения

2.1.1 Планирование заданий

Автомат состояний задания

Состояние	Значение
NEED_SCHEDULE	Ожидает первого планирования или нуждается в повторном планировании
RUNNING	Потребление данных выполняется в штатном режиме
PAUSED	Приостановлено вручную или из-за исключения; возможно автоматическое возобновление
CANCELLED	Прекращено из-за неустранимой ошибки, например удаления базы данных или таблицы
STOPPED	Остановлено вручную и не может быть восстановлено

Поведение планировщика

В каждом цикле планирования (10 с) поток планировщика выполняет следующие действия в зависимости от состояния задания:

• NEED_SCHEDULE: Получает метаданные топика (количество партиций, начальные смещения), создаёт задачи в соответствии со следующей формулой и помещает их в needScheduleTasksQueue, где они ожидают обработки потоком планирования задач:

  taskNum = min(topic_partition_num,
                desired_concurrent_number,
                max_routine_load_task_concurrent_num)

• RUNNING: Периодически получает метаданные топика и немедленно выполняет повторное планирование, если количество партиций изменилось.

• PAUSED: Для обеспечения высокой доступности задания предусмотрен механизм автоматического возобновления. При непредвиденной приостановке поток планировщика Routine Load пытается автоматически возобновить задание. При неожиданном отказе Kafka или в других ситуациях, когда Kafka недоступна, этот механизм позволяет заданию загрузки продолжить работу после восстановления Kafka без ручного вмешательства. Обратите внимание: в трёх случаях задание не будет возобновлено автоматически:

  1. Пользователь вручную выполнил команду PAUSE ROUTINE LOAD.
  2. Обнаружены проблемы с качеством данных.
  3. Возникла ситуация, после которой автоматическое восстановление невозможно, например удалена база данных или таблица.

  Во всех остальных случаях для приостановленных заданий будут выполняться попытки автоматического возобновления.

• CANCELLED / STOPPED: Ресурсы освобождаются с задержкой.

2.1.2 Планирование задач

Условия планирования

Для планирования задачи должно выполняться одно из следующих условий:

• Задача ещё не дочитала партицию до конца, то есть в ней остаются данные для потребления. Это позволяет избежать напрасного расходования ресурсов.
• Если при предыдущем запуске уже был достигнут EOF, новый цикл планирования начинается только после того, как с момента начала предыдущего выполнения пройдёт время, превышающее max_batch_interval. Это позволяет накапливать данные в пакеты, когда скорость потребления выше скорости поступления, и предотвращает создание слишком большого количества мелких транзакций.

Стратегия балансировки нагрузки

Планировщик выбирает узлы BE для выполнения задач в следующем порядке:

1. Предпочтение отдаётся узлам BE, на которых в данный момент выполняется наименьшее количество задач.
2. Если на нескольких узлах BE выполняется одинаковое количество задач, предпочтение отдаётся узлам с кэшированным Kafka Consumer, чтобы сократить затраты на инициализацию.

Условия завершения пакета

Текущая задача завершается при выполнении любого из следующих условий:

• Истекло время, заданное параметром max_batch_interval.
• Достигнуто количество строк, заданное параметром max_batch_rows.
• Достигнут объём в байтах, заданный параметром max_batch_size.
• Прочитан Kafka EOF, то есть потребление достигло конца потока.

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

2.2 Семантика Exactly Once

Routine Load обеспечивает потребление данных Kafka без потерь и дублирования за счёт двух механизмов: долговременного сохранения прогресса потребления и проверки при фиксации транзакции.

2.2.1 Долговременное сохранение прогресса потребления

При фиксации транзакции каждая задача записывает прогресс потребления вместе с информацией о транзакции в журнал изменений FE и синхронизирует эти сведения со всеми FE Followers с помощью Berkeley DB JE. После переключения Master или перезапуска информация о прогрессе остаётся корректной.

2.2.2 Проверка при фиксации

Когда задание повторно планируется после ручной приостановки, переключения Master или изменения метаданных топика, две задачи могут некоторое время одновременно потреблять данные из одной партиции. Чтобы предотвратить повторную запись:

• Каждое задание хранит в памяти routineLoadTaskInfoList.
• Перед фиксацией транзакции задача проверяет, что она всё ещё находится в routineLoadTaskInfoList; в противном случае фиксация отклоняется.

2.3 Запись из одного потока в несколько таблиц

Запись из одного потока в несколько таблиц позволяет одному заданию Routine Load одновременно записывать данные в несколько целевых таблиц. Основной процесс выглядит следующим образом:

1. Этап планирования: Поскольку при создании задания невозможно полностью определить набор целевых таблиц, построение плана выполнения откладывается до момента выполнения, когда BE динамически запрашивает его у FE Master.
2. Кэширование данных: Сначала BE кэширует данные в локальном канале multi-table pipe. Если в кэше накопилось 200 записей или появились 5 новых таблиц, для которых ещё не запрашивался план выполнения, BE запрашивает план и выполняет его, предотвращая накопление необработанных данных.
3. Повторное использование планов выполнения: В пределах одной транзакции кэшированные планы выполнения используются повторно, а для каждой новой транзакции планы запрашиваются заново, чтобы учитывать актуальные метаданные.

3. Рекомендации по использованию

Параметры Routine Load по умолчанию подходят для подавляющего большинства сценариев. Ручная настройка требуется только в следующих трёх типичных случаях:

Сценарий	Рекомендуемые изменения параметров
Требуется низкая задержка	Уменьшить max_batch_interval относительно значения по умолчанию 60 с
Небольшой объём данных, ограниченные ресурсы	Уменьшить desired_concurrent_number
Требуется высокая пропускная способность	Увеличить max_batch_interval со значения по умолчанию 60 с до 120-180 с

4. Устранение распространённых неполадок

4.1 Накопление необработанных данных

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

Шаг 1: Проверьте состояние задач

Используйте SHOW ROUTINE LOAD\G, чтобы проверить состояние задач. Обратите внимание на следующее:

• Имеет ли State значение RUNNING. Если состояние другое, посмотрите причину в поле ReasonOfStateChanged.
• Есть ли сообщение об ошибке в OtherMsg.

Шаг 2: Определите, достигнут ли предел пропускной способности

По журналам BE определите, достигнут ли предел пропускной способности. Найдите записи consumer group done; значения left_time / left_rows / left_bytes показывают, какой порог был достигнут первым. В зависимости от этого можно увеличить max_batch_size или max_batch_rows:

consumer group done: 894fc32d5b9d3e93-7387a02da6dafd88. consume time(ms)=34004, received rows=2679540, received bytes=2147484043, eos: 0, left_time: 25996, left_rows: 17320460, left_bytes: -395, blocking get time(us): 949236, blocking put time(us): 28730419, id=69616a41fc064f1e-a93ff0ddd217f0a0, job_id=48121487, txn_id=61763720, label=ods_hq_market_unique_jobs_0-48121487-69616a41fc064f1e-a93ff0ddd217f0a0-61763720, elapse(s)=34

В приведённом примере left_bytes: -395 указывает, что обработка пакета завершилась за 34 секунды, поскольку был достигнут предел max_batch_size. В этом случае можно увеличить max_batch_size, чтобы за время max_batch_interval загрузить максимально полный пакет и повысить пропускную способность.

Шаг 3: Увеличьте параллелизм и пропускную способность

• Увеличьте desired_concurrent_number до количества партиций в топике.
• Умеренно увеличьте max_batch_interval (например, со 120 с до 180 с), max_batch_size или max_batch_rows, чтобы увеличить объём данных в каждой транзакции и каждом пакете и сократить накладные расходы на транзакции.

4.2 Непредвиденная приостановка задачи

В Routine Load встроен механизм автоматического возобновления, поэтому после большинства непредвиденных приостановок предпринимаются повторные попытки запуска. Если задача остаётся в состоянии PAUSED и не возобновляется автоматически, выполните SHOW ROUTINE LOAD и проверьте следующее:

• Не выполнялась ли вручную команда PAUSE ROUTINE LOAD.
• Нет ли проблем с качеством данных, например ошибок формата или отсутствующих полей.
• Не истёк ли срок хранения данных Kafka, что приводит к ошибке out of range.

5. Часто задаваемые вопросы

Вопрос 1: Как Routine Load гарантирует семантику Exactly Once?

За счёт двух механизмов: (1) При фиксации транзакции каждой задачи прогресс потребления записывается в журнал изменений FE и синхронизируется со всеми FE Followers через Berkeley DB JE. (2) Перед фиксацией транзакции задача проверяет, что она всё ещё находится в routineLoadTaskInfoList; в противном случае фиксация отклоняется, что предотвращает повторную запись при параллельном выполнении задач.

Вопрос 2: Возобновится ли автоматически задание в состоянии PAUSED?

После большинства непредвиденных приостановок работа возобновляется автоматически. Однако в следующих трёх случаях автоматического возобновления не будет: ручное выполнение команды PAUSE ROUTINE LOAD, проблемы с качеством данных и неустранимые ошибки, например удаление базы данных или таблицы.

Вопрос 3: От чего зависит количество задач?

Оно определяется по следующей формуле: taskNum = min(topic_partition_num, desired_concurrent_number, max_routine_load_task_concurrent_num).

Вопрос 4: Когда следует увеличивать max_batch_interval?

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

Вопрос 5: Что означает ошибка out of range в журнале?

Она означает, что срок хранения данных Kafka, которые предстоит прочитать, истёк и они были удалены в соответствии с политикой хранения Kafka. В результате смещение потребления стало недействительным. Необходимо проверить срок хранения данных в топике Kafka или заново задать начальное смещение потребления для Routine Load.