Te presentamos Apache Spark® 4.3. Ya disponible en Databricks Runtime 19

Introducción

Apache Spark 4.3 incorpora procesamiento con estado al modo en tiempo real, logrando una latencia P99 de extremo a extremo inferior a 150 ms en tres benchmarks representativos con estado. Las agregaciones de streaming, la deduplicación y el procesamiento personalizado con estado ahora pueden ejecutarse en el modo de ejecución de baja latencia de Structured Streaming, ampliándolo a aplicaciones que necesitan recordar el contexto entre eventos.

El impacto va más allá de canalizaciones más rápidas. Para cargas de trabajo compatibles, ya no necesitas un motor de streaming separado solo para lograr una latencia a escala de milisegundos. Desde la detección de fraude hasta la actividad de usuarios actualizada continuamente y la supervisión de dispositivos, los desarrolladores pueden crear aplicaciones con estado utilizando las conocidas API de Structured Streaming en el mismo motor abierto que usan para el procesamiento por lotes, la analítica y la inteligencia artificial.

El streaming en tiempo real con estado es la principal novedad de Spark 4.3, pero las mejoras se extienden por todo el motor. La versión ofrece cinco beneficios clave:

• Crea aplicaciones en tiempo real con estado: Utiliza la agregación de streaming, la deduplicación y transformWithState en el modo en tiempo real para mantener el contexto a medida que llegan los eventos.
• Conserva el historial con SCD Type 2 Auto CDC: Realiza un seguimiento de cómo cambian los registros a lo largo del tiempo utilizando Python, SQL y Spark Connect, con menos lógica de procesamiento de cambios personalizada.
• Itera más rápido en datos pequeños: Project Feather se centra en la sobrecarga de ejecución innecesaria, mientras que el almacenamiento en caché nativo de Arrow admite una reutilización de datos más eficiente.
• Conecta más aplicaciones a Spark: Las mejoras en la observabilidad de Spark Connect, la integración con JDBC y Python facilitan el desarrollo y la operación de la ejecución remota.
• Expresa análisis más complejos directamente en SQL: Las transformaciones VARIANT, JSON_TABLE, las uniones temporales y una vista previa de marcas de tiempo en nanosegundos amplían el procesamiento nativo de datos anidados y orientados al tiempo.

En conjunto, estos cambios ayudan a los desarrolladores a utilizar un único motor abierto para explorar datos, mantener el contexto cambiante y potenciar aplicaciones analíticas y de inteligencia artificial.

Spark 4.3 también marca el inicio de un ciclo de lanzamiento de funciones trimestral, pasando de un lanzamiento cada seis meses a uno cada tres meses. Este ciclo más corto pone las mejoras terminadas en manos de los usuarios más rápidamente y ofrece a la comunidad oportunidades más frecuentes para enviar comentarios y dar forma a lo que viene.

Streaming en tiempo real: aplicaciones con estado y latencia a escala de milisegundos

Del procesamiento de eventos al mantenimiento del contexto

Una señal de fraude no solo depende de la transacción más reciente, sino también de la actividad que la precedió. La personalización debe reflejar el comportamiento reciente del usuario. La supervisión de dispositivos debe actualizar los contadores y agregados a medida que llegan nuevas lecturas. Estas aplicaciones necesitan tanto baja latencia como estado.

Spark 4.3 amplía el modo en tiempo real a estos patrones de procesamiento con estado, convirtiéndolo en un motor de streaming completo. Agrega compatibilidad con la agregación de streaming, dropDuplicates y Scala/Java transformWithState (SPARK-58635, SPARK-57228). Los desarrolladores pueden mantener un estado con clave y actualizar los resultados utilizando el modelo de programación existente de Structured Streaming, en lugar de adoptar otro motor únicamente para lograr una menor latencia.

Latencia P99 inferior a 150 ms en tres benchmarks con estado

En los benchmarks informados en SPARK-58884, las tres cargas de trabajo con estado representativas lograron una latencia P99 de extremo a extremo inferior a 150 milisegundos:

