Presentamos el reparticionamiento de estado bajo demanda para Apache Spark™ Structured Streaming en Databricks

Cualquiera que ejecute consultas con estado de Apache Spark™ Structured Streaming en producción tarde o temprano se topa con el mismo muro incómodo.

Iniciaste la consulta hace meses. En aquel entonces, el volumen de datos era modesto, así que aceptaste el valor predeterminado de 200 particiones de shuffle y continuaste. El pipeline funcionó sin problemas. Luego, el negocio creció, el tráfico se triplicó y el almacenamiento de estado se disparó. De repente, esas 200 particiones ya no tienen el tamaño adecuado. Algunas particiones están sesgadas y se sobrecargan, el clúster está al límite y cada microbatch tarda más de lo debido.

Así que haces lo lógico: aumentas spark.sql.shuffle.partitions y reinicias la consulta. No cambia nada.

La consulta ignora silenciosamente tu nuevo valor porque el recuento de particiones se integró en el checkpoint cuando iniciaste el stream por primera vez. Históricamente, la única forma de aplicar un nuevo número ha sido abandonar el checkpoint existente y empezar de nuevo, lo que, para una consulta con estado, significa perder todo el estado acumulado que has estado manteniendo con tanto cuidado. Para un modelo de fraude que rastrea millones de cuentas, o un trabajo de sesionización que retiene días de ventanas, "empezar de nuevo" no es una frase que nadie quiera pronunciar en una revisión de incidentes de producción.

La repartición de estado bajo demanda (Vista previa pública), disponible en Databricks Runtime 18 y versiones superiores, elimina ese muro. Ahora puedes cambiar el tamaño del número de particiones para una consulta de streaming con estado y mantener intacto el estado de tu checkpoint.  

Esto se aplica a cualquier consulta de streaming con estado, ya sea que ejecutes agregaciones, uniones stream-stream, deduplicación, sesionización o transformWithState, y a cualquier carga de trabajo, desde la detección de fraudes hasta el monitoreo en tiempo real.

Para los primeros usuarios como Coveo, la capacidad de adaptar correctamente el tamaño de su infraestructura de streaming bajo demanda se tradujo de inmediato en importantes ahorros operativos. 

En Coveo, ejecutamos pipelines de streaming con estado a gran escala donde los volúmenes de datos fluctúan significativamente con el tiempo. Con Databricks y la capacidad de repartición de estado, hemos reducido nuestros costos relacionados con la API de Amazon S3 en un 40 %.  Antes, cada decisión de escalado obligaba a un compromiso: sobredimensionar o reconstruir a partir de nuevos checkpoints, lo que elevaba los costos de la API de almacenamiento a casi el mismo nivel que los costos de cómputo. Ahora escalamos libremente a medida que cambia la demanda, sin alterar el estado existente ni provocar costosas migraciones de checkpoints”. —Alexis Chicoine, desarrollador sénior de software, Coveo

Bajo el capó: ¿Por qué estaban bloqueadas las particiones de estado?

Para entender por qué los resultados de Coveo representan un avance tan significativo para Structured Streaming, debemos analizar por qué se congelaba el recuento de particiones en primer lugar.

Una consulta de streaming con estado guarda su estado en un almacenamiento de estado, y ese estado se particiona físicamente. Cada clave en tu stream (un ID de usuario, un número de cuenta, una ventana) se asocia mediante hash a una partición específica, y los datos de cada partición se almacenan en su propia instancia independiente de RocksDB dentro del checkpoint. El número de particiones define la distribución de todo el almacenamiento de estado en el disco.

Si simplemente cambiaras el recuento de particiones entre reinicios, el hash ya no coincidiría. Una clave que antes residía en una partición (por ejemplo, la partición 47) ahora podría asociarse a una diferente (la partición 12), pero su estado acumulado seguiría estando en los archivos de la partición original. En la práctica, la consulta perdería el rastro de su propia memoria. Para evitar exactamente este tipo de corrupción silenciosa, Structured Streaming bloqueaba el recuento de particiones al crear el checkpoint e ignoraba cualquier cambio posterior en spark.sql.shuffle.partitions.

Seguro, pero poco flexible. Los dos costos que pagabas eran:

