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

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

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

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

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

Тип вычисления	Поддерживается
Выделенный (ранее— один пользователь)	✓
Стандартный (ранее: общий)	✓ (только Python)
Конвейеры Lakeflow в классической версии	Не поддерживается как структурированная потоковая передача. Поддерживается с помощью конфигурации конвейера. См. раздел "Использование режима реального времени" в конвейерах Lakeflow.
Конвейеры Lakeflow на бессерверных серверах	Не поддерживается как структурированная потоковая передача. Поддерживается с помощью конфигурации конвейера. См. раздел "Использование режима реального времени" в конвейерах Lakeflow.
Серверлесс	Не поддерживаются

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

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

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

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

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

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

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

Операторы

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

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

Оператор	Поддерживается
Отбор	✓
Проекция	✓
mapPartitions	Не поддерживается (см. ограничение)
Союз	✓ (с некоторыми ограничениями)

UDFs

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

Aggregation

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

Windowing

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

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

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

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

Оператор	Поддерживается
Внутреннее соединение	✓
Внешнее соединение	✓
Присоединение к широковещательной таблице (размер таблицы составляет 10 мб или меньше)	✓
Присоединение к таблице (без трансляции)	Не поддерживаются

Поток для присоединения к потоку

Оператор	Поддерживается
Внутреннее соединение	✓ (Databricks Runtime 18 LTS и более поздних версий с некоторыми конфигурациями)
Внешнее соединение	Не поддерживаются

Note

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

Произвольный оператор с отслеживанием состояния

Оператор	Поддерживается
(плоская)MapGroupsWithState	Не поддерживаются
transformWithState	✓ (с некоторыми различиями)

Определяемые пользователем приемники

Sink	Поддерживается
forEach	✓
forEachBatch	Не поддерживаются

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

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

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

Для создания пользовательских приложений с отслеживанием состояния Databricks поддерживает transformWithState, API в Структурированной потоковой передаче Apache Spark. См. раздел «Создать пользовательское приложение transformWithStateс состоянием» для получения дополнительной информации об 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 скалярные определяемые пользователем функции (UDFs))	✓
Скалярный 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.