Estos resultados demuestran una ejecución de baja latencia en la deduplicación, la agregación basada en ventanas y el procesamiento personalizado con estado. Son mediciones de las tres cargas de trabajo probadas, no una garantía de latencia para cada canalización.

APIs conocidas, ejecución de baja latencia

El modo en tiempo real se basa en Structured Streaming en lugar de introducir un modelo de programación independiente. Los operadores con estado continúan utilizando el almacén de estados y los mecanismos de puntos de control existentes de Spark para garantizar la durabilidad y la recuperación.

Ejemplo: Mantener un recuento continuo para cada dispositivo. Dado un DataFrame de streaming respaldado por Kafka events con una columna device_id:

Si llegan eventos para los dispositivos A, A y B, Spark mantiene el estado y actualiza los recuentos a A=2 y B=1. Cuando llega otro evento para A, su estado se actualiza a A=3.

Un motor de ejecución diseñado para el flujo continuo de datos

La compatibilidad con el modo en tiempo real con estado requirió un cambio fundamental en la forma en que Spark planifica y conecta las etapas. Con una mezcla (shuffle) de bloqueo tradicional, una etapa productora debe finalizar antes de que pueda comenzar su etapa descendente. Ese modelo no puede admitir un flujo continuo de datos entre tareas de streaming de larga duración.

Spark 4.3 introduce la programación de etapas concurrentes nativas y mezclas canalizadas, lo que permite que las etapas productoras y consumidoras se ejecuten al mismo tiempo (SPARK-54699, SPARK-58076, SPARK-56664). Streaming Shuffle implementa esta capacidad para el modo en tiempo real, transfiriendo registros de forma incremental a la tarea descendente responsable de cada clave en lugar de esperar a que finalice toda la etapa productora.

Conceptualemente, la ejecución cambia de:

Etapa productora → materializar mezcla → etapa consumidora

a:

Etapa productora ⇄ mezcla de streaming ⇄ etapa consumidora

Spark 4.3 agrega PipelinedShuffleDependency, un nuevo tipo de dependencia de mezcla que permite a los consumidores comenzar a leer mientras los productores aún se están ejecutando (SPARK-58185). El planificador trata las etapas conectadas por estas dependencias como un grupo canalizado, una unidad de programación coordinada. Sus etapas se programan juntas y un fallo en una etapa hace que todo el grupo falle (SPARK-58263, SPARK-58398). En resumen, se ejecutan juntas y fallan juntas.

Debido a que todas las etapas de un grupo canalizado se ejecutan de forma concurrente, Spark admite el grupo solo cuando hay suficientes ranuras de tareas disponibles para todas sus tareas. De lo contrario, falla rápidamente en lugar de quedarse colgado.

Spark enruta cada mezcla a un administrador de mezclas según su tipo de dependencia:

• Una ShuffleDependency normal utiliza un administrador de mezclas de bloqueo, que materializa toda la salida de la mezcla antes de que las etapas descendentes la lean. SortShuffleManager es la implementación integrada.
• Un PipelinedShuffleDependency utiliza un administrador de shuffle con canalización, lo que hace que los resultados estén disponibles a medida que se producen. StreamingShuffleManager es la implementación integrada (SPARK-57141).

El tipo de dependencia define el contrato de ejecución; el administrador de shuffle proporciona el mecanismo de transferencia de datos. El planificador coordina las etapas canalizadas sin necesidad de saber qué carga de trabajo las creó.

Esta base va más allá del modo en tiempo real. La programación de etapas concurrentes, el intercambio de datos canalizados y la ejecución coordinada son capacidades genéricas del motor de Spark, no una infraestructura de streaming aislada. En Spark 4.3, permiten el procesamiento con estado de baja latencia. A más largo plazo, proporcionan una base para otras cargas de trabajo que necesitan ir más allá del modelo de ejecución tradicional etapa por etapa.

Para obtener información sobre el diseño y los escenarios de aplicación del modo en tiempo real, consulta la introducción al modo en tiempo real. Dicha publicación describe la implementación de Databricks; el alcance de las características y los resultados de las pruebas de rendimiento de Spark 4.3 analizados aquí son los indicados anteriormente.

