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

На этой странице описана поддержка запросов в режиме реального времени, которые напрямую используют API структурированного потокового потока. Для управляемых конвейеров см. режим реального времени в Lakeflow pipelines, где есть отдельные требования к времени выполнения, источнику, поглотителю и оператору.

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

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

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

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

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

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

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

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

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

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

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

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

Операторы

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

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

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

UDFs

Оператор	Поддерживается
Scala UDF	✓ (скалярные UDF). Сгруппированные функции, такие как flatMapGroups и mapGroupsWithState, не поддерживаются.
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 и фрагментах кода.

transformWithState Поддерживает как синхронные, так и асинхронные процессоры в режиме реального времени. Для требований к асинхронной обработке см. Асинхронная обработка с transformWithState.

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

• Для синхронных процессоров режим реального времени вызывает handleInputRows(key: String, inputRows: Iterator[T], timerValues: TimerValues) метод для каждой строки.
  • Итератор inputRows возвращает одно значение. Режим микропакета вызывает его один раз для каждого ключа, и inputRows итератор возвращает все значения ключа в микропакете.
  • Учитывайте это различие при написании кода.
• Для асинхронных процессоров режим реального времени объединяет строки, которые приходят вместе и вызывают handleInputRows по одному разу для каждого ключа в группе, используя все строки этого ключа. Поскольку режим реального времени выполняет длительные партии, один и тот же ключ может принимать множество вызовов в одной партии.
• Таймеры времени событий не поддерживаются в режиме реального времени.
• 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 в режиме реального времени следует учитывать несколько моментов.

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