1. No podías realizar ajustes. Si resultaba que 200 particiones no eran la opción correcta, te quedabas con esa configuración durante toda la vida útil del checkpoint.
2. No podías escalar con la carga de trabajo. A medida que el volumen de datos crecía o disminuía, el recuento de particiones no podía seguir el ritmo.

La repartición de estado bajo demanda soluciona ambos problemas haciendo lo único que el diseño anterior se negaba a hacer (pero haciéndolo de forma segura): redistribuir físicamente el estado para que coincida con el nuevo recuento de particiones.

Qué necesitas para empezar

Los requisitos son breves:

• Databricks Runtime 18 o superior.
• El proveedor de almacenamiento de estado RocksDB. En DBR 17.3 y versiones superiores, RocksDB es el valor predeterminado, y las nuevas consultas creadas en esas versiones lo usarán a menos que se cambie explícitamente.  Si deseas confirmarlo o configurarlo explícitamente, consulta Configurar el almacenamiento de estado RocksDB en Databricks.

Esa es toda la lista de requisitos previos. Si estás en DBR 18 con el almacenamiento de estado predeterminado, ya tienes todo lo que necesitas.

Cambiar el número de particiones

El mecanismo es sencillo y reutiliza un patrón que todo desarrollador de streaming ya conoce: detener, reconfigurar y reiniciar.

En lugar de spark.sql.shuffle.partitions, estableces una configuración dedicada,  spark.sql.streaming.stateStore.partitions, y reinicias la consulta:

El detalle clave es la propia configuración nueva. Para consultas con estado, spark.sql.streaming.stateStore.partitions tiene prioridad sobre spark.sql.shuffle.partitions. Esto es lo que hace que el cambio "se mantenga" a diferencia del enfoque anterior.

Cuando la consulta se reinicia, no reanuda el procesamiento normal de inmediato. Primero, finaliza el último microbatch planificado, si hay alguno pendiente. Luego, realiza una operación de repartición única: redistribuye físicamente los datos de estado entre el nuevo número de particiones, volviendo a asociar las claves mediante hash a sus nuevas ubicaciones correctas para que nada se pierda ni se traspapele. Una vez completada esa redistribución, la consulta reanuda el procesamiento habitual, ahora utilizando el recuento de particiones que solicitaste.

Ese paso de repartición es el núcleo de la función. Es la diferencia entre "cambiamos un número" y "trasladamos de forma segura tu estado a una nueva distribución".

Monitorear la operación de repartición

Dado que la repartición es una operación real cuyo tiempo de ejecución es proporcional a la cantidad de estado, querrás tener visibilidad sobre ella. Structured Streaming muestra esto a través de sus informes de progreso estándar.

Una vez completado el siguiente microbatch, los eventos StreamingQueryProgress incluyen la duración de la operación de repartición. Busca en las métricas durationMs del evento el campo controlBatch.REPARTITION, que informa la duración de la repartición en milisegundos.

Un mayor volumen de estado significa una repartición más larga, pero esperamos que tarde solo unos segundos para la mayoría de las cargas de trabajo. Por lo tanto, en trabajos grandes, vale la pena capturar esta métrica para comprender la duración. Para obtener más información sobre cómo leer estos eventos, consulta Monitorear consultas de Structured Streaming en Databricks.

Ejemplo: reducir la escala de una consulta

Hagamos esto concreto con una agregación simple: un recuento de eventos por id en una ventana de saltos (tumbling window). Comenzaremos con el valor predeterminado de 200 particiones, decidiremos que es más de lo que necesita esta carga de trabajo y reduciremos la escala a 100.

Primero, la consulta tal como se ejecuta hoy, con el recuento de particiones predeterminado:

Ahora, hemos observado este stream durante un tiempo y hemos llegado a la conclusión de que 200 particiones es excesivo. Estamos pagando una sobrecarga de coordinación por un paralelismo que no necesitamos. Detenemos la consulta, configuramos el nuevo recuento de particiones y la reiniciamos con las mismas opciones y el mismo checkpoint:

Cuando se inicia la consulta reiniciada, finaliza el último microbatch planificado, si hay alguno pendiente, ejecuta la repartición para redistribuir el estado de 200 particiones a 100 y luego continúa contando con cada ventana y cada total acumulado completamente preservados. El mismo procedimiento funciona a la inversa: para escalar verticalmente bajo una carga más pesada, simplemente establecerías un número mayor.