Proyecto Feather: Iteración más rápida en datos pequeños

El desarrollo asistido por LLM suele seguir un ciclo de retroalimentación corto: generar una transformación, ejecutarla en una muestra, inspeccionar los resultados y refinar el código. Cada iteración puede procesar solo un archivo pequeño o unas pocas filas, pero la programación de tareas y el shuffle pueden costar más que el cálculo en sí.

El proyecto Feather se centra en esta sobrecarga de ejecución fija, haciendo que Spark sea más receptivo para explorar y validar código SQL y DataFrame en conjuntos de datos pequeños. Los desarrolladores pueden usar las mismas API a medida que pasan de una muestra a cargas de trabajo más grandes, ya sea que escriban el código ellos mismos o trabajen con un asistente de IA.

Spark 4.3 introduce la ejecución opcional de tarea única sin shuffle para fragmentos de consultas pequeñas de una sola partición elegibles (SPARK-57851). Cuando un análisis que cumple los requisitos informa que su salida ya está en una partición, las operaciones posteriores compatibles pueden evitar un shuffle innecesario.

Ejemplo: Ejecuta e inspecciona una pequeña agregación. Usa un pequeño conjunto de datos Parquet almacenado como un solo archivo, con una columna de categoría:

Seis casos:

category, Exchange, SPARK_LOCAL_CONNECT_REUSE=1, zip, counter_diff and jaro_winkler_similarity.

El resultado es un recuento por categoría. Para una consulta elegible, el plan físico puede omitir el elemento `Exchange` que de otro modo introduciría un shuffle. Esto admite el mismo flujo de trabajo de ejecución e inspección, ya sea que la transformación haya sido escrita por un desarrollador o sugerida por un asistente de código.

Con los límites de elegibilidad predeterminados, los análisis de archivos que califican leen un archivo pequeño no vacío que cabe en una sola partición, mientras que las relaciones locales en memoria que califican contienen hasta 1000 filas. Las formas de consulta compatibles incluyen análisis seleccionados, agregaciones, operaciones distinct, ventanas, ordenamientos y límites.

Feather también agrega almacenamiento en caché de conjuntos de datos opcional nativo de Arrow (SPARK-57268), almacenando lotes en caché en formato Arrow con compresión y estadísticas de columnas. Las entradas respaldadas por Arrow pueden evitar conversiones innecesarias, admitiendo flujos de trabajo que exploran repetidamente los mismos datos. Tanto la optimización de tarea única como la caché de Arrow son opcionales. Para activarlas, configura la propiedad estática spark.sql.cache.serializer=org.apache.spark.sql.execution.columnar.ArrowCachedBatchSerializer al iniciar una nueva aplicación de Spark. La repartición local en el proceso más amplia es un trabajo posterior previsto para la próxima versión (SPARK-57399).

Estas mejoras son solo el principio. La comunidad de Spark continuará invirtiendo en el rendimiento para datos pequeños en futuras versiones, reduciendo la sobrecarga de ejecución y acortando los bucles de retroalimentación para la exploración interactiva y el desarrollo asistido por LLM.

SCD tipo 2 en Auto CDC de código abierto: preserva el historial a medida que cambian los datos

Una tabla de estado actual puede decirte dónde vive un cliente hoy, pero no a qué región pertenecía cuando se realizó un pedido. Los informes, la auditoría y el análisis histórico a menudo necesitan ambas cosas. Spark 4.3 agrega compatibilidad integral con Slowly Changing Dimension (SCD) Tipo 2 a Auto CDC (SPARK-56249), para que las canalizaciones puedan rastrear el historial de cada registro en lugar de sobrescribir el valor más reciente con cada cambio. La capacidad está disponible a través de Python, SQL y Spark Connect (SPARK-58321, SPARK-58247, SPARK-58261) y permite a los usuarios aplicar cambios automáticamente en lugar de realizar operaciones de combinación manuales.

Ejemplo: Realiza un seguimiento del estado y la región del cliente a lo largo del tiempo. Dado un origen de canalización customer_changes con un ID de cliente, hora del evento, estado y región:

