Справочник по режиму реального времени

Поддерживаемые языки

Режим реального времени поддерживает Scala, Java и Python.

Типы вычислений

Режим реального времени поддерживает следующие типы вычислений:

Тип вычисления	Поддерживается
Выделенный (ранее— один пользователь)	✓
Стандартный (ранее: общий)	✓ (только Python)
Декларативные конвейеры Lakeflow Spark Classic	Не поддерживается (см. примечание ниже)
Декларативные конвейеры Lakeflow Spark без сервера	Не поддерживается (см. примечание ниже)
Серверлесс	Не поддерживаются

Note

Режим реального времени не поддерживается как прямой тип триггера структурированной потоковой передачи в декларативных конвейерах Spark Lakeflow. Однако Декларативные конвейеры Spark Lakeflow поддерживают режим реального времени с помощью конфигурации уровня конвейера. См. раздел "Использование режима реального времени" в декларативных конвейерах Spark Lakeflow.

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

Режимы выполнения

Режим реального времени поддерживает только режим обновления:

Режим выполнения	Поддерживается
Режим обновления	✓
Append mode	Не поддерживаются
Полный режим	Не поддерживаются

Источники и ресиверы

Режим реального времени поддерживает следующие источники и приемники:

Источник или приемник	Источник	Как приемник
Apache Kafka	✓	✓
Центры событий (с помощью соединителя Kafka)	✓	✓
Kinesis	✓ (только режим EFO)	Не поддерживаются
AWS MSK	✓	Не поддерживаются
Дельта	Не поддерживаются	Не поддерживаются
Google Pub/Sub	Не поддерживаются	Не поддерживаются
Apache Pulsar	Не поддерживаются	Не поддерживаются
Произвольные приемники (с помощью forEachWriter)	Неприменимо	✓

Операторы

Режим реального времени поддерживает большинство структурированных операторов потоковой передачи:

Операции без отслеживания состояния

Оператор	Поддерживается
Отбор	✓
Проекция	✓

UDFs

Оператор	Поддерживается
Scala UDF	✓ (с некоторыми ограничениями)
UDF на языке Python	✓ (с некоторыми ограничениями)

Aggregation

Оператор	Поддерживается
sum	✓
количество	✓
max	✓
min	✓
avg	✓
Агрегатные функции	✓

Windowing

Оператор	Поддерживается
Tumbling	✓
Sliding	✓
Session	Не поддерживаются

Дедупликация

Оператор	Поддерживается
удалить дубликаты	✓ (состояние неограниченное)
удалитьДубликатыВнутриВодяногоЗнака	Не поддерживаются

Потоковая передача в присоединение к таблице

Оператор	Поддерживается
Присоединение к широковещательной таблице (таблица должна быть небольшой)	✓
Поток для присоединения к потоку	Не поддерживаются
(плоская)MapGroupsWithState	Не поддерживаются
transformWithState	✓ (с некоторыми различиями)
union	✓ (с некоторыми ограничениями)
forEach	✓
forEachBatch	Не поддерживаются
mapPartitions	Не поддерживается (см. ограничение)

Особые соображения

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

transformWithState в режиме реального времени

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

Однако API работает по-разному в режиме реального времени, чем в микропакетных запросах.

• Режим реального времени вызывает метод для каждой handleInputRows(key: String, inputRows: Iterator[T], timerValues: TimerValues) строки.
  • Итератор inputRows возвращает одно значение. Микро пакетный режим вызывает его один раз для каждого ключа, итератор inputRows возвращает все значения ключа в микропакете.
  • Учет этой разницы при написании кода
• Таймеры времени событий не поддерживаются в режиме реального времени.
• transformWithStateInPandas не поддерживается в режиме реального времени. Вместо этого используйте API на основе transformWithState строк, который использует Row объекты, а не pandas DataFrames.
• В режиме реального времени таймеры задерживаются при запуске в зависимости от поступления данных:
  • Если таймер запланирован на 10:00:00, но данные не поступают, таймер не запускается немедленно.
  • Если данные поступают в 10:00:10, таймер запускается с задержкой в 10 секунд.
  • Если данные не поступают и долго выполняющийся пакет завершается, таймер запускается до завершения пакета.

Note

В Databricks Runtime 18.1 и ниже, если вы используете transformWithState и режим реального времени для Python с низкой пропускной способностью, менее 5 записей в секунду, может возникнуть увеличение задержки до нескольких сотен миллисекунд. Databricks рекомендует выполнить обновление до Databricks Runtime 18.2 и более поздних версий.

Python определяемые пользователем функции в режиме реального времени

Databricks поддерживает большинство пользовательских функций Python в режиме реального времени:

Стейтлесс

Тип UDF	Поддерживается
Python скалярные скалярные функции UDF (Узер-определяемые скалярные функции — Python)	✓
Скалярный UDF со стрелками	✓
Скалярная UDF Pandas (определяемые пользователем функции Pandas)	✓
Стрелочная функция (mapInArrow)	✓
Функция Pandas (Map)	✓

Группирование с сохранением состояния (UDAF)

Тип UDF	Поддерживается
transformWithState (только Row интерфейс)	✓
transformWithStateInPandas	Не поддерживается. Вместо этого используйте API на основе transformWithState строк, который использует Row объекты, а не pandas DataFrames. Дополнительные сведения смtransformWithStateInPandas. не в поддержке.
applyInPandasWithState	Не поддерживаются

Бесcостояние группирование (UDAF)

Тип UDF	Поддерживается
apply	Не поддерживаются
applyInArrow	Не поддерживаются
applyInPandas	Не поддерживаются

Функции таблиц

Тип UDF	Поддерживается
UDTF (Python определяемые пользователем функции таблиц (UDTFs))	Не поддерживаются
UC UDF	Не поддерживаются

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

• Чтобы свести к минимуму задержку, задайте размер пакета со стрелкой (spark.sql.execution.arrow.maxRecordsPerBatch) значение 1.
  • Компромисс. Эта конфигурация оптимизирует задержку за счет пропускной способности. Для большинства рабочих нагрузок этот параметр рекомендуется.
  • Увеличьте размер блока только в том случае, если для обработки объема входных данных требуется более высокая пропускная способность, принимая потенциальное увеличение задержки.
• Пользовательские функции (UDF) Pandas и функции плохо работают с размером пакета Arrow, равным 1.
  • Если вы используете функции pandas UDF или другие функции, задайте размер пакета Arrow на более высокое значение (например, 100 или выше).
  • Это означает более высокую задержку. Databricks рекомендует использовать UDF или функцию со стрелкой, если это возможно.
• transformWithStateInPandas не поддерживается в режиме реального времени. Вместо этого используйте API на основе transformWithState строк, который использует Row объекты, а не pandas DataFrames. Примеры transformWithStateInPandas/c1 и для примера рабочего Python с помощью API на основе строк.
• Для рабочих нагрузок с учетом задержки с помощью определяемых пользователем функций Databricks рекомендуется использовать выделенный режим доступа. В стандартном режиме доступа затраты на изоляцию безопасности могут замедлить производительность UDF.