$hoppingWindow

Definição

O estágio $hoppingWindow especifica umajanela de salto para agregação de dados. As janelas do Atlas Stream Processing são com estado, podem ser recuperadas se interrompidas e têm mecanismos para processar dados atrasados. Você deve aplicar todas as outras queries de agregação aos seus dados de streaming dentro desse estágio da janela.

$hoppingWindow

Um estágio de pipeline do $hoppingWindow tem a seguinte forma de protótipo:

{
  "$hoppingWindow": {
    "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>"
    },
  }
}

Sintaxe

O estágio $hoppingWindow recebe um documento com os seguintes campos:

Comportamento

O Atlas Stream Processing permite apenas um estágio de janela por pipeline.

Quando você aplica o estágio $group ao estágio da janela, uma única chave de grupo tem um limite de 100 megabytes de RAM.

O suporte para determinados estágios de agregação pode ser limitado ou indisponível no Windows. Para saber mais, consulte Estágios de aggregation pipeline suportados.

No caso de uma interrupção de serviço, você pode retomar o pipeline interno de uma janela de seu estado no ponto de interrupção. Para saber mais, consulte Checkpoints.

Exemplos

Uma fonte de dados de streaming gera relatórios meteorológicos detalhados a partir de vários locais, em conformidade com o esquema do Conjunto de dados meteorológicos de amostra. A seguinte aggregation tem três estágios:

1. O $source estágio estabelece uma conexão com o Apache Kafka o corretor coleta esses relatórios em um tópico chamado my_weatherdata, expondo cada registro à medida que ele é ingerido para os estágios de agregação subsequentes.

2. O estágio define janelas sobrepostas de tempo que $hoppingWindow têm 100 segundos de duração e que começam a cada 20 segundos. Cada janela executa um interno pipeline que encontra a média liquidPrecipitation.depth, conforme definido nos sample_weatherdata documentos transmitidos do Apache Kafka corretor, durante a duração de uma determinada janela. Em pipeline seguida, o gera um único documento com um _id equivalente ao carimbo de data/hora de início da janela que representa e o averagePrecipitation para essa janela.

3. O estágio $merge escreve o resultado em uma collection do Atlas chamada stream no banco de dados sample_weatherstream . Se não existir tal banco de dados ou coleção, o Atlas os criará.

pipeline = [
 { $source:
     {
         "connectionName": "streamsExampleConnectionToKafka",
         "topic": "my_weatherdata"
     }
 },
 { $hoppingWindow:
     {
       "interval": {
         "size": 100,
         "unit": "second"
       },
       "hopSize": {
         "size": 20,
         "unit": "second"
       },
       "pipeline" : [
         {
             $group: {
                 // The resulting document's _id is the $hoppingWindow's start timestamp
                 _id: "$_stream_meta.window.start",
                 averagePrecipitation: { $avg: "$liquidPrecipitation.depth" }
             }
         }
       ],
     }
 },
 { $merge:
     {
         "into":
         {
             "connectionName":"streamsExampleConnectionToAtlas",
             "db":"streamDB",
             "coll":"streamCollection"
         }
     }
  }
]

Para visualizar os documentos na coleção sample_weatherstream.stream resultante, conecte-se ao cluster do Atlas e execute o seguinte comando:

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

{
  _id: ISODate('2024-08-28T19:30:20.000Z'),
  _stream_meta: {
    source: { type: 'kafka' },
    window: {
      start: ISODate('2024-08-28T19:30:20.000Z'),
      end: ISODate('2024-08-28T19:32:00.000Z')
    }
  },
  averagePrecipitation: 2264.3973214285716
},
{
  _id: ISODate('2024-08-28T19:30:40.000Z'),
  _stream_meta: {
    source: { type: 'kafka' },
    window: {
      start: ISODate('2024-08-28T19:30:40.000Z'),
      end: ISODate('2024-08-28T19:32:20.000Z')
    }
  },
  averagePrecipitation: 2285.7061611374406
},
{
  _id: ISODate('2024-08-28T19:31:00.000Z'),
  _stream_meta: {
    source: { type: 'kafka' },
    window: {
      start: ISODate('2024-08-28T19:31:00.000Z'),
      end: ISODate('2024-08-28T19:32:40.000Z')
    }
  },
  averagePrecipitation: 2357.6940154440153
},
{
  _id: ISODate('2024-08-28T19:31:20.000Z'),
  _stream_meta: {
    source: { type: 'kafka' },
    window: {
      start: ISODate('2024-08-28T19:31:20.000Z'),
      end: ISODate('2024-08-28T19:33:00.000Z')
    }
  },
  averagePrecipitation: 2378.374061433447
}