Alternativamente, esto se puede expresar en SQL:

En una tabla de salida SCD-2, cada clave única está representada por una secuencia de registros, cada uno de los cuales marca el inicio y el final de su validez utilizando dos columnas de metadatos adicionales, __START_AT y __END_AT. Por ejemplo, cuando un registro de cambio indica que el cliente con id 17 se mudó de la región West a East, la canalización conserva el registro con la región anterior West y marca su final como la hora del evento del registro de cambio; inserta un nuevo registro con la región East y la hora del evento del registro como su inicio. Un análisis puede entonces utilizar la región que se aplicaba en el punto relevante de la secuencia de tiempo de los eventos.

En segundo plano, Auto CDC conecta la definición de flujo declarativo con la derivación del esquema de destino y el procesamiento de flujos SCD Tipo 2 dedicado (SPARK-58319, SPARK-57395, SPARK-58321). Las claves identifican la entidad, la secuenciación determina el orden de los cambios y las columnas de seguimiento del historial especifican qué cambios de atributos crean versiones históricas. El motor también valida las columnas reservadas, los tipos de secuenciación y la configuración de seguimiento del historial (SPARK-57251, SPARK-58313, SPARK-58391), y concilia micro lotes más estrechos con un esquema de destino evolucionado (SPARK-58418).

Auto CDC aplica cada micro lote al destino usando MERGE, por lo que la tabla de destino debe estar en un catálogo cuyo conector implemente operaciones a nivel de fila de Data Source V2. Los formatos de archivo integrados de Spark, incluido Parquet, no califican. Por lo tanto, una opción común es Iceberg.

Spark Connect: más fácil de desarrollar, observar y operar

Las aplicaciones y los agentes de IA necesitan invocar Spark desde sus propios procesos, administrar sesiones y conectar una solicitud fallida al trabajo que se ejecuta en el servidor. Spark 4.3 fortalece Spark Connect para estos flujos de trabajo con visibilidad de sesiones y operaciones, atribución de solicitudes y compatibilidad más amplia con clientes.

Spark Connect envía las operaciones de DataFrame como planes lógicos a través de gRPC, las ejecuta mediante el motor Spark del servidor y transmite los resultados de vuelta en lotes de Arrow. Spark 4.3 añade puntos de conexión REST para listar e inspeccionar las sesiones y operaciones de Connect, además de identificadores de operación para realizar un seguimiento de la ejecución entre los clientes y el servidor (SPARK-57941, SPARK-58547). Para el desarrollo local de PySpark en sistemas POSIX, la opción experimental `SPARK_LOCAL_CONNECT_REUSE=1` reutiliza un servidor de Connect persistente entre ejecuciones, lo que evita la sobrecarga de inicios repetidos (SPARK-57787). El contenedor de compatibilidad SQLContext de PySpark también funciona con Spark Connect (SPARK-57021).

El controlador JDBC de Spark Connect añade compatibilidad con resultados ARRAY, MAP y STRUCT, así como un manejo más completo de conjuntos de resultados (SPARK-57380, SPARK-57274), metadatos de tipos SQL e inspección de esquemas más compatible (SPARK-57627), y un manejo mejorado de excepciones (SPARK-57933, SPARK-58619). La cobertura de la API se amplía con DataFrame.zip, SCD Type 2 Auto CDC, y tipos y literales de marca de tiempo en nanosegundos (SPARK-57247, SPARK-58247, SPARK-57160, SPARK-57161).

Por otra parte, el trabajo de la comunidad incluye Spark Connect Gateway (SPARK-58455) para el enrutamiento consciente de sesiones y operaciones multiinquilino, clientes de Python basados en Rust y Rust (SPARK-58960) para otra ruta de implementación de clientes, y una propuesta de descubrimiento de capacidades (SPARK-56398) para identificar las operaciones de servidor compatibles antes de enviar una consulta. Estos esfuerzos tienen cronogramas de desarrollo independientes y no son funciones integradas de la versión 4.3.

