`$sessionWindow`

Definición

La $sessionWindow stage especifica una ventana de sesión para la agregación de datos. Las ventanas de sesión permiten ejecutar una pipeline en cada "sesión" de actividad en un flujo de entrada. Dos documentos están en la misma sesión si tienen la misma partición, y la diferencia de sus marcas de tiempo es menor que el intervalo de la sesión. Cuando se cierra una ventana, sus resultados se liberan para la siguiente etapa.

Una ventana de sesión se cierra cuando la watermark avanza una duración igual al valor gap más el valor allowedLateness más allá de la marca de tiempo máxima del documento en la sesión. El tiempo de inicio de una ventana de sesión cerrada es el timestamp del primer evento en la sesión, y el tiempo de finalización es el timestamp del último evento en la sesión, más el intervalo. El tiempo de finalización de la ventana de sesión suma el intervalo al timestamp máximo de la sesión. Los resultados de la ventana son la salida de $sessionWindow.pipeline en los documentos que encajan en la ventana.

$sessionWindow

Una etapa de pipeline $sessionWindow tiene la siguiente forma de prototipo:

{
  $sessionWindow: {
      partitionBy: "$userId",
      gap: {unit: "minute", size: 5},
      pipeline: [{$match: {ad: true}}, {$group: { _id: null, total: {$sum: "$value"}}}],
      boundary: "eventTime",
      allowedLateness: {unit: "second", size: 5}
  }
}

Sintaxis

La etapa $sessionWindow procesa un documento con los siguientes campos:

Campo	Tipo	Necesidad	Descripción
`partitionBy`	expresión	Requerido	Campos para particionar el `$sessionWindow`. MongoDB procesa conjuntamente los documentos entrantes que comparten los mismos campos `partitionBy`. `partitionBy` solo admite expresiones de campo único o simples. No admite documentos incrustados con varias claves.
`gap`	Documento	Requerido	Documento que define la cantidad de tiempo como una combinación de un `size` y un `unit` para esperar registros adicionales que compartan `partitionBy` valores antes de finalizar la sesión, donde: El valor de `size` debe ser un número entero positivo distinto de cero. El valor de `unit` debe ser uno de los siguientes: `"ms"` (milisegundo) `"second"` `"minute"` `"hour"` `"day"` Por ejemplo, con un `size` de "1" y un `unit` de "minuto", esta etapa espera un minuto para registros adicionales con los mismos valores `partitionBy` antes de cerrar la ventana de sesión.
`boundary`	string	Opcional	Cadena que especifica si los límites de la ventana se determinan por el tiempo del evento o el tiempo de procesamiento. El valor puede ser `eventTime` o `processingTime`. Véase Stream Processing timing para aprender más. Si se omite, este campo se establece por defecto en `eventTime`. No puedes configurar el campo `allowedLateness` cuando `boundary` está configurado en `processingTime`.
`pipeline`	arreglo	Requerido	pipeline de agregación anidada evaluada contra los mensajes dentro de la ventana.
`allowedLateness`	duration	Opcional	Documento que especifica la cantidad de tiempo como una combinación de un `size` y un `unit` para mantener las ventanas generadas a partir de la fuente abiertos para aceptar datos que lleguen tarde después del procesamiento de documentos para la hora final de la ventana, donde: El valor de `size` debe ser un número entero positivo distinto de cero. El valor de `unit` debe ser uno de los siguientes: `"ms"` (milisegundo) `"second"` `"minute"` `"hour"` `"day"` Por ejemplo, un `size` de "3" y un `unit` de "segundo" espera 3 segundos después de la brecha para que los registros que llegan tarde antes de pasar los registros a la siguiente etapa. Si se omite, el valor predeterminado es 3 segundos.

Comportamiento

A cada documento de entrada que llega a la etapa $sessionWindow se le asigna una partición de la expresión partitionBy. Cada documento se asigna a una ventana de sesión según su partición y marca de tiempo. Esto podría ser una ventana de sesión nueva o ya existente. Cuando se cierra una ventana, sus resultados se transfieren a la siguiente etapa. Los resultados de la ventana son la salida del pipeline en los documentos de la ventana.

Una ventana de sesión se cierra cuando el watermark avanza la brecha y los valores de allowedLateness superan la marca de tiempo máxima de un documento en la sesión. La hora de inicio de una ventana de sesión cerrada es la marca de tiempo del primer evento en la sesión. La hora de finalización de una ventana de sesión cerrada es la marca de tiempo del último evento en la sesión, más la brecha. La hora de finalización de la ventana de sesión suma la brecha a la marca de tiempo máxima en la sesión.

Nota

Supongamos que existe una ventana de sesión para la partición A, con una marca de tiempo máxima de 2024-01-01 00:40:00. Con un intervalo de 1 hora, si no llegan documentos posteriores para la partición A, la ventana de sesión se cierra cuando la marca de agua llega a 2024-01-01 01:40:00.

Si la expresión partitionBy falla en un documento de entrada, este se envía a la cola de espera (DLQ). Al procesar datos desordenados, un documento puede llegar con una marca de tiempo dentro del intervalo entre las ventanas de sesión existentes de la misma partición. Esto fusionará las ventanas de sesión.

Si un documento llega a la etapa de $sessionWindow con una marca de tiempo que es menor que el valor de la marca de agua más reciente de la fuente, y no hay una sesión abierta para ese documento, este se envía a la DLQ. Cuando un documento alcanza la etapa $sessionWindow, se establecen sus campos de metadatos de flujo window.partition, window.start y window.end.

Los eventos pueden llegar desordenados. Es posible que lleguen dos eventos con el mismo valor partitionBy y que sus marcas de tiempo estén separadas por una diferencia mayor que la indicada. Inicialmente, estos eventos se ubicarán en ventanas de sesión separadas. Posteriormente, podría llegar otro evento que fusione las dos ventanas de sesión.

Volver

$tumblingWindow

$setStreamMeta