El mismo enfoque se aplica a Spark Declarative Pipelines (SDP). Consulta el ejemplo de SDP en la documentación para ver un tutorial completo.

Cuándo utilizar el reparticionamiento de estado

El reparticionamiento de estado bajo demanda es una herramienta de ajuste y escalado más que una operación rutinaria. Resulta muy útil en algunas situaciones clave: 

• Dimensionamiento correcto después del lanzamiento. Iniciaste el pipeline con las 200 particiones predeterminadas el primer día porque el flujo era pequeño y no valía la pena realizar un ajuste fino. Seis meses después, ese número está consolidado en un checkpoint que no puedes permitirte perder, y ya no es suficiente. Por ejemplo: un flujo de puntuación de fraude que se lanzó en una sola región piloto ahora cubre todos los mercados, y 200 particiones hacen que cada una almacene demasiado estado. Con el reparticionamiento bajo demanda, puedes aumentar el número de particiones para adaptarlo a tu carga actual sin perder el checkpoint existente.
• Cambios en las cargas de trabajo. Dimensionaste el flujo para el tráfico pico. Por ejemplo: un pipeline de ofertas de anuncios tiene una gran actividad durante el día y se calma por la noche, por lo que un valor ajustado para el pico diurno deja a la mayoría de las particiones inactivas a las 3 a. m. Con el reparticionamiento bajo demanda, escalas hacia arriba al entrar en el periodo de mayor actividad y vuelves a bajar una vez que pasa, de modo que el particionamiento se adapta a la carga real en lugar de al peor de los casos.
• Backfill de datos históricos:  El backfill y el procesamiento en estado estable necesitan diferentes números de particiones y, anteriormente, tenías que elegir uno para toda la vida útil del checkpoint. Por ejemplo: volver a procesar dos años de historial requiere un número elevado de particiones para distribuir el trabajo y terminar rápido, pero ese mismo número es un desperdicio una vez que vuelves al tráfico de estado estable. El reparticionamiento bajo demanda te permite escalar hacia arriba para el backfill y hacia abajo al tamaño de estado estable después de ponerte al día, todo sin perder el checkpoint ni el estado.
• Ajuste de rendimiento. El número de particiones afecta al paralelismo, al tamaño del estado y a la sobrecarga de shuffle, y el valor óptimo es difícil de predecir. Por ejemplo: podrías pensar que 200 es muy poco y que 400 reduciría la latencia de los microlotes, pero antes las pruebas requerían reconstruir el estado y volver a procesar los datos, lo que desperdiciaba recursos. El reparticionamiento bajo demanda te permite ajustar el número de particiones con tu checkpoint activo, monitorear controlBatch.REPARTITION y las duraciones de los microlotes, y tomar decisiones basadas en mediciones en lugar de suposiciones.

Dado que cada cambio requiere una parada y un reinicio con una pausa de reparticionamiento única, considéralo como una acción de mantenimiento deliberada. Planifica el cambio de tamaño para una ventana de tiempo en la que sea aceptable una breve pausa en el procesamiento, observa controlBatch.REPARTITION para confirmar cuánto tiempo tomó y deja que la consulta vuelva a su ritmo normal.

Conclusión

Durante años, el número de particiones de una consulta de streaming con estado era una decisión que se tomaba una sola vez, al principio, y que luego nunca se volvía a revisar, o se pagaba un alto precio para reconstruir el estado desde cero. El reparticionamiento de estado bajo demanda elimina estas limitaciones. Redistribuir el estado de forma segura en un nuevo número de particiones convierte una decisión que antes solo se tomaba al inicio en una que puedes volver a evaluar siempre que tu carga de trabajo lo requiera.

El resultado es exactamente lo que los operadores de flujos de larga duración siempre han querido: la libertad de dimensionar correctamente una consulta en función de sus necesidades de escalado, con tan solo una parada, un cambio de configuración y un reinicio, sin perder su estado.

El reparticionamiento de estado bajo demanda está disponible en Databricks Runtime 18 y versiones superiores, utilizando el proveedor de almacenamiento de estado RocksDB. Para obtener la referencia completa, consulta Reparticionamiento de estado bajo demanda para consultas de streaming con estado.

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