PySpark: API más completas y mejor ejecución en Python

Los flujos de trabajo en Python suelen crear columnas derivadas en pasos separados, combinar los resultados y mover datos entre pandas, Arrow y Spark. Spark 4.3 pone una mayor parte de este trabajo a disposición a través de las API de DataFrame mientras mejora las rutas de ejecución subyacentes. DataFrame.zip añade una forma de combinar proyecciones de columnas compatibles sin recurrir a las API de RDD, incluso al usar Spark Connect (SPARK-55886, SPARK-57247).

Ejemplo: Combina identificadores con un valor derivado. Crea dos proyecciones a partir de la misma entrada y une sus columnas:

El resultado empareja id=1 con doubled=20 y id=2 con doubled=40. `zip` combina transformaciones que solo implican proyecciones del mismo plano subyacente; el filtrado, la ordenación, las uniones o las agregaciones por debajo de esas proyecciones deben ser idénticos en ambos lados para preservar la alineación de las filas.

El trabajo de ejecución más amplio se centra en la transferencia de datos, el procesamiento por lotes y el cálculo nativo. Los lotes de entrada de UDF basados en Pickle se pueden delimitar por bytes en lugar de solo por el recuento de filas, lo que ofrece un mejor control cuando el tamaño de las filas varía (SPARK-57593). Las proyecciones por lotes de pandas en Spark y las implementaciones nativas adicionales de NumPy reducen el trabajo evitable (SPARK-58230, SPARK-58234), mientras que la conversión de pandas a Spark añade compatibilidad con columnas PyArrow ChunkedArray (SPARK-46776).

La versión también añade enlaces de Python y Scala para funciones vectoriales existentes (SPARK-57284), covarianza agrupada y estadística descriptiva (SPARK-40336, SPARK-40337), operaciones de mediana móvil y de expansión y de error estándar (SPARK-40341, SPARK-40340), y diagnósticos con show_versions (SPARK-40373). Las variables locales de seguimiento de errores (traceback locals) son opcionales porque pueden contener datos confidenciales (SPARK-57271).

Spark SQL: Prepara un mejor contexto para la analítica y la IA

Las cargas útiles de eventos llegan como documentos anidados y los eventos muy cercanos entre sí pueden requerir mayor precisión que los microsegundos. La preparación de estas entradas no debe requerir una capa de procesamiento personalizada independiente para cada transformación. Spark 4.3 amplía el SQL nativo para datos semiestructurados y análisis temporales, lo que incluye la manipulación de documentos, la coincidencia temporal y una vista previa opcional de marcas de tiempo en nanosegundos (SPARK-56822).

Ejemplo: Distingue dos eventos dentro del mismo microsegundo. Habilita la vista previa (la función aún no es estable, por lo que podrían producirse cambios incompatibles) y compara marcas de tiempo con un nanosegundo de diferencia (SPARK-56969, SPARK-56965, SPARK-57211):

La comparación se evalúa como true: el dígito final se conserva en lugar de descartarse. La vista previa es compatible con TIMESTAMP_NTZ(p) y TIMESTAMP_LTZ(p) en precisiones de 7 a 9.

Esto requirió algo más que aceptar un nuevo nombre de tipo SQL. La compatibilidad con nanosegundos se extiende a través del análisis sintáctico, las representaciones de valores internos, las conversiones (casts), las expresiones, Spark Connect, el almacenamiento en caché y la integración de fuentes de datos (SPARK-56822). Las nuevas funciones timestamp_nanos y unix_nanos gestionan las conversiones numéricas (SPARK-57526, SPARK-57527). La compatibilidad con el almacenamiento y los conectores sigue siendo específica para cada ruta de datos, por lo que la vista previa debe evaluarse de extremo a extremo.

Convierte cargas útiles JSON cambiantes en datos utilizables

Los eventos de aplicaciones, las exportaciones de documentos y las respuestas de modelos suelen llegar como JSON anidado. JSON_TABLE de Spark 4.3 (SPARK-58366) combina la expansión de matrices y la extracción de campos tipados, mientras que las transformaciones VARIANT permiten que las canalizaciones actualicen o eliminen campos sin UDF personalizadas de procesamiento de documentos (SPARK-57804, SPARK-57417). Juntas, simplifican la preparación de contenido para canalizaciones de análisis, enriquecimiento y recuperación.

