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

# `$hoppingWindow` Stage (Stream Processing)

## Definition

The [`$hoppingWindow`](https://www.mongodb.com/docs/atlas/atlas-stream-processing/sp-agg-hopping.md#mongodb-pipeline-pipe.-hoppingWindow) stage specifies a [hopping window](https://www.mongodb.com/docs/atlas/atlas-stream-processing/windows.md#std-label-hopping-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 `$hoppingWindow` pipeline stage has the following prototype form:

```json
{
  "$hoppingWindow": {
    "boundary": "eventTime" | "processingTime",
    "interval": {
      "size": <int>,
      "unit": "<unit-of-time>"
    },
    "hopSize": {
      "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>"
    },
  }
}
```

## Syntax

The `$hoppingWindow` 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 [hopping window](https://www.mongodb.com/docs/atlas/atlas-stream-processing/windows.md#std-label-hopping-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` can 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. |
| `hopSize` | document | Required | Document that specifies the length of the [hop](https://www.mongodb.com/docs/atlas/atlas-stream-processing/windows.md#std-label-hopping-windows) between window start times 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"` For example, a `size` of `10` and a `unit` of `second` defines a 10-second hop between window start times. |
| `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. 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. 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 [`$hoppingWindow`](https://www.mongodb.com/docs/atlas/atlas-stream-processing/sp-agg-hopping.md#mongodb-pipeline-pipe.-hoppingWindow) stage defines overlapping windows of time that are 100 seconds in duration, and which begin every 20 seconds. Each window executes an internal `pipeline` which finds the average `liquidPrecipitation.depth`, as defined in the `sample_weatherdata` documents streamed from the [Apache Kafka](https://kafka.apache.org/) broker, for the duration of a given window. The `pipeline` then outputs a single document with an `_id` equivalent to the start timestamp of the window it represents and the `averagePrecipitation` 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
pipeline = [
 { $source:
     {
         "connectionName": "streamsExampleConnectionToKafka",
         "topic": "my_weatherdata"
     }
 },
 { $hoppingWindow:
     {
       "interval": {
         "size": 100,
         "unit": "second"
       },
       "hopSize": {
         "size": 20,
         "unit": "second"
       },
       "pipeline" : [
         {
             $group: {
                 _id: { $meta: "stream.window.start" },
                 averagePrecipitation: { $avg: "$liquidPrecipitation.depth" }
             }
         }

       ],
     }
 },
 { $merge:
     {
         "into":
         {
             "connectionName":"streamsExampleConnectionToAtlas",
             "db":"streamDB",
             "coll":"streamCollection"
         }
     }
  }
]
```

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-08-28T19:30:20.000Z'),
  averagePrecipitation: 2264.3973214285716
},
{
  _id: ISODate('2024-08-28T19:30:40.000Z'),
  averagePrecipitation: 2285.7061611374406
},
{
  _id: ISODate('2024-08-28T19:31:00.000Z'),
  averagePrecipitation: 2357.6940154440153
},
{
  _id: ISODate('2024-08-28T19:31:20.000Z'),
  averagePrecipitation: 2378.374061433447
}
```

**Note:**

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