/ /

`$sessionWindow`

定义

The $sessionWindow stage specifies a session window for aggregation of data. Session windows allow you to run a pipeline on each "session" of activity in an input stream. Two documents are in the same session if they have the same partition, and the difference of their timestamps is less than the session gap. When a window is closed, its results are released to the next stage.

当水印推进的时间超过会话中最大文档时间戳的 gap 值加上 allowedLateness 值的持续时间时，会话窗口将关闭。关闭会话窗口的开始时间是会话中第一个事件的时间戳，结束时间是会话中最后一个事件的时间戳，再加上间隔时间。会话窗口结束时间是在会话中的最大时间戳上加上间隔。窗口结果是 $sessionWindow.pipeline 在适合该窗口的文档上的输出。

$sessionWindow

$sessionWindow 管道阶段采用以下原型形式：

{
  $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}
  }
}

语法

$sessionWindow 阶段采用包含以下字段的文档：

字段	类型	必要性	说明
`partitionBy`	表达式(expression)	Optional	用于对 `$sessionWindow` 进行分区的字段。MongoDB 会一起处理具有相同 `partitionBy` 字段的传入文档。如果省略，所有文档都会放在单个空分区会话中。 `partitionBy` 仅支持单字段或简单表达式。不支持多键嵌入式文档。
`gap`	文档	必需	该文档将关闭会话之前等待股票`partitionBy` 值的其他记录的时间量定义为 `size` 和 `unit` 的组合，其中： `size`的值必须是非零正整数。 `unit`的值必须是以下值之一： `"ms"` （毫秒） `"second"` `"minute"` `"hour"` `"day"` 例如，如果 `size` 为“1”，`unit` 为“minute”，则此阶段会等待一分钟，以获取具有相同 `partitionBy` 值的其他记录，然后再关闭会话窗口。
`boundary`	字符串	Optional	指定窗口边界是由事件时间还是处理时间决定的字符串。值可以是 `eventTime` 或 `processingTime`。请参阅流处理时间以了解更多信息。如果省略，该字段默认值为 `eventTime`。当 `boundary` 设置为 `processingTime` 时，您无法设置 `allowedLateness` 字段。
`pipeline`	阵列	必需	根据窗口内的消息对嵌套聚合管道进行评估。
`allowedLateness`	duration	Optional	该文档指定了时间量的组合，由 `size` 和 `unit` 组成，用于在窗口结束时间处理文档后，保持从源生成的窗口开放，以接受迟到的数据，其中： `size`的值必须是非零正整数。 `unit`的值必须是以下值之一： `"ms"` （毫秒） `"second"` `"minute"` `"hour"` `"day"` 示例，“3”的 `size` 和“秒”的 `unit` 会在迟到记录的间隔后等待 3 秒，然后再将记录移动到下一阶段。如果省略，则默认值为 3 秒。

行为

每个到达 $sessionWindow 阶段的输入文档都会分配一个来自 partitionBy 表达式的分区。每个文档都会根据其分区和时间戳分配到一个会话窗口。这可能是新的或现有的会话窗口。关闭窗口时，其结果将发布到下一个阶段。窗口结果是管道在窗口中的文档上的输出。

当水印推进的间隙时间和 allowedLateness 值超过会话中最大文档时间戳时，会话窗口将关闭。已关闭的会话窗口的开始时间是会话中第一个事件的时间戳。已关闭的会话窗口的结束时间是会话中最后一个事件的时间戳加上间隔时间。会话窗口结束时间将间隔时间添加到会话中的最大时间戳中。

注意

假设分区 A 有一个会话窗口，其最大时间戳为 2024-01-01 00:40:00。如果间隔 1 小时后分区 A 没有后续文档到达，当水印达到 2024-01-01 01:40:00 时，会话窗口将关闭。

如果输入文档的 partitionBy表达式失败，则将该文档发送到 DLQ。处理无序数据时，文档可以在同一分区的现有会话Windows的间隙内到达并带有时间戳。这将合并会话Windows。

如果文档到达 $sessionWindow 阶段，其时间戳小于源中最新水印的值，并且该文档没有打开的会话，则该文档将被发送到 DLQ。当文档到达 $sessionWindow 阶段时，会设立其流元 window.partition、window.start 和 window.end 字段。

事件可能会无序到达。两个具有相同 partitionBy 值的事件可能会到达，并且它们的时间戳可能相隔超过间隔。最初，这些事件将被放置在单独的会话窗口中。稍后，可能会有另一个事件到来，合并这两个会话窗口。