`$sessionWindow`

정의

$sessionWindow 단계는 데이터 집계를 위한 세션 창을 지정합니다. 세션 창을 활용하면 입력 스트림의 각 '세션' 활동에서 파이프라인을 실행할 수 있습니다. 두 문서의 파티션이 동일하며, 타임스탬프의 차이가 세션 간격보다 작을 경우 동일한 세션에 속합니다. 창이 닫히면 그 결과가 다음 단계로 전달됩니다.

세션의 최대 문서 타임스탬프를 초과하여 워터마크가 gap 값과 allowedLateness 값을 더한 시간만큼 진행되면 세션 창이 닫힙니다. 닫힌 세션 창의 시작 시간은 세션의 첫 번째 이벤트의 타임스탬프이고, 종료 시간은 세션의 마지막 이벤트의 타임스탬프에 간격을 더한 것입니다. 세션 창 종료 시간은 세션 내 최대 타임스탬프에 간격을 추가합니다. 창 결과는 창에 맞는 문서에 대한 $sessionWindow.pipeline의 출력입니다.

$sessionWindow

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

{
  $sessionWindow: {
      partitionBy: "$userId",
      gap: {unit: "minute", size: 5},
      pipeline: [{$match: {ad: true}}, {$group: { _id: null, total: {$sum: "$value"}}}],
      boundary: "eventTime",
      allowedLateness: {unit: "second", size: 5}
  }
}

구문

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

필드	유형	필요성	설명
`partitionBy`	표현식	필수 사항	`$sessionWindow` 이(가) 분할될 필드입니다. 동일한 `partitionBy` 필드를 공유하는 각 수신 문서 함께 처리됩니다.
`gap`	문서	필수 사항	세션을 닫기 전에 `partitionBy` 값을 주식 추가 레코드를 기다리는 `size` 와 `unit` 의 조합으로 시간을 정의하는 문서입니다. `size` 값은 0이 아닌 양의 정수여야 합니다. `unit` 의 값은 다음 중 하나여야 합니다. `"ms"` (밀리초) `"second"` `"minute"` `"hour"` `"day"` 예를 들어, `size`가 "1"이고 `unit`이 "분"인 경우, 이 단계는 동일한 `partitionBy` 값을 가진 추가 레코드를 1분 동안 기다린 후 세션 창을 닫습니다.
`boundary`	문자열	옵션	창 경계를 이벤트 시간으로 결정할지 처리 시간으로 결정할지 지정하는 string입니다. 값은 `eventTime` 또는 `processingTime`일 수 있습니다. 자세한 내용을 보려면 스트림 처리 타이밍을 참조하세요. 이 필드를 생략하면 기본값은 `eventTime`입니다. `boundary`가 `processingTime`으로 설정된 경우 `allowedLateness` 필드를 설정할 수 없습니다.
`pipeline`	배열	필수 사항	창 내의 메시지에 대해 평가된 중첩 집계 파이프라인입니다.
`allowedLateness`	기간	옵션	`size` 및 `unit`의 조합으로 소스에서 생성된 창을 문서 종료 시간에 대해 문서를 처리한 후 늦게 도착한 데이터를 받기 위해 얼마나 오래 열어둘지 지정하는 문서입니다. `size` 값은 0이 아닌 양의 정수여야 합니다. `unit` 의 값은 다음 중 하나여야 합니다. `"ms"` (밀리초) `"second"` `"minute"` `"hour"` `"day"` 예시 를 들어, '3'의 `size` 와 'second'의 `unit` 는 레코드를 다음 단계로 이동하기 전에 늦게 도착하는 레코드에 대한 간격 후 3 초 동안 대기합니다. 생략하면 기본값은 3초입니다.

행동

$sessionWindow 단계에 도달한 각 입력 문서에는 partitionBy 표현식에 따라 파티션이 할당됩니다. 각 문서는 파티션과 타임스탬프를 기준으로 세션 창에 할당됩니다. 이는 새 세션 창일 수도 있고, 기존 세션 창일 수도 있습니다. 창이 닫히면 그 결과가 다음 단계로 전달됩니다. 창 결과는 해당 창의 문서에 대한 파이프라인의 출력입니다.

워터마크가 세션의 최대 문서 타임스탬프를 초과하여 간격 및 allowedLateness 값만큼 진행되면 세션 창이 닫힙니다. 닫힌 세션 창의 시작 시간은 세션의 첫 번째 이벤트의 타임스탬프고, 종료 시간은 세션의 마지막 이벤트의 타임스탬프에 간격을 더한 것입니다. 세션 창 종료 시간은 세션 내 최대 타임스탬프에 간격을 추가합니다.

참고

파티션 A에 대해 최대 타임스탬프가 2024-01-01 00:40:00인 세션 창이 있다고 가정합니다. 간격이 1시간일 경우, 이후 파티션 A에 문서가 도착하지 않으면 워터마크가 2024-01-01 01:40:00에 도달할 때 세션 창이 닫힙니다.

입력 문서 에 대해 partitionBy 표현식 실패하면 문서 DLQ로 전송됩니다. 순서가 맞지 않는 데이터를 처리 때 동일한 파티션의 기존 세션 Windows 간격 내에 타임스탬프와 함께 문서 도착할 수 있습니다. 이렇게 하면 세션 Windows 병합됩니다.

문서 소스의 가장 최근 워터마크 값보다 작은 타임스탬프로 $sessionWindow 단계에 도착하고 해당 문서 에 대한 열린 세션이 없는 경우 문서 DLQ로 전송됩니다. 문서 $sessionWindow 단계에 도달하면 스트림 메타 window.partition, window.start 및 window.end 필드가 설정하다 됩니다.

이벤트는 순서대로 도착하지 않을 수 있습니다. partitionBy 값이 동일한 두 이벤트가 도착할 수 있으며, 그 타임스탬프 간 차이는 간격보다 클 수 있습니다. 처음에는 이러한 이벤트가 서로 다른 세션 창에 배치됩니다. 이후 두 세션 창을 병합하는 또 다른 이벤트가 도착할 수 있습니다.

돌아가기

$externalFunction

보안