/ /

`$tumblingWindow` 단계(Stream Processing)

정의

단계는 데이터 $tumblingWindow 집계 를 위한 텀블링 창 지정합니다. Atlas Stream Processing 창은 상태를 저장하며, 중단된 경우 복구할 수 있으며, 늦게 도착하는 데이터를 처리 하는 메커니즘을 갖추고 있습니다. 이 창 단계 내에서 다른 모든 집계 쿼리를 스트리밍 데이터에 적용 해야 합니다.

$tumblingWindow

$tumblingWindow 파이프라인 단계의 프로토타입 형식은 다음과 같습니다.

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

또는 $tumblingWindow 파이프라인 단계는 아래와 같이 정수 값이 0인 allowedLateness 및 idleTimeout 필드를 가질 수 있습니다.

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

구문

$tumblingWindow 단계에서는 다음 필드가 있는 문서를 사용합니다.

필드	유형	필요성	설명
`boundary`	문자열	옵션	창 경계를 이벤트 시간으로 결정할지 처리 시간으로 결정할지 지정하는 문자열입니다. 값은 `eventTime` 또는 `processingTime`일 수 있습니다. 자세한 내용을 보려면 스트림 처리 타이밍을 참조하세요. 이 필드를 생략하면 기본값은 `eventTime`입니다. `idleTimeout` `allowedLateness` 필드는 `boundary`가 `processingTime`으로 설정된 경우 설정할 수 없습니다.
`interval`	문서	필수 사항	크기와 시간 단위의 조합으로 터블링 창 의 간격을 지정하는 문서: `size` 값은 0이 아닌 양의 정수여야 합니다. `unit` 의 값은 다음 중 하나여야 합니다. `"ms"` (밀리초) `"second"` `"minute"` `"hour"` `"day"` 예를 들어 의 `size` `20` 와 `unit` 는 `second` 각 창이 20 초 동안 열린 상태로 유지되도록 설정합니다.
`pipeline`	배열	필수 사항	창 내의 메시지에 대해 평가된 중첩 집계 파이프라인입니다.
`offset`	문서	옵션	UTC를 기준으로 창 경계의 시간 오프셋을 지정하는 문서입니다. 문서는 크기 필드 `offsetFromUtc` 와 시간 단위의 조합으로, 여기서 다음과 같습니다. `offsetFromUtc` 값은 0이 아닌 양의 정수여야 합니다. `unit` 의 값은 다음 중 하나여야 합니다. `"ms"` (밀리초) `"second"` `"minute"` `"hour"` 예를 들어 `8` 의 `offsetFromUtc` 와 `hour` 의 `unit` 는 UTC보다 8시간 앞서 이동하는 경계를 생성합니다. 오프셋을 지정하지 않으면 창 경계가 UTC에 맞춰 정렬됩니다.
`idleTimeout`	문서	옵션	`$source` 이(가) 유휴 상태인 경우 Windows 를 닫기 전에 대기할 시간을 지정하는 문서입니다. 이 설정을 시간의 `size` 와 `unit` 의 조합으로 정의합니다. `size` 값은 0이 아닌 양의 정수여야 합니다. `unit` 의 값은 다음 중 하나일 수 있습니다. `"ms"` (밀리초) `"second"` `"minute"` `"hour"` `"day"` `idleTimeout`을(를) 설정하면 `$source` 이(가) 남은 창 시간 또는 `idleTimeout` 시간 중 더 긴 시간 동안 유휴 상태인 경우에만 Atlas Stream Processing 이 열려 있는 Windows 을 닫습니다. `$source` 이(가) 유휴 상태가 되는 즉시 유휴 타이머가 시작됩니다. 예를 들어 12:00 오후부터 1:00 오후 기간과 `idleTimeout` 시간이 2 시간이라고 가정해 보겠습니다. 마지막 이벤트가 12:02 오후에 `$source` 이(가) 유휴 상태인 경우 남은 창 시간은 58 분입니다. Atlas Stream Processing은 남은 창 시간과 `idleTimeout` 시간보다 긴 2:02 오후에 2 시간의 유휴 시간 후 창을 닫습니다. `idleTimeout` 시간이 10 분으로만 설정된 경우, Atlas Stream Processing은 `idleTimeout` 시간보다 긴 1:00 오후에 58 분의 유휴 시간 후 창을 닫으며, 나머지 창은 시간. 또는 정수 값 을 사용하여 이 설정을 정의할 수 0 있습니다. 자세한 내용은 파이프라인 정의를 참조하세요. 유휴 시간 초과가 창 기간 동안 워터마크와 상호 작용하는 방법은 유휴 시간 초과를 참조하세요.
`allowedLateness`	문서	옵션	창 종료 시간에 대해 문서를 처리한 후 늦게 도착한 데이터를 받기 위해 소스에서 생성된 창을 얼마나 오래 열어둘지 지정하는 문서입니다. 생략하면 기본값은 3초입니다. 또는 정수 값 을 사용하여 이 설정을 정의할 수 0 있습니다. 자세한 내용은 파이프라인 정의를 참조하세요. 허용된 지연이 워터마크에 대해 창 닫힘을 지연시키는 방법은 허용된 지연을 참조하세요.

행동

Atlas Stream Processing은 파이프라인당 하나의 윈도우 단계만 지원합니다.

단계를 창 단계에 $group 적용 때 단일 그룹 키의 100 RAM 은 메가바이트로 제한됩니다.

특정 집계 단계에 대한 지원은 Windows 내에서 제한되거나 사용할 수 없을 수 있습니다. 자세한 내용은 지원되는 집계 파이프라인 단계를 참조하세요.

서비스 중단 시, 중단 시점에서 창의 내부 파이프라인을 재개할 수 있습니다 자세한 내용은 체크포인트를 참조하세요.

예시

스트리밍 데이터 소스 샘플 날씨 데이터 세트의 스키마 에 따라 다양한 위치에서 자세한 일기 예보를 생성합니다. 다음 집계 세 단계로 구성됩니다.

단계는 $source 라는 주제 에서 이러한 보고서를 수집하는 Apache Kafka 브로커와의 연결을 my_weatherdata 설정하여 수집되는 각 기록 후속 집계 단계에 노출합니다.
단계는 $tumblingWindow 30지속 시간이 초인 연속 창을 정의합니다. 각 창 pipeline 내부 를 실행하여 atmosphericPressureObservation.altimeterSetting.value 해당 창 기간 동안의 평균, 중앙값 , 최대값, 최소 를 구합니다. 그런 pipeline 다음 는 나타내는 창 _id 시작 타임스탬프와 해당 창 에 지정된 값에 해당하는 가 포함된 단일 문서 출력합니다.
$merge 단계는 데이터베이스 의 이라는 Atlas 컬렉션 에 출력을 stream sample_weatherstream 씁니다. 이러한 데이터베이스 나 컬렉션 없는 경우 Atlas 생성합니다.

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

결과 sample_weatherstream.stream 컬렉션의 문서를 보려면 Atlas 클러스터에 연결하고 다음 명령을 실행하세요.

  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
 }

참고

위 사례는 대표적인 예시입니다. 스트리밍 데이터는 정적이지 않으며 각 사용자는 서로 다른 문서를 보게 됩니다.

돌아가기

$hoppingWindow

$sessionWindow

$tumblingWindow 단계(Stream Processing)

정의

구문

행동

예시

참고

`$tumblingWindow` 단계(Stream Processing)