Настройка режима реального времени

На этой странице описано, как запускать запросы в режиме реального времени напрямую с помощью API Structured Streaming на классических вычислительных ресурсах. Для пошагового руководства см. Руководство: Запуск рабочей нагрузки потоковой передачи в режиме реального времени. Для получения концептуальной информации о режиме реального времени см. Концепции режима реального времени.

Режим реального времени в конвейерах Lakeflow поддерживает управляемые конвейеры на классических и бессерверных вычислениях с отдельными требованиями к выполнению, источнику, приемнику и оператору.

Необходимые условия

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

  • Используйте классические вычисления. Поддерживаются выделенные и стандартные режимы доступа. Стандартный режим доступа поддерживается только для Python. Прямая конфигурация запросов на этой странице не поддерживается в бессерверных ноутбуках или заданиях. Для бессерверной обработки в реальном времени используйте бессерверный режим реального времени в конвейерах Lakeflow.
  • Используйте Databricks Runtime 16.4 LTS и более поздних версий.
  • Отключите автомасштабирование.
  • Отключите Photon.
  • Задайте для параметра spark.databricks.streaming.realTimeMode.enabled значение true.
  • Отключите точечные экземпляры, чтобы избежать прерываний.

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

Инструкции по созданию и настройке классических вычислений см. в справочнике по конфигурации вычислений.

Соединения между потоками

Для внутренних соединений между потоками необходима дополнительная настройка для режима реального времени. Внешние соединения между потоками не поддерживаются. См. Объединение потоков.

Это важно

Чтобы запустить поток для потокового соединения в режиме реального времени с несколькими другими потоками в одном кластере, необходимо использовать Databricks Runtime 18 LTS и выше.

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

Python

spark.conf.set("spark.databricks.streaming.realTimeMode.streamStreamJoin.enabled", "true")
spark.conf.set("spark.sql.streaming.join.stateFormatVersion", "4")
spark.conf.set("spark.sql.streaming.join.stateFormatV4.enabled", "true")
spark.conf.set("spark.sql.streaming.stateStore.rocksdb.mergeOperatorVersion", "2")
spark.conf.set("spark.sql.streaming.realTimeMode.controlMessage.enabled", "true")

Scala

spark.conf.set("spark.databricks.streaming.realTimeMode.streamStreamJoin.enabled", "true")
spark.conf.set("spark.sql.streaming.join.stateFormatVersion", "4")
spark.conf.set("spark.sql.streaming.join.stateFormatV4.enabled", "true")
spark.conf.set("spark.sql.streaming.stateStore.rocksdb.mergeOperatorVersion", "2")
spark.conf.set("spark.sql.streaming.realTimeMode.controlMessage.enabled", "true")

SQL

SET spark.databricks.streaming.realTimeMode.streamStreamJoin.enabled = true;
SET spark.sql.streaming.join.stateFormatVersion = 4;
SET spark.sql.streaming.join.stateFormatV4.enabled = true;
SET spark.sql.streaming.stateStore.rocksdb.mergeOperatorVersion = 2;
SET spark.sql.streaming.realTimeMode.controlMessage.enabled = true;

Конфигурация запроса

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

Python

query = (
    spark.readStream
        .format("kafka")
        .option("kafka.bootstrap.servers", broker_address)
        .option("subscribe", input_topic)
        .load()
        .writeStream
        .format("kafka")
        .option("kafka.bootstrap.servers", broker_address)
        .option("topic", output_topic)
        .option("checkpointLocation", checkpoint_location)
        .outputMode("update")
        # In PySpark, the realTime trigger requires specifying the interval.
        .trigger(realTime="5 minutes")
        .start()
)

Scala

import org.apache.spark.sql.streaming.Trigger

val readStream = spark.readStream
      .format("kafka")
      .option("kafka.bootstrap.servers", brokerAddress)
      .option("subscribe", inputTopic).load()
      .writeStream
      .format("kafka")
      .option("kafka.bootstrap.servers", brokerAddress)
      .option("topic", outputTopic)
      .option("checkpointLocation", checkpointLocation)
      .outputMode("update")
      .trigger(Trigger.RealTime())
      // Trigger.RealTime can also accept an argument specifying the batch duration.
      // For example, this code sets a batch duration of 5 minutes:
      // .trigger(Trigger.RealTime("5 minutes"))
      .start()

Оценка размеров вычислительных ресурсов

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

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

Примеры вычислений слотов

Тип конвейера Конфигурация Обязательные слоты
Одноэтапное без отслеживания состояния (источник Kafka + приемник) maxPartitions = 8 8 слотов
Двухэтапное состояние (источник Kafka + перемещение) maxPartitions = 8, разбиение на части = 20 28 слотов (8 + 20)
Три этапа (источник Kafka + перетасовка + перераспределение) maxPartitions = 8, два этапа перетасовки 20 каждый 48 слотов (8 + 20 + 20)

Если не задано maxPartitions, используйте количество секций в разделе Kafka.

Дополнительные ресурсы

  • Последнее обновление: