/ /

`$sessionWindow`

Definición

La etapa $sessionWindow especifica una ventana de sesión para la agregación de datos. Las ventanas de sesión te 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 entre sus marcas de tiempo es menor que la brecha de sesión. Cuando se cierra una ventana, sus resultados se liberan a 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	Opcional	Campos para particionar el `$sessionWindow`. MongoDB procesa los documentos entrantes que comparten los mismos campos `partitionBy` juntos. Si se omite, todos los documentos se colocan en una única sesión de partición nula. `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 recibir registros adicionales con los mismos valores de `partitionBy` antes de cerrar la ventana de sesión.
`boundary`	string	Opcional	string que especifica si los límites de la ventana se determinan por el tiempo del evento o por el tiempo de procesamiento. El valor puede ser `eventTime` o `processingTime`. Consulta Stream Processing timing para obtener más información. 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 por defecto 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

Supón que hay 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 horas, si no llega ningún documento posterior para la partición A, la ventana de sesión se cierra cuando el marcador de agua alcanza 2024-01-01 01:40:00.

Si la expresión partitionBy falla para un documento de entrada, el documento se envía a la DLQ. Al procesar datos fuera de orden, un documento puede llegar con una marca de tiempo dentro del intervalo de 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 en un orden diferente al esperado. Pueden llegar dos eventos que tienen el mismo valor partitionBy y sus marcas de tiempo pueden estar separadas por más que un hueco. Inicialmente, estos eventos se colocarán en ventanas de sesión separadas. Más tarde, podría llegar otro evento que fusione las dos ventanas de sesión.

Volver

$tumblingWindow

$setStreamMeta