Ejemplo: Extraer secciones relevantes de documentos.

Esto produce una fila tipada que una canalización puede enriquecer con metadatos de origen antes de otras operaciones posteriores. JSON_TABLE resuelve las rutas de sus columnas mediante un recorrido compartido de cada elemento de fila, evitando un análisis independiente para cada columna. Por otro lado, Spark amplía la inserción hacia abajo (pushdown) de extracción de campos VARIANT a consultas compatibles que agrupan, ordenan o unen. (SPARK-58089).

Trabajo posterior en SQL/JSON (SPARK-58686) amplía esta dirección con JSON_VALUE, JSON_EXISTS y JSON_QUERY para la extracción escalar, comprobaciones de presencia y recuperación de fragmentos (SPARK-58685, SPARK-58687, SPARK-58723). El manejo explícito de errores y valores faltantes ayudará a las aplicaciones a procesar esquemas cambiantes de manera más predecible. Estos son desarrollos posteriores, no funciones adicionales de la versión 4.3.

Relacionar eventos con la información disponible en ese momento

El análisis histórico y la evaluación de modelos necesitan el contexto disponible cuando ocurrió un evento, no la información agregada más tarde. SQL ASOF JOIN de Spark 4.3 expresa (función en vista previa, pueden ocurrir cambios incompatibles) esta coincidencia temporal directamente, reemplazando las uniones de rangos escritas a mano seguidas de clasificación (SPARK-58092).

Ejemplo: Reconstruir las características disponibles para una predicción. Aquí, available_at registra cuándo estuvo disponible una versión de característica (SPARK-58122, SPARK-58123):

Una predicción a las 10:03 coincide con la versión de característica disponible a las 10:00, no a las 10:05. Una ruta de ejecución de combinación por ordenamiento dedicada hace que esta coincidencia sea una operación del motor en lugar de una lógica de consulta específica de la aplicación. El resultado es un enriquecimiento puntual más sencillo para datos de entrenamiento, evaluaciones y análisis de incidentes, siempre que las marcas de tiempo de disponibilidad representen con precisión lo que se sabía en ese momento.

Mejoras notables y agradecimientos

Spark 4.3 también introduce nuevas API y mejoras de rendimiento para el procesamiento de datos diario. SQL incorpora ANSI UNNEST para expandir matrices en filas, junto con la función de ventana `counter_diff` para convertir contadores acumulativos en deltas y `jaro_winkler_similarity` para comparaciones de similitud de cadenas. Las fuentes de archivos compatibles agregan soporte opcional para leer archivos tar, zip y 7z, eliminando un paso de extracción independiente. El motor mejora la ejecución de agregaciones y habilita la generación de código de etapa completa para operaciones UNION elegibles de forma predeterminada. Una optimización opcional distribuye las claves de unión nulas entre particiones de mezcla (shuffle) en uniones externas y anti uniones elegibles. El procesamiento de datos anidados obtiene una mayor poda de campos no utilizados dentro de matrices de estructuras y generación de código para más funciones de orden superior de matrices. Las lecturas vectorizadas de Parquet mejoran la conversión de tipos y la decodificación de diccionarios.

Spark 4.3 refleja la solidez de la comunidad de Apache Spark, con más de 200 colaboradores que ayudaron a resolver más de 1.100 tickets de Jira. Agradecemos a todos los que contribuyeron con código, revisiones, pruebas, documentación y comentarios para hacer posible este lanzamiento.

Comience a usar Spark 4.3

Visite la página de descargas de Apache Spark para obtener las descargas de la versión y la documentación, y revise las guías de migración antes de actualizar. Para conocer la disponibilidad en Databricks, consulte las notas de la versión de Databricks Runtime.

(Esta entrada del blog ha sido traducida utilizando herramientas basadas en inteligencia artificial) Publicación original