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

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

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

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

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

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

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

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

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

Это важно

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

В Databricks Runtime 18.2 и ниже структурированная потоковая передача не поддерживает следующие конфигурации для других режимов обработки, включая processingTime и availableNow.

Чтобы включить операции соединения между потоками в режиме реального времени, задайте следующие параметры конфигурации 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.execution.streaming.RealTimeTrigger

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(RealTimeTrigger.apply())
      // RealTimeTrigger can also accept an argument specifying the checkpoint interval.
      // For example, this code indicates a checkpoint interval of 5 minutes:
      // .trigger(RealTimeTrigger.apply("5 minutes"))
      .start()

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

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

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

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

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

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

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

  • Last updated on