Fundamentos de Arquitectura
La abstracción principal de Atlas Stream Processing es el procesador de flujo. Un procesador de flujo es un pipeline de agregación de MongoDB que opera continuamente sobre datos de transmisión de una fuente específica y guarda la salida en un receptor. Para aprender más, consultar Estructura de un procesador de flujo.
Stream Processing takes place on Stream Processing workspaces. Cada espacio de trabajo de Stream Processing es un namespace de Atlas que asocia lo siguiente:
Uno o más procesadores de transmitir , cada uno de los cuales se ejecuta con su propia asignación de RAM y CPU.
Un nivel por defecto, que determina la cantidad de memoria y cómputo disponible para cada procesador de flujo cuando no se especifica un nivel.
Un nivel máximo, que determina la mayor cantidad de memoria y recursos de cómputo que se pueden asignar a un pod dentro de ese espacio de trabajo de Stream Processing.
Un proveedor de nube y una región de nube.
Un registro de conexiones, que almacena la lista de orígenes y destinos disponibles de datos en transmisión.
Un contexto de seguridad en el que definir las autorizaciones de usuario.
Una cadena de conexión al mismo espacio de trabajo de Stream Processing.
Niveles
Cuando se define un Stream Processor, este solo está disponible para el espacio de trabajo de Stream Processing en el que se define. Cada procesador de streaming se ejecuta en recursos asignados según su nivel. Atlas Stream Processing factura a los usuarios por un procesador de streaming solo mientras se está ejecutando.
Si inicia un procesador de flujo sin declarar un tamaño de nivel, se ejecutará el nivel del espacio de trabajo de Stream Processing. Puedes iniciar un procesador de Stream Processing de cualquier nivel hasta el nivel máximo del espacio de trabajo de Stream Processing
Ejemplo
El usuario define un procesador de flujo en un espacio de trabajo de Stream Processing denominado myWorkspace con un nivel por defecto de SP10 y un nivel máximo de SP50. Si se inicia el procesador sin especificar un nivel, Atlas Stream Processing lo asigna a un pod SP10. Sin embargo, el usuario puede declarar cualquier nivel desde SP2 hasta SP50 y Atlas Stream Processing asignará el procesador a un pod del tamaño adecuado.
Registro de conexões
Los registros de conexión almacenan una o más conexiones. Cada conexión asigna un nombre a la combinación de detalles de red y seguridad que permiten que un procesador de flujos interactúe con servicios externos. Las conexiones presentan el siguiente comportamiento:
Solamente una conexión definida en el registro de conexiones de un espacio de trabajo de Stream Processing específico puede dar servicio a procesadores de flujo alojados en ese espacio de trabajo de Stream Processing.
Cada conexión puede dar servicio a un número arbitrario de procesadores de flujo
Solo una única conexión puede servir como fuente de un determinado procesador de flujos.
Solo una única conexión puede servir como receptor de un procesador de flujo determinado.
Una conexión no se define intrínsecamente como fuente o como sumidero. Cualquier conexión dada puede servir para cualquiera de las dos funciones según cómo un procesador de flujos invoque esa conexión.
Atlas Stream Processing ejecuta cada procesador de streams en un contenedor dedicado para el cliente con aislamiento a nivel de VM en una infraestructura multiusuario. Para obtener más información sobre la seguridad y el cumplimiento de MongoDB, consulte el Centro de confianza de MongoDB.
Puntos de control
Atlas Stream Processing captura el estado de un procesador de flujo utilizando puntos de control. Cada punto de control tiene una ID única y está sujeto al flujo de la lógica de tu procesador de flujo. Una vez que todos los operadores de un procesador de flujo han añadido su estado a un punto de control, Atlas Stream Processing confirma el punto de control, generando dos tipos de registros:
Un registro de commit único que valida el ID del punto de control y el procesador de flujo al que pertenece.
Un conjunto de registros que describe el estado de cada operación con estado en el respectivo procesador de flujo justo en el momento en que Atlas Stream Processing realiza el compromiso del punto de control.
Cuando vuelves a iniciar un procesador de flujo tras una interrupción, Atlas Stream Processing query el último punto de control confirmado y reanuda la operación desde el estado descrito.
Los puntos de control registran los offsets de origen en lugar de copias de los documentos de origen. Por ejemplo, Atlas Stream Processing almacena los tokens de reanudación oplog para fuentes de flujo de cambios y los desplazamientos de partición para fuentes Kafka. La eliminación o el vencimiento de los documentos fuente después de que Atlas Stream Processing confirme un punto de control no impide que el procesador se reanude.
Si el token de reanudación del oplog que Atlas Stream Processing almacenó en el último punto de control ya no se encuentra en el oplog del clúster de origen cuando reinicias el procesador, Atlas Stream Processing no puede reanudar desde ese punto de control. Para recuperar, reinicia el procesador con resumeFromCheckpoint=false. Para reducir el riesgo de que esto ocurra, configure su clúster de origen con una oplog window adecuada. Para aprender cómo configurar una oplog window, consulta Establecer ventana mínima de oplog window.
Para las configuraciones de flujo de cambios $source con initialSync activado, Atlas Stream Processing verifica tanto la fase de sincronización inicial, que copia las colecciones fuente, como la fase normal de procesamiento de eventos de cambios. Durante la sincronización inicial, Atlas Stream Processing registra la posición de inicio del oplog y divide las colecciones fuente en particiones. El procesador copia los documentos existentes a través de las particiones y luego reproduce los eventos de cambios desde la posición registrada inicial una vez que se completa la copia. Atlas Stream Processing crea puntos de control durante la sincronización inicial, y el procesador puede reanudarse en caso de que ocurra una interrupción.
Fuentes de Kafka y desplazamientos del grupo de consumidores
Después de que Atlas Stream Processing realice un commit de un punto de control, actualiza los desplazamientos del grupo de consumidores en Kafka de manera asíncrona. Cuando utilizas Apache Kafka como $source, los grupos de consumidores rastrean estos puntos de control como desplazamientos comprometidos en el clúster de Kafka para cada partición.
Debido a que estas actualizaciones se producen periódicamente y de forma asincrónica, los offsets confirmados del grupo de consumidores pueden retrasarse temporalmente con respecto al punto de control más reciente. Esto puede causar retrasos a corto plazo en la supervisión y las métricas de retraso de las herramientas que leen los offsets comprometidos de Kafka, como la herramienta CLI kafka-consumer-group. Estas herramientas pueden mostrar el grupo de consumidores "detrás" de la verdadera posición interna del procesador de flujos.
La cantidad de retraso entre los puntos de control internos y los desplazamientos comprometidos de Kafka no es fija y puede variar según la carga de trabajo y la configuración. En cargas de trabajo típicas, está en el rango de segundos, pero no hay un límite superior estricto.
fila de letra muerta
Atlas Stream Processing admite el uso de una colección de bases de datos Atlas como fila de letra muerta (DLQ). Cuando Atlas Stream Processing no puede procesar un documento de tu flujo de datos, graba el contenido del documento en la DLQ junto con detalles del fallo de procesamiento. Puedes asignar una colección como DLQ en tus definiciones de procesador de flujo.
Para aprender más sobre cómo habilitar DLQ, consulta Crear un procesador de flujo. Para aprender sobre el esquema DLQ de Atlas Stream Processing, consulta la guía de supervisión.
Estados del procesador de flujos
Los procesadores de streaming cambian entre diferentes estados a lo largo de su ciclo de vida. Comprender estos estados te ayuda a supervisar y solucionar problemas en tus operaciones de Stream Processing.
Los estados posibles incluyen:
Estado | Descripción |
|---|---|
| El procesador de flujos está procesando activamente los datos y operando normalmente. Este es el estado operativo deseado donde tu procesador consume datos de fuentes, aplica transformaciones y escribe en destinos. |
| El procesador de flujos se ha detenido manualmente y no está procesando datos. Para aprender a iniciar un procesador de transmisión, consulta Iniciar un Procesador de Transmisión." |
| El procesador de flujos está en proceso de asignar recursos o escalar la infraestructura. Las causas comunes incluyen:
Los posibles impactos incluyen:
Una vez finalizado el provisionamineto, los procesos se reanudan automáticamente. No se produce pérdida de datos, ya que Atlas Stream Processing conserva el estado a través de puntos de control. |
| El procesador de transmisión ha encontrado un error que impide su ejecución. Esto ocurre debido a errores del usuario o problemas de configuración. Para obtener información detallada sobre las causas de los errores y la recuperación, consulte Manejo de errores y políticas de reintento. |
Nota
Es posible que los procesadores de flujo aparezcan intermitentemente en estado provisionamiento en los tableros de supervisión a medida que el sistema asigna recursos. Este es un comportamiento normal durante las operaciones de escalado. Si un procesador permanece en estado de PROVISIONING por períodos prolongados (> 10 minutos), verifica los límites de recursos y la configuración del nivel en tu espacio de trabajo de Stream Processing.
Procesadores de conmutación por error
Atlas Stream Processing ofrece alta disponibilidad mediante procesadores de conmutación por error. Un procesador de conmutación por error replica el procesador de flujo activo mediante la replicación de puntos de control, lo que permite reanudar el procesamiento de flujo sin interrupciones desde el estado más reciente en caso de interrupción del servicio en la región principal del espacio de trabajo de procesamiento de flujo. Atlas Stream Processing sincroniza automáticamente cualquier cambio que realice en la canalización de agregación del procesador activo con sus procesadores de conmutación por error.
Para utilizar procesadores de conmutación por error, debe:
Configure su procesador con un nivel de
SP10o superior.Configure una o más regiones de conmutación por error en el espacio de trabajo de procesamiento de flujo de destino.
Configure un clúster multirregión con un nodo en la región de conmutación por error dentro del mismo proyecto que su espacio de trabajo de procesamiento de flujos.
Configure las conexiones en su espacio de trabajo de procesamiento de flujos para que admitan regiones de conmutación por error. Puede hacerlo al crear la conexión o modificar una conexión existente para que admita regiones de conmutación por error.
Configure uno o más procesadores de conmutación por error para su procesador de flujo de destino. Puede hacerlo al crear el procesador o modificar uno existente para que utilice procesadores de conmutación por error.
Puede iniciar uno de tres tipos de eventos de conmutación por error:
Tipo | Descripción |
|---|---|
Agraciado | Apaga un procesador activo, crea y copia un punto de control y reanuda el procesamiento desde el punto de control más reciente. |
Forzado | Activa el procesador de conmutación por error sin reanudar desde un punto de control. Esto se recomienda únicamente cuando no sea posible realizar una conmutación por error controlada. |
Prueba en seco | Prueba el proceso de conmutación por error sin consumir datos, lo que permite validar la configuración de conmutación por error antes de ponerla en producción. |
Consideraciones sobre la recuperación ante desastres
Puede crear regiones de conmutación por error para un espacio de trabajo de procesamiento de flujos al crear inicialmente dicho espacio de trabajo, o bien puede modificar un espacio de trabajo de procesamiento de flujos existente para que utilice regiones de conmutación por error.
Puede activar la conmutación por error para procesadores individuales, para todos los procesadores dentro de un espacio de trabajo determinado o para todos los procesadores con conmutación por error configurada en una región específica.
Los procesadores de conmutación por error garantizan una semántica de procesamiento de al menos una vez.
Para obtener más información sobre los requisitos y las limitaciones de los procesadores de conmutación por error, consulte la sección Limitaciones.
Para obtener información sobre los precios de los procesadores de conmutación por error, consulte la documentación de facturación.
Manejo de errores y políticas de reintentos
Atlas Stream Processing implementa un manejo integral de errores y políticas de reintentos para garantizar un procesamiento fiable de flujos. El sistema distingue entre diferentes tipos de errores y aplica estrategias de reintento apropiadas en función de la clasificación de errores.
Clasificación de errores
Atlas Stream Processing clasifica los errores en dos categorías principales:
Errores del usuario
Errores causados por la configuración del usuario, problemas de datos o problemas de servicios externos que están fuera del control de Atlas Stream Processing. Se incluyen ejemplos:
Credenciales de conexión inválidas
Problemas de conectividad de red con servicios externos
Datos con formato incorrecto que no se pueden procesar
Problemas de permisos al acceder a recursos externos
Errores internos
Errores que se producen dentro del propio sistema Atlas Stream Processing, generalmente debido a problemas temporales de infraestructura o interrupciones del servicio. Se considera que es responsabilidad del servicio Atlas Stream Processing resolverlos.
Comportamiento de reintentos
El comportamiento de reintento varía en función de la clasificación del error:
Política de reintentos ante errores de usuario
Atlas Stream Processing attempts to restart the stream processor a limited number of times over a 5-minute period
Si todos los intentos de reintento fallan dentro de este plazo, el procesador de flujo pasa a un estado
FAILEDAlgunos errores de usuario se clasifican como no reintentables y provocan un fallo inmediato del procesador. Los ejemplos incluyen:
StreamProcessorWorkerOutOfMemory (418): Pipeline supera los límites de memoria del nivelStreamProcessorInvalidOptions (420): Sintaxis o configuración de la pipeline no válida
Puedes reiniciar manualmente un procesador de flujos fallido llamando al método
start()
Política de reintento por error interno
Atlas Stream Processing reintenta continuamente errores internos sin límite de tiempo
Los errores internos activan alertas al equipo de ingeniería de Atlas Stream Processing
El sistema depende de mecanismos automáticos de reintento para recuperar los procesadores una vez resueltos los problemas internos
Los procesadores de flujo se reanudan automáticamente desde su último punto de control cuando se reinician
Proceso de recuperación
Cuando un procesador de flujos encuentra un error y requiere un reinicio, el proceso de recuperación sigue estos pasos:
Este proceso de recuperación garantiza que los fallos temporales no resultan en la pérdida de datos ni requieren intervención manual en la mayoría de los casos.
Tiempo de Stream Processing
En el procesamiento de datos en transmisión, los documentos están sujetos a dos sistemas de temporización:
hora del evento
tiempo de procesamiento
Atlas Stream Processing ofrece varios parámetros para controlar cómo los procesadores de flujos interactúan con estos sistemas temporales.
Hora del evento
El tiempo de evento es el momento en el que el flujo fuente genera un documento, o el sistema de mensajería (por ejemplo, Apache Kafka) recibe el documento. Esto se comprueba mediante la marca de tiempo del documento.
La latencia de red, el procesamiento ascendente y otros factores no sólo pueden causar discrepancias entre estos tiempos para un documento determinado, sino que también pueden provocar que los documentos lleguen a un procesador de flujo fuera del orden del tiempo de evento. En cualquier caso, las ventanas pueden dejar pasar documentos que usted desea que capturen. Atlas Stream Processing considera tales documentos de llegada tardía y los envía a tu fila de letra muerta si configuras una.
Event Time es una opción configurable para el campo boundary habilitada en Ventanas Tumbling y Ventanas Hopping.
Tiempo de procesamiento
El tiempo de procesamiento es el momento en el que el procesador de flujo consume un documento. Esto se determina por el reloj del sistema que aloja el procesador de flujo.
El tiempo de procesamiento es una opción configurable para el campo boundary admitido en Tumbling Windows y Hopping Windows. Permite crear una pipeline con una especie de ventana que acumula datos según la hora del reloj del servidor. A diferencia de tiempo de evento ventanas, las ventanas de procesamiento asignan a cada evento una marca de tiempo basada en la hora del reloj del servidor cuando llega al procesador del flujo.
Las marcas de tiempo de los documentos y las marcas de tiempo de los límites de las ventanas están en UTC. No puedes especificar las opciones idleTimeout o allowedLateness al configurar una ventana processingTime.
Ejemplo
Se crea un pipeline con una ventana de tiempo de evento de 5 minutos. Se añade un evento a un clúster origen de Kafka en 09:33. Debido a cierto retraso en el clúster de Kafka, llega al procesador de flujo en 09:37.
Si el pipeline tiene una ventana de tiempo de eventos de 5 minutos, este evento se asignará a la ventana 09:30-09:35. Si el pipeline tiene una ventana de tiempo de procesamiento de 5 minutos, el evento se asignará a la ventana 09:35-09:40.
Marcas de agua
Una marca de agua reemplaza el tiempo de procesamiento y solo se actualiza cuando el procesador consume un documento con un tiempo de evento posterior al de cualquier documento consumido previamente. Atlas Stream Processing aplica marcas de agua solo cuando un pipeline incluye una etapa de ventana ($tumblingWindow, $hoppingWindow o $sessionWindow).
Ejemplo
Usted configura un procesador de flujo con ventanas de 5minutos. Inicia el procesador en 12:00, para que las dos primeras ventanas tengan duraciones de 12:00-12:05 y 12:05-12:10. La siguiente tabla ilustra qué ventanas capturarán qué eventos dada las variaciones de los retardos, con y sin marcas de agua.
Hora del evento | Tiempo de procesamiento | Tiempo de ventana (sin marcas de agua) | Tiempo de ventana (marcas de agua) |
|---|---|---|---|
12:00 | 12:00 | 12:00-12:05 | 12:00-12:05 |
12:01 | 12:03 | 12:00-12:05 | 12:00-12:05 |
12:02 | 12:05 | 12:05-12:10 | 12:00-12:05 |
12:01 | 12:06 | 12:05-12:10 | 12:00-12:05 |
12:06 | 12:07 | 12:05-12:10 | 12:05-12:10 |
Sin marcas de agua, la ventana 12:00-12:05 se cierra a las 12:05 según el reloj del sistema del espacio de trabajo de Stream Processing, y la ventana 12:05-12:10 se abre de inmediato. Así, si bien la fuente generó cuatro de los documentos durante el intervalo 12:00-12:05, la ventana relevante solo captura dos documentos.
Con las marcas de agua, la ventana 12:00-12:05 no se cierra en 12:05 porque entre los documentos que ingiere hasta ese momento, el tiempo de evento más reciente —y, por lo tanto, el valor de la marca de agua— es 12:03. La ventana 12:00-12:05 no se cierra hasta 12:07 en el reloj del sistema, cuando el procesador de flujo ingiere un documento con una hora de evento de 12:05, avanza la marca de agua a esa hora y abre la ventana 12:05-12:10. Cada ventana captura todos los documentos adecuados.
Al leer de Apache Kafka, Atlas Stream Processing espera que todas las particiones pasen la marca de agua. Si una partición queda inactiva y no produce eventos con marcas de tiempo posteriores a la watermark, la ventana no se cierra ni produce resultados. Para abordar esto, establece partitionIdleTimeout para asegurar que las particiones inactivas no detengan el avance de las marcas de agua. Para obtener más información, consulta $source Etapa (Procesamiento de flujo).
Atraso permitido
Si las diferencias entre el tiempo del evento y el tiempo de procesamiento varían lo suficiente, los documentos pueden llegar a un procesador de flujo después de que la marca de agua haya avanzado lo suficiente para cerrar la ventana esperada. Para mitigar esto, Atlas Stream Processing admite el Retardo Permitido, una configuración que retrasa el cierre de una ventana por un intervalo determinado en relación con el watermark.
Mientras que las marcas de agua son propiedades de los procesadores de flujo, la Tolerancia permitida es una propiedad de la ventana y solo afecta cuándo se cierra esa ventana. Si la marca de agua del procesador de flujo avanza hasta un punto que desencadenaría la apertura de una nueva ventana, la Tolerancia permitida mantiene abiertas las ventanas anteriores sin impedir este hecho.
Ejemplo
Se configura un procesador de flujo con ventanas de tumbling de 5minutos. Inicias el procesador en 12:00, de modo que las dos primeras ventanas tendrán duraciones de 12:00-12:05 y 12:05-12:10. Configuraste una Tolerancia permitida de 2 minutos.
La tabla siguiente refleja el orden en que el procesador de flujo ingiere los documentos descritos.
Hora del evento | Marca de agua | Tiempo de retraso permitido | Ventana de tiempo |
|---|---|---|---|
12:00 | 12:00 | 11:58 | 12:00-12:05 |
12:02 | 12:03 | 12:01 | 12:00-12:05 |
12:01 | 12:04 | 12:02 | 12:00-12:05 |
12:05 | 12:05 | 12:03 | 12:00-12:05, 12:05-12:10 |
12:04 | 12:06 | 12:04 | 12:00-12:05, 12:05-12:10 |
12:07 | 12:07 | 12:05 | 12:05-12:10 |
Cuando la marca de agua avance a 12:05, se abre la ventana 12:05-12:10. Sin embargo, debido a que el intervalo de retraso permitido es de 2 minutos, desde dentro de la ventana 12:00-12:05, es efectivamente solo 12:03, por lo que permanece abierto. Solo cuando la marca de agua avanza a 12:07 la hora ajustada alcanza 12:05. En este punto, la ventana 12:00-12:05 se cerrará.
Tiempo de espera por inactividad
Desacoplar por defecto el comportamiento de las ventanas del tiempo de procesamiento mejora la exactitud del Stream Processing en la mayoría de los casos. Sin embargo, una fuente de datos en transmisión puede tener períodos prolongados de inactividad. En este escenario, una ventana puede capturar eventos anteriores al período de inactividad y no poder devolver resultados procesados mientras se espera a que la marca de agua avance lo suficiente como para cerrar.
Atlas Stream Processing permite a los usuarios configurar un tiempo de espera por inactividad para las ventanas a fin de mitigar estos escenarios utilizando el tiempo de procesamiento. Un tiempo de inactividad es un período que comienza cuando el tiempo de procesamiento supera el final del intervalo de una ventana abierta y la fuente del procesador de flujo está inactiva. Si la fuente permanece inactiva durante un intervalo igual al tiempo de espera de inactividad, la ventana se cierra y la marca de agua avanza independientemente de cualquier ingestión de documentos.
Ejemplo
Configura una ventana por lotes con un intervalo de 3minutos y un tiempo de espera por inactividad de 1minutos. La siguiente tabla ilustra los efectos del tiempo de espera de inactividad durante y después del intervalo de una ventana.
Tiempo de procesamiento | Hora o estado del evento | Marca de agua | Ventana de tiempo |
|---|---|---|---|
12:00 | 12:00 | 12:00 | 12:00-12:03 |
12:01 | Fuente inactiva | 12:00 | 12:00-12:03 |
12:02 | Fuente inactiva | 12:00 | 12:00-12:03 |
12:03 | Fuente inactiva | 12:00 | 12:00-12:03 |
12:04 | 12:02 | 12:02 | 12:00-12:03 |
12:05 | 12:05 | 12:05 | 12:03-12:06 |
12:06 | Fuente inactiva | 12:05 | 12:03-12:06 |
12:07 | Fuente inactiva | 12:00 | 12:06-12:09 |
12:08 | Fuente inactiva | 12:00 | 12:06-12:09 |
12:09 | 12:09 | 12:09 | 12:09-12:12 |
Durante el intervalo 12:00-12:03, la fuente permanece inactiva durante tres minutos, pero el procesador de flujo no cierra la ventana porque el tiempo de procesamiento no ha pasado el final del intervalo de la ventana y la fuente no permanece inactiva después de que termina el intervalo de la ventana. Cuando la marca de agua avance a 12:05, la ventana se cerrará normalmente y se abrirá la ventana 12:03-12:06.
Cuando la fuente queda inactiva en 12:06, permanece inactiva hasta 12:07, lo que activa el tiempo de espera de inactividad y avanza la marca de agua a 12:06.