Nota
El acceso a esta página requiere autorización. Puede intentar iniciar sesión o cambiar directorios.
El acceso a esta página requiere autorización. Puede intentar cambiar los directorios.
Esta página describe el soporte para consultas en modo en tiempo real que utilizan directamente las APIs de Streaming Estructurado. Para tuberías gestionadas, véase el modo en tiempo real en las tuberías Lakeflow, que tiene requisitos separados en tiempo de ejecución, fuente, sumidero y operador.
Idiomas compatibles
El modo en tiempo real admite Scala, Java y Python.
Tipos de computación
El modo en tiempo real admite los siguientes tipos de proceso:
| Tipo de proceso | Soportado |
|---|---|
| Dedicado (anteriormente: usuario único) | ✓ |
| Estándar (anteriormente: compartido) | ✓ (solo Python) |
| Canalizaciones de Lakeflow en el modelo clásico | Soportado mediante la configuración de la tubería |
| Canalizaciones de Lakeflow en sin servidor | Soportado mediante la configuración de la tubería |
| Cuadernos y trabajos sin servidor | No soportado para consultas directas de streaming estructurado en tiempo real |
Para cargas de trabajo sensibles a la latencia con UDF, Databricks recomienda usar el modo de acceso dedicado. Consulte Funciones de tabla.
Modos de ejecución
El modo en tiempo real solo admite el modo de actualización:
| Modo de ejecución | Soportado |
|---|---|
| Modo de actualización | ✓ |
| Append mode | No soportado |
| Modo completo | No soportado |
Orígenes y sumideros
El modo en tiempo real admite los siguientes orígenes y receptores:
| Origen o receptor | Como origen | Como sumidero |
|---|---|---|
| Apache Kafka | ✓ | ✓ |
| Event Hubs (mediante el conector de Kafka) | ✓ | ✓ |
| Kinesis | ✓ (Modo EFO recomendado; sondeo en Databricks Runtime 19 y superior) | No soportado |
| AWS MSK | ✓ | No soportado |
| Delta | No soportado | No soportado |
| Google Pub/Sub | No soportado | No soportado |
| Apache Pulsar | No soportado | No soportado |
Receptores arbitrarios (mediante forEachWriter) |
No es aplicable | ✓ |
Operadores
El modo en tiempo real admite la mayoría de los operadores de streaming estructurados:
Operaciones sin estado
| Operador | Soportado |
|---|---|
| Selection | ✓ |
| Proyección | ✓ |
mapPartitions |
No compatible (consulte limitación) |
| Union | ✓ (con algunas limitaciones) |
UDFs
| Operador | Soportado |
|---|---|
| Scala UDF | ✓ (UDFs escalares). No se soportan funciones agrupadas, como flatMapGroups y mapGroupsWithState, |
| Funciones Definidas por el Usuario (UDF) de Python | ✓ (con algunas limitaciones) |
Agregación
| Function | Soportado |
|---|---|
| sum | ✓ |
| conteo | ✓ |
| max | ✓ |
| min | ✓ |
| avg | ✓ |
| Funciones de agregación | ✓ |
Windowing
| Operador | Soportado |
|---|---|
| Saltos de tamaño constante | ✓ |
| Deslizante | ✓ |
| Sesión | No soportado |
Eliminación de duplicados
| Operador | Soportado |
|---|---|
| dropDuplicates | ✓ |
| EliminarDuplicadosDentroDeMarcaDeAgua | ✓ |
Transmisión a combinación de tabla
| Operador | Soportado |
|---|---|
| Combinación interna | ✓ |
| Combinación externa | ✓ |
| Combinación de tabla de difusión (tamaño de tabla de 10 mb o menos) | ✓ |
| Combinación de tabla (sin difusión) | No soportado |
Transmisión a combinación de secuencias
| Operador | Soportado |
|---|---|
| Combinación interna | ✓ (Databricks Runtime 18 LTS y versiones posteriores, con algunas configuraciones) |
| Combinación externa | No soportado |
Nota:
Para usar la secuencia para transmitir combinaciones en modo en tiempo real, debe establecer configuraciones adicionales de Spark. Para obtener más información sobre las configuraciones y para los requisitos para ejecutar varios flujos, consulte Stream to stream joins (Secuencias para transmitir combinaciones).
Operador con estado arbitrario
| Operador | Soportado |
|---|---|
| (plano)MapGroupsWithState | No soportado |
| transformWithState | ✓ (con algunas diferencias) |
Receptores definidos por el usuario
| Receptor | Soportado |
|---|---|
| forEach | ✓ |
| forEachBatch | No soportado |
Consideraciones especiales
Algunos operadores y características tienen consideraciones o diferencias específicas cuando se usan en modo en tiempo real.
transformWithState en modo en tiempo real
Para crear aplicaciones con estado personalizadas, Databricks admite transformWithState, una API en Apache Spark Structured Streaming. Consulta Construir una aplicación con estado personalizada para transformWithState más información sobre la API y fragmentos de código.
transformWithState Soporta tanto procesadores síncronos como asíncronos en modo en tiempo real. Para los requisitos de procesamiento asincrónico, véase Procesamiento asincrónico con transformWithState.
Sin embargo, la API se comporta de forma diferente en modo en tiempo real que en las consultas de microproceso.
- Para procesadores síncronos, el modo en tiempo real llama al
handleInputRows(key: String, inputRows: Iterator[T], timerValues: TimerValues)método para cada fila.- El
inputRowsiterador devuelve un valor único. El modo micro-batch lo llama una vez para cada clave, y elinputRowsiterador devuelve todos los valores de una clave en el micro-batch. - Ten en cuenta esta diferencia al escribir tu código.
- El
- Para procesadores asíncronos, el modo en tiempo real agrupa filas que llegan juntas y llama
handleInputRowsuna vez por cada clave del grupo, con todas las filas de esa clave. Como el modo en tiempo real ejecuta lotes de larga duración, la misma clave puede recibir muchas llamadas dentro de un mismo lote. - Los temporizadores de eventos no son compatibles con el modo de tiempo real.
-
transformWithStateInPandasno se admite en modo en tiempo real. Use la API basada entransformWithStatefilas en su lugar, que usaRowobjetos en lugar de dataFrames de pandas. - En modo en tiempo real, los temporizadores se retrasan al activarse en función de la llegada de datos:
- Si un temporizador está programado para las 10:00:00, pero no llega ningún dato, el temporizador no se activa inmediatamente.
- Si los datos llegan a las 10:00:10, el temporizador se activa con un retraso de 10 segundos.
- Si no llegan datos y el lote de ejecución prolongada finaliza, el temporizador se activa antes de que finalice el lote.
Nota:
En Databricks Runtime 18.1 y versiones posteriores, si usa transformWithState y el modo en tiempo real para Python con un rendimiento bajo, menos de 5 registros por segundo, es posible que vea un aumento de las latencias de hasta unos cientos de milisegundos. Databricks recomienda actualizar a Databricks Runtime 18.2 y versiones posteriores para resolverlo.
Python UDF en modo en tiempo real
Databricks admite la mayoría de Python funciones definidas por el usuario (UDF) en modo en tiempo real:
Sin estado
| Tipo de UDF | Soportado |
|---|---|
| Python UDF escalar (Python funciones escalares definidas por el usuario (UDF) | ✓ |
| UDF escalar de flecha | ✓ |
| UDF escalar de Pandas (funciones definidas por el usuario de Pandas) | ✓ |
Función Arrow (mapInArrow) |
No compatible (consulte limitación) |
| Función Pandas (Mapa) | No compatible (consulte limitación) |
Agrupación con estado (UDAF)
| Tipo de UDF | Soportado |
|---|---|
transformWithState (solo Row interfaz) |
✓ |
transformWithStateInPandas |
No está soportado. Use la API basada en transformWithState filas en su lugar, que usa Row objetos en lugar de dataFrames de pandas. Consulte transformWithStateInPandas no compatible para obtener más información. |
applyInPandasWithState |
No soportado |
Agrupación no con estado (UDAF)
| Tipo de UDF | Soportado |
|---|---|
apply |
No soportado |
applyInArrow |
No soportado |
applyInPandas |
No soportado |
Funciones de tabla
| Tipo de UDF | Soportado |
|---|---|
| UDTF (Python funciones de tabla definidas por el usuario (UDTFs)) | No soportado |
| UC UDF | No soportado |
Hay varios puntos que se deben tener en cuenta al usar Python UDF en modo en tiempo real:
- En Databricks Runtime 19 y superiores, el modo en tiempo real agrupa las filas de entrada por tiempo para todos los UDF que el modo en tiempo real soporta. Cada lote recoge las filas que llegan en una ventana de tiempo corta (10 milisegundos), o hasta alcanzar el tamaño máximo del lote, y luego las envía a la UDF. Esto equilibra latencia y rendimiento, así que no necesitas poner el tamaño de lote de Arrow (
spark.sql.execution.arrow.maxRecordsPerBatch) en 1. - En Databricks Runtime 18 LTS e inferiores, las filas se agrupan solo por conteo:
- Para minimizar la latencia, establezca el tamaño del lote de flecha () en
spark.sql.execution.arrow.maxRecordsPerBatch1. Esta configuración optimiza la latencia a costa del rendimiento. Aumente el tamaño del lote solo si se requiere un mayor rendimiento para dar cabida al volumen de entrada, aceptando el posible aumento de la latencia. - Los UDF de Pandas no rinden bien con un lote de Arrow de tamaño 1. Si usas UDFs de pandas, pon el tamaño del lote de Arrow a un valor superior (por ejemplo, 100 o más). Esto implica una mayor latencia. Databricks recomienda usar un UDF Arrow si es posible.
- Para minimizar la latencia, establezca el tamaño del lote de flecha () en
-
transformWithStateInPandasno se admite en modo en tiempo real. Use la API basada entransformWithStatefilas en su lugar, que usaRowobjetos en lugar de dataFrames de pandas. ConsultetransformWithStateInPandasno se admite y Ejemplos de modo en tiempo real para ver un ejemplo de Python de trabajo mediante la API basada en filas. - Para cargas de trabajo sensibles a la latencia con UDF, Databricks recomienda usar el modo de acceso dedicado. En el modo de acceso estándar, la sobrecarga de aislamiento de seguridad podría ralentizar el rendimiento de UDF.