$tumblingWindow

Definição

O estágio $tumblingWindow especifica uma janela de tumbling 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 que chegam tarde. Você deve aplicar todas as outras consultas de agregação aos seus dados de streaming nesse estágio de janela.

$tumblingWindow

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

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

Sintaxe

O estágio $tumblingWindow 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.

A compatibilidade para determinados estágios de agregação pode ser limitada ou indisponível nas janelas. Para saber mais, consulte Estágios de pipeline de agregação com compatibilidade.

No caso de uma interrupção de serviço, você pode retomar o pipeline interno de uma janela a partir do estado em que se encontrava no momento da interrupção. Para saber mais, consulte checkpoints.

Exemplos

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

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

2. O estágio $tumblingWindow define janelas consecutivas com duração de 30segundos. Cada janela executa um pipelineinterno, que localiza a média, a mediana, o máximo e o mínimo atmosphericPressureObservation.altimeterSetting.value para a duração dessa janela. O pipeline então gera um único documento com _id equivalente à data e hora de início da janela que representa e aos valores especificados para essa janela.

3. O estágio $merge grava a saída na coleção do Atlas chamada stream no banco de dados sample_weatherstream. Se não existir tal banco de dados de dados ou coleção, o Atlas os criará.

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

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

  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
 }