Desarrollar procesadores de flujo

Un procesador de flujo de Atlas Stream Processing aplica la lógica de un pipeline de agregación de flujo con un nombre único a los datos de transmisión. Atlas Stream Processing guarda cada definición de procesador de flujo en almacenamiento persistente para que pueda reutilizarse. Solo puede usar un procesador de flujo determinado en el espacio de trabajo de Stream Processing en el que está almacenada su definición.

Interfaz

Requisitos previos

Para crear y gestionar un procesador de flujo, debes tener:

Un espacio de trabajo de Stream Processing
Un usuario de base de datos con el rol atlasAdmin para crear y ejecutar procesadores de flujo
Un clúster Atlas

Considerations

Muchos de los comandos del procesador de flujos requieren que se especifique el nombre del procesador de flujos relevante en la invocación del método. La sintaxis descrita en las siguientes secciones supone nombres estrictamente alfanuméricos. Si el nombre de su procesador de flujo incluye caracteres no alfanuméricos, como guiones (-) o puntos (puntos) (.), debe encerrar el nombre entre corchetes ([]) y comillas double ("") en la invocación del método, como en sp.["special-name-stream"].stats().

Crear un procesador de flujos de datos de manera interactiva

Crea un Procesador de Streams

Inicia un procesador de flujo

Nota

Atlas Stream Processing descarta el estado interno de los procesadores de flujo que han sido stopped por 45 días o más. Cuando inicias uno de estos procesadores, funciona e informa estadísticas de forma idéntica a su ejecución inicial.

Detener un procesador de flujo

Nota

Modificar un procesador de flujos

Puedes modificar los siguientes elementos de un procesador de flujo existente:

Para modificar un procesador de flujo, sigue estos pasos:

Detener el procesador de flujo.

Consulte Detener un procesador de flujo.

Modificar el procesador de flujo.

Consulte el procedimiento para la interfaz que haya elegido.

Reinicia el procesador de flujos.

Consulte Iniciar un procesador de flujo.

Por defecto, los procesadores modificados se restauran desde el último punto de control. Alternativamente, se puede configurar resumeFromCheckpoint=false, en cuyo caso el procesador solo mantiene estadísticas resumidas. Cuando se modifica un procesador con ventanas abiertas, las ventanas se recomputan completamente en el pipeline actualizado.

Nota

Si cambias el nombre de un procesador de flujo para el que has configurado la alerta de El estado del procesador de flujo falló usando un Operator (que contiene expresiones coincidentes como is, contains y más), Atlas no activará alertas para el procesador de flujo renombrado si la expresión coincidente no coincide con el nuevo nombre. Para supervisar el procesador de transmisión renombrado, reconfigura la alerta.

Limitaciones

Cuando se activa la configuración por defecto resumeFromCheckpoint=true, se aplican las siguientes limitaciones:

No puedes modificar la etapa $source.
No se puede modificar el intervalo de la ventana.
No puedes remover una ventana.
Solo puedes modificar un pipeline con una ventana si esa ventana tiene una etapa $group o $sort en su pipeline interno.
No puedes cambiar un tipo de ventana existente. Por ejemplo, no puedes cambiar de $tumblingWindow a $hoppingWindow o viceversa.
Los procesadores con ventanas pueden reprocesar algunos datos como resultado de recalcular las ventanas.
Las estadísticas por operador no se conservan después de una operación de modificación.

Procedimiento

Iniciar la conmutación por error para un procesador de flujo.

Descartar un procesador de flujo

Listar los procesadores de flujo disponibles

Enumere los valores por defecto del espacio de trabajo

Muestra de un procesador de flujo

Para devolver un arreglo de resultados muestreados desde un procesador de flujo existente a STDOUT con mongosh, use el método sp.processor.sample(). Por ejemplo, el siguiente comando toma muestras de un procesador de flujo llamado proc01.

sp.proc01.sample()

Este comando se ejecuta de forma continua hasta que se cancele usando CTRL-C, o hasta que las muestras devueltas acumulen un tamaño de 40 MB. El procesador de flujos reporta documentos inválidos en la muestra en un documento _dlqMessage del siguiente formato:

 {
   _dlqMessage: {
     errInfo: {
    reason: "<reasonForError>"
     },
     doc: {
    _id: ObjectId('<group-id>'),
    ...
     },
     processorName: '<procName>',
     workspaceName: '<workspaceName>',
     dlqTime: ISODate('2024-09-19T20:04:34.263+00:00')
   }
}

Puedes utilizar estos mensajes para diagnosticar problemas de higiene de datos sin definir una colección de fila de letra muerta.

Ver estadísticas de un procesador de flujos

Nota

Volver

Administrar conexiones VPC

Aggregation Operators