Оптимизация запросов потоковой передачи без отслеживания состояния

На этой странице описываются функции оптимизации, доступные для бессостояточных запросов потоковой передачи в Databricks Runtime 18.0 и выше.

Запросы структурированной потоковой передачи, не отслеживающие состояние, обрабатывают данные без сохранения промежуточного состояния. Эти запросы не используют операторы с сохранением состояния, такие как агрегация потоков, dropDuplicates или объединения потоков. Примеры включают запросы, использующие потоково-статические соединения, MERGE INTO с таблицами Delta и другие операции, которые отслеживают обработанные строки от источника до приемника.

Адаптивное выполнение запросов и автооптимизатор перетасовки

Azure Databricks поддерживает адаптивное выполнение запросов (AQE) и автооптимизированную перетасовку (AOS) для потоковых запросов без состояния. Эти функции помогают оптимизировать рабочие нагрузки потоковой передачи, использующие статические соединения потока, MERGE INTO с таблицами Delta и аналогичными операциями.

Чтобы включить AQE для запросов потоковой передачи без отслеживания состояния, задайте для следующей конфигурации значение true. Эта функция включена по умолчанию:

spark.sql.adaptive.streaming.stateless.enabled true

Чтобы включить AOS для бессостояточных потоковых запросов, включите AQE и задайте следующую конфигурацию:

spark.sql.shuffle.partitions auto

Изменение секций перетасовки во время перезапуска запроса

Запросы потоковой передачи без отслеживания состояния поддерживают изменение количества секций перетасовки при перезапуске запроса. Это позволяет настроить параллелизм для учета изменяющихся объемов входных данных.

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

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

spark.sql.shuffle.partitions <number>