Настройка интервалов триггера структурированной потоковой передачи

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

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

Внимание

Azure Databricks рекомендует задать режим триггера, который балансирует задержку и затраты для вашего варианта использования. В противном случае могут возникнуть непредвиденные затраты на хранилище от поставщика облачных служб. Дополнительные сведения см. в разделе "Управление затратами на облачное хранилище ".

Обзор режимов триггера

В следующей таблице перечислены режимы триггера, доступные в структурированной потоковой передаче.

Режим триггера	Пример синтаксиса (Python)	Лучше всего для
Не указано (по умолчанию)	N/A	Потоковая передача общего назначения с задержкой 3–5 секунд. Эквивалентно обработке триггераTime с интервалами в 0 мс. Потоковая обработка выполняется непрерывно до тех пор, пока будут поступать новые данные.
Время обработки	.trigger(processingTime='10 seconds')	Балансировка затрат и производительности. Снижает нагрузку, предотвращая проверку данных системой слишком часто.
Доступно сейчас	.trigger(availableNow=True)	Плановая инкрементная пакетная обработка. Обрабатывает столько данных, сколько доступно во время активации задания потоковой передачи.
Режим реального времени	.trigger(realTime='5 minutes')	Для оперативных нагрузок с сверхнизкой задержкой, требующих обработки в доли секунды, таких как обнаружение мошенничества или персонализация в режиме реального времени. Общедоступная предварительная версия. Значение "5 минут" указывает продолжительность микробатча. Используйте 5 минут, чтобы свести к минимуму затраты на пакет, такие как компиляция запросов.
Непрерывный	.trigger(continuous='1 second')	Не поддерживается. Это экспериментальная функция, включенная в OSS Spark. Вместо этого используйте режим реального времени.

:::note Бессерверные вычисления

На бессерверных вычислительных ресурсах поддерживаются только Trigger.AvailableNow() и Trigger.Once(). Databricks рекомендует Trigger.AvailableNow().

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

См. ограничения потоковой передачи.

:::

processingTime: интервалы триггера, зависящие от времени

Структурированная потоковая передачаStructured Streaming' относится к интервалам времени на триггерах как к "микропакетам фиксированного интервала". С помощью ключевого слова processingTime укажите длительность в виде строки, например .trigger(processingTime='10 seconds').

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

AvailableNow: добавочная пакетная обработка

Внимание

В Databricks Runtime 11.3 LTS и выше Trigger.Once устарело. Используйте Trigger.AvailableNow для всех рабочих нагрузок инкрементной пакетной обработки.

Параметр AvailableNow триггера использует все доступные записи в качестве инкрементного пакета с возможностью настройки размера пакета, например, с помощью maxBytesPerTrigger. Параметры размера зависят от источника данных.

Поддерживаемые источники данных

Azure Databricks поддерживает использование Trigger.AvailableNow для инкрементной обработки пакетов из множества структурированных источников потоковой передачи. В следующей таблице приведена минимальная поддерживаемая версия Databricks Runtime, необходимая для каждого источника данных:

Источник	Минимальная версия среды выполнения Databricks
Источники файлов (JSON, Parquet и т. д.)	9.1 LTS
Delta Lake	10.4 долгосрочная поддержка (LTS)
Автозагрузчик	10.4 долгосрочная поддержка (LTS)
Apache Kafka	10.4 долгосрочная поддержка (LTS)
Кинезис	Версия 13.1

realTime: операционные нагрузки с ультранизкой задержкой

Режим реального времени работы для структурированной потоковой передачи достигает сквозной задержки менее 1 секунды на конечном этапе и в распространенных случаях около 300 мс. Дополнительные сведения о том, как эффективно настраивать и использовать режим реального времени, см. в режиме реального времени в структурированной потоковой передаче.

Apache Spark имеет дополнительный интервал триггера, известный как непрерывная обработка. Этот режим классифицируется как экспериментальный с версии Spark 2.3. Azure Databricks не поддерживает или не рекомендует этот режим. Вместо этого используйте режим реального времени для вариантов использования с низкой задержкой.

Примечание.

Режим непрерывной обработки на этой странице не связан с непрерывной обработкой в декларативных конвейерах Spark Lakeflow.

Управление затратами на облачное хранилище

По умолчанию, если режим триггера не задан, структурированная потоковая передача задает режим processingTime триггера и интервал 0, который проверяет наличие новых данных каждые несколько миллисекунд. Это может создать большой объем вызовов API облачного хранилища в день и привести к непредвиденным затратам от поставщика облачных служб.

Azure Databricks рекомендует настроить режим триггера, соответствующий требованиям к задержке и затратам. Дополнительные сведения о настройке интервала триггера на основе времени см. в разделе processingTime "Настройка интервала триггера на основе времени".

Изменение интервалов триггера между запусками

Интервал триггера можно изменить, используя ту же самую контрольную точку.

Поведение при изменении интервалов

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

• От интервала AvailableNowна основе времени до: микропакет может обрабатываться как добавочный пакет до всех доступных записей.
• От AvailableNow до временного интервала: обработка может продолжаться для всех записей, доступных в момент выполнения последней AvailableNow работы.

Восстановление после сбоев запросов

Если попытаться восстановиться после сбоя запроса с инкрементальным пакетом, изменение интервала триггера не решает проблему. Предыдущий неудачный пакет должен завершиться, так как структурированная потоковая передача требует идемпотентных микропакетов. См. семантику отказоустойчивости для Apache Spark.

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