> For the complete MongoDB documentation index, see www.mongodb.com/docs/llms.txt

# `$tumblingWindow` Stage (Stream Processing)

## Definition

The [`$tumblingWindow`](https://www.mongodb.com/docs/atlas/atlas-stream-processing/sp-agg-tumbling.md#mongodb-pipeline-pipe.-tumblingWindow) stage specifies a [tumbling window](https://www.mongodb.com/docs/atlas/atlas-stream-processing/windows.md#std-label-tumbling-windows) for aggregation of data. Atlas Stream Processing windows are stateful, can be recovered if interrupted, and have mechanisms for processing late-arriving data. You must apply all other aggregation queries to your streaming data within this window stage.

A `$tumblingWindow` pipeline stage has the following prototype form:

```json
{
  "$tumblingWindow": {
    "boundary": "eventTime" | "processingTime",
    "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>"
    }
  }
}
```

Alternatively, a `$tumblingWindow` pipeline stage can have `allowedLateness` and `idleTimeout` fields with an integer value of 0, as shown below:

```json
{
  "$tumblingWindow": {
    "boundary": "eventTime" | "processingTime",
    "interval": {
      "size": <int>,
      "unit": "<unit-of-time>"
    },
    "pipeline" : [
      <aggregation-stage-array>
    ],
    "offset": {
      "offsetFromUtc": <int>,
      "unit": "<unit-of-time>"
    },
    "idleTimeout": 0,
    "allowedLateness": 0
  }
}
```

## Syntax

The `$tumblingWindow` stage takes a document with the following fields:

| Field | Type | Necessity | Description |
| --- | --- | --- | --- |
| `boundary` | string | Optional | String specifying whether window boundaries are determined by event time or processing time. Value can be either `eventTime` or `processingTime`. See [stream processing timing](https://www.mongodb.com/docs/atlas/atlas-stream-processing/architecture.md#std-label-atlas-sp-timing) to learn more. If omitted, this field defaults to `eventTime`. `idleTimeout` and `allowedLateness` fields can't be set when `boundary` is set to `processingTime`. |
| `interval` | document | Required | Document specifying the interval of a [tumbling window](https://www.mongodb.com/docs/atlas/atlas-stream-processing/windows.md#std-label-tumbling-windows) as a combination of a size and a unit of time where: The value of `size` must be a non-zero positive integer.; The value of `unit` must be one of the following:`"ms"` (millisecond); `"second"`; `"minute"`; `"hour"`; `"day"` For example, a `size` of `20` and a `unit` of `second` sets each window to remain open for 20 seconds. |
| `pipeline` | array | Required | Nested aggregation pipeline evaluated against the messages within the window. |
| `offset` | document | Optional | Document specifying a time offset for window boundaries relative to UTC. The document is a combination of the size field `offsetFromUtc` and a unit of time where: The value of `offsetFromUtc` must be a non-zero positive integer.; The value of `unit` must be one of the following:`"ms"` (millisecond); `"second"`; `"minute"`; `"hour"` For example, an `offsetFromUtc` of `8` and a `unit` of `hour` generates boundaries that are shifted eight hours ahead of UTC. If you do not specify an offset, the window boundaries align with UTC. |
| `idleTimeout` | document | Optional | Document specifying how long to wait before closing windows if `$source` is idle. Define this setting as a combination of a `size` and a `unit` of time where: The value of `size` must be a non-zero positive integer.; The value of `unit` can be one of the following:`"ms"` (millisecond); `"second"`; `"minute"`; `"hour"`; `"day"` If you set `idleTimeout`, Atlas Stream Processing closes open windows only if `$source` is idle for longer than the greater of either the remaining window time or `idleTimeout` time. The idle timer starts as soon as `$source` goes idle. For example, consider a 12:00 pm to 1:00 pm window and `idleTimeout` time of 2 hours. If the last event is at 12:02 pm after which `$source` goes idle, the remaining window time is 58 minutes. Atlas Stream Processing closes the window after 2 hours of idleness at 2:02 pm, which is longer than the remaining window time and the `idleTimeout` time. If the `idleTimeout` time is set to only 10 minutes, Atlas Stream Processing closes the window after 58 minutes of idleness at 1:00 pm, which is longer than `idleTimeout` time and the remaining window time. Alternatively, you can define this setting with an integer value of 0. See the [pipeline defintion](https://www.mongodb.com/docs/atlas/atlas-stream-processing/sp-agg-tumbling.md#std-label-atlas-sp-agg-tumbling-def) for more information. To see how an idleness timeout interacts with watermarks over the course of a window, see [Idleness Timeout.](https://www.mongodb.com/docs/atlas/atlas-stream-processing/architecture.md#std-label-atlas-sp-idle-timeout) |
| `allowedLateness` | document | Optional | Document that specifies how long to keep [windows](https://www.mongodb.com/docs/atlas/atlas-stream-processing/windows.md#std-label-atlas-sp-windows) generated from the source open to accept late-arriving data after processing documents for window end time. If omitted, defaults to 3 seconds. Alternatively, you can define this setting with an integer value of 0. See the [pipeline defintion](https://www.mongodb.com/docs/atlas/atlas-stream-processing/sp-agg-tumbling.md#std-label-atlas-sp-agg-tumbling-def) for more information. To see how allowed lateness delays a window closing relative to the watermark, see [Allowed Lateness.](https://www.mongodb.com/docs/atlas/atlas-stream-processing/architecture.md#std-label-atlas-sp-allowed-late) |

## Behavior

Atlas Stream Processing supports only one [window](https://www.mongodb.com/docs/atlas/atlas-stream-processing/windows.md#std-label-atlas-sp-windows) stage per pipeline.

When you apply the [`$group`](https://www.mongodb.com/docs/manual/reference/operator/aggregation/group.md#mongodb-pipeline-pipe.-group) stage to your window stage, a single [group key](https://www.mongodb.com/docs/manual/reference/operator/aggregation/group/) has a limit of 100 megabytes of RAM.

Support for certain aggregation stages might be limited or unavailable within windows. To learn more, see [Supported Aggregation Pipeline Stages.](https://www.mongodb.com/docs/atlas/atlas-stream-processing/stream-aggregation-stages.md#std-label-atlas-sp-aggregation-support)

In the event of a service interruption, you can resume the internal pipeline of a window from its state at the point of interruption. To learn more, see [Checkpoints.](https://www.mongodb.com/docs/atlas/atlas-stream-processing/architecture.md#std-label-atlas-sp-checkpoints)

## Examples

A streaming data source generates detailed weather reports from various locations, conformant to the schema of the [Sample Weather Dataset](https://www.mongodb.com/docs/manual/sample-data/sample-weather.md#std-label-sample-weather). The following aggregation has three stages:

1. The [`$source`](https://www.mongodb.com/docs/atlas/atlas-stream-processing/sp-agg-source.md#mongodb-pipeline-pipe.-source) stage establishes a connection with the [Apache Kafka](https://kafka.apache.org/) broker collecting these reports in a topic named `my_weatherdata`, exposing each record as it is ingested to the subsequent aggregation stages.

2. The [`$tumblingWindow`](https://www.mongodb.com/docs/atlas/atlas-stream-processing/sp-agg-tumbling.md#mongodb-pipeline-pipe.-tumblingWindow) stage defines consecutive windows with 30-second duration. Each window executes an internal `pipeline`, which finds the average, median, maximum, and minimum `atmosphericPressureObservation.altimeterSetting.value` for the duration of that window. The `pipeline` then outputs a single document with an `_id` equivalent to the start timestamp of the window it represents and the specified values for that window.

3. The [`$merge`](https://www.mongodb.com/docs/manual/reference/operator/aggregation/merge.md#mongodb-pipeline-pipe.-merge) stage writes the output to an Atlas collection named `stream` in the `sample_weatherstream` database. If no such database or collection exist, Atlas creates them.

```js
{
  '$source': {
    connectionName: 'sample_weatherdata',
    topic: 'my_weatherdata',
    tsFieldName: 'ingestionTime'
  }
},
{
  '$tumblingWindow': {
    interval: {
      size: 30,
      unit: "second"
    },
    pipeline: [{
      $group: {
        _id:
  { $meta: "stream.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'
    }
  }
}
```

To view the documents in the resulting `sample_weatherstream.stream` collection, connect to your Atlas cluster and run the following command:

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

**Output:**

```json
{
   _id: ISODate('2024-09-26T16:34:00.000Z'),
   averagePressure: 5271.47894736842,
   maxPressure: 9999.9,
   medianPressure: 1015.9,
   minPressure: 1015.9
 },
 {
   _id: ISODate('2024-09-26T16:34:30.000Z'),
   averagePressure: 5507.9,
   maxPressure: 9999.9,
   medianPressure: 1015.9,
   minPressure: 1015.9
 }
```

**Note:**

The preceding is a representative example. Streaming data are not static, and each user sees distinct documents.