/ /

`$tumblingWindow` Estágio (Stream Processing)

Definição

The $tumblingWindow stage specifies a tumbling window 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.

$tumblingWindow

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

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

Como alternativa, uma etapa de pipeline $tumblingWindow pode ter os campos allowedLateness e idleTimeout com um valor inteiro de 0, conforme demonstrado abaixo:

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

Sintaxe

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

Campo	Tipo	necessidade	Descrição
`boundary`	string	Opcional	String que especifica se os limites da janela são determinados pelo tempo de evento ou pelo tempo de processamento. O valor pode ser `eventTime` ou `processingTime`. Consulte o tempo de processamento de fluxo para saber mais. Se omitido, este campo é padronizado como `eventTime`. `idleTimeout` e os campos `allowedLateness` não podem ser definidos quando `boundary` está definido como `processingTime`.
`interval`	documento	Obrigatório	Documento que especifica o intervalo de uma janela de tombamento como uma combinação de um tamanho e uma unidade de tempo em que: O valor de `size` deve ser um número inteiro positivo diferente de zero. O valor de `unit` deve ser um dos seguintes: `"ms"` (milésimo de segundo) `"second"` `"minute"` `"hour"` `"day"` Por exemplo, um `size` de `20` e um `unit` de `second` define cada janela para permanecer aberta por 20 segundos.
`pipeline`	array	Obrigatório	Pipeline de agregação aninhado avaliado em relação às mensagens dentro da janela.
`offset`	documento	Opcional	Documento que especifica uma compensação de horário para os limites da janela em relação ao UTC. O documento é uma combinação do campo de tamanho `offsetFromUtc` e uma unidade de tempo em que: O valor de `offsetFromUtc` deve ser um número inteiro positivo diferente de zero. O valor de `unit` deve ser um dos seguintes: `"ms"` (milésimo de segundo) `"second"` `"minute"` `"hour"` Por exemplo, um `offsetFromUtc` de `8` e um `unit` de `hour` gera limites que são deslocados oito horas à frente do UTC. Se você não especificar um deslocamento, os limites da janela se alinharão com UTC.
`idleTimeout`	documento	Opcional	Documento que especifica quanto tempo esperar antes de fechar Windows se `$source` estiver ocioso. Defina essa configuração como uma combinação de `size` e `unit` de tempo em que: O valor de `size` deve ser um número inteiro positivo diferente de zero. O valor de `unit` pode ser um dos seguintes: `"ms"` (milésimo de segundo) `"second"` `"minute"` `"hour"` `"day"` Se você definir `idleTimeout`, o Atlas Stream Processing fechará as Windows abertas somente se `$source` estiver ocioso por mais tempo do que o maior entre o tempo restante da janela ou o tempo `idleTimeout`. O temporizador ocioso começa assim que `$source` fica ocioso. Por exemplo, considere uma janela 12:00 pm a 1:00 pm e `idleTimeout` tempo de 2 horas. Se o último evento for às 12:02 pm, após o qual `$source` fica inativo, o tempo restante da janela é 58 minutos. O Atlas Stream Processing fecha a janela após 2 horas de inatividade às 2:02 pm, que é maior que o tempo restante da janela e o tempo `idleTimeout` . Se o tempo `idleTimeout` estiver definido para apenas 10 minutos, o Atlas Stream Processing fechará a janela após 58 minutos de inatividade às 1:00 pm, que é mais longo que o tempo `idleTimeout` e a janela restante tempo. Alternatively, you can define this setting with an integer value of 0. See the pipeline defintion for more information. Para ver como ocorre a interação de um tempo limite de inatividade com as marcas d'água ao longo de uma janela, consulte Tempo limite de inatividade.
`allowedLateness`	documento	Opcional	Documento que especifica por quanto tempo manter abertas as janelas geradas a partir da fonte para aceitar dados que chegam tarde após o processamento dos documentos até o horário de término da janela. Se omitido, o padrão é 3 segundos. Alternatively, you can define this setting with an integer value of 0. See the pipeline defintion for more information. Para ver como o atraso permitido atrasa o fechamento de uma janela em relação à marca d'água, consulte Atraso permitido.

Comportamento

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

When you apply the $group stage to your window stage, a single group key has a limit of 100 megabytes of 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

A streaming data source generates detailed weather reports from various locations, conformant to the schema of the Sample Weather Dataset. The following aggregation has three stages:

O estágio $source estabelece uma conexão com o broker do Apache Kafka coletando esses relatórios em um tópico chamado my_weatherdata, expondo cada registro à medida que ele é assimilado nos estágios de agregação subsequentes.
The $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.
The $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.

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

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'),
   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
 }

Observação

O exemplo anterior é representativo. Os dados de streaming não são estáticos e cada usuário vê documentos distintos.

Voltar

$hoppingWindow

$sessionWindow

$tumblingWindow Estágio (Stream Processing)

Definição

Sintaxe

Comportamento

Exemplos

`$tumblingWindow` Estágio (Stream Processing)