$tumblingWindow

定义

$tumblingWindow阶段指定用于数据聚合的滚动窗口。 Atlas Stream Processing 窗口是有状态的，中断后可恢复，并具有处理迟到数据的机制。您必须在此窗口阶段将所有其他聚合查询应用于流数据。

$tumblingWindow

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

{
  "$tumblingWindow": {
    "interval": {
      "size": <int>,
      "unit": "<unit-of-time>"
    },
    "pipeline" : [
      <aggregation-stage-array>
    ],
    "offset": {
      "offsetFromUtc": <int>,
      "unit": "<unit-of-time>"
    },
    "idleTimeout": {
      "size": <int>,
      "unit": "<unit-of-time>"
    },
    "allowedLateness": {
      size: <int>,
      unit: "<unit-of-time>"
    }
  }
}

语法

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

行为

Atlas Stream Processing 仅支持每个管道一个窗口阶段。

当您将$group阶段应用于窗口阶段时，单个组密钥的 RAM 限制为100 MB。

Windows 中对某些聚合阶段的支持可能有限或不可用。要了解更多信息，请参阅支持的聚合管道阶段。

如果发生服务中断，您可以将窗口的内部管道从中断时的状态恢复。要了解更多信息，请参阅检查点。

举例

流数据源从不同位置生成详细的天气报告，符合样本天气数据集的模式。以下聚合分为三个阶段：

1. $source阶段与 Apache Kafka 建立连接 代理在名为my_weatherdata 的主题中收集这些报告，并在将每条记录摄取到后续聚合阶段时将其公开。

2. $tumblingWindow阶段定义持续时间为30秒的连续窗口。每个窗口都执行内部pipeline ，查找该窗口持续时间的平均值、中位数、最大值和最小值atmosphericPressureObservation.altimeterSetting.value 。然后， pipeline输出单个文档，其中的_id等于其表示的窗口的开始时间戳，以及该窗口的指定值。

3. $merge阶段会将输出写入sample_weatherstream数据库中名为stream的 Atlas 集合。如果不存在此类数据库或集合，Atlas 会进行创建。

{
  '$source': {
    connectionName: 'sample_weatherdata',
    topic: 'my_weatherdata',
    tsFieldName: 'ingestionTime'
  }
},
{
  '$tumblingWindow': {
    interval: {
      size: 30,
      unit: "second"
    },
    pipeline: [{
      $group: {
        _id: "$_stream_meta.window.start",
        averagePressure: { $avg: "$atmosphericPressureObservation.altimeterSetting.value" },
        medianPressure: {
          $median: {
            input: "$atmosphericPressureObservation.altimeterSetting.value",
            method: "approximate"
          }
        },
        maxPressure: { $max: "$atmosphericPressureObservation.altimeterSetting.value" },
        minPressure: { $min: "$atmosphericPressureObservation.altimeterSetting.value" }
      }
    }]
  }
},
{
  '$merge': {
    into: {
      connectionName: 'weatherStreamOutput',
      db: 'sample_weatherstream',
      coll: 'stream'
    }
  }
}

要查看生成的sample_weatherstream.stream集合中的文档，请连接到您的 Atlas 集群并运行以下命令：

  db.getSiblingDB("sample_weatherstream").stream.find()

{
   _id: ISODate('2024-09-26T16:34:00.000Z'),
   _stream_meta: {
     source: { type: 'kafka' },
     window: {
       start: ISODate('2024-09-26T16:34:00.000Z'),
       end: ISODate('2024-09-26T16:34:30.000Z')
     }
   },
   averagePressure: 5271.47894736842,
   maxPressure: 9999.9,
   medianPressure: 1015.9,
   minPressure: 1015.9
 },
 {
   _id: ISODate('2024-09-26T16:34:30.000Z'),
   _stream_meta: {
     source: { type: 'kafka' },
     window: {
       start: ISODate('2024-09-26T16:34:30.000Z'),
       end: ISODate('2024-09-26T16:35:00.000Z')
     }
   },
   averagePressure: 5507.9,
   maxPressure: 9999.9,
   medianPressure: 1015.9,
   minPressure: 1015.9
 }