/ /

`$throttle` 애그리게이션 단계(스트림 처리)

정의

$throttle

$throttle 단계는 스트림 프로세서가 데이터를 다음 단계로 전달하는 속도를 제한합니다. $throttle를 사용하여 다운스트림 시스템을 트래픽 버스트로부터 보호하고 해당 시스템에서 시행하다 속도 제한을 유지합니다.

버스트는 스트림 프로세서가 대규모 데이터 세트 의 초기 스냅샷 찍거나, 중단 후 복구하거나, 기록 데이터를 다시 처리할 때 발생할 수 있습니다. 속도 제한이 없으면 이러한 버스트가 대상 시스템을 압도하거나, 강제로 확장하다 하거나, 타사 API 할당량을 초과할 수 있습니다.

구문

$throttle 단계의 프로토타입 형식은 다음과 같습니다.

{
  $throttle: {
    bytesPerSec: <integer>,
    messagesPerSec: <integer>
  }
}

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

필드	유형	필요성	설명
`bytesPerSec`	integer	옵션	스테이지가 다운스트림 스테이지로 전달하는 초당 최대 바이트 수입니다. 양의 정수여야 합니다.
`messagesPerSec`	integer	옵션	해당 단계가 다운스트림 단계로 전달하는 초당 최대 메시지 수입니다. 각 문서 하나의 메시지로 계산됩니다. 양의 정수여야 합니다.

bytesPerSec 또는 messagesPerSec 중 하나 이상을 지정해야 합니다. 둘 다 지정하지 않으면 Atlas Stream Processing 다음 오류를 반환합니다.

$throttle requires at least one of 'bytesPerSec' or 'messagesPerSec'

행동

Atlas Stream Processing 사용자가 설정하다 각 필드 에 대해 별도의 속도 제한을 추적합니다. 각 제한은 사용자가 구성한 속도로 지속적으로 용량 회복하며, Atlas Stream Processing 각 제한을 다른 제한과 독립적으로 추적합니다.

구성된 모든 제한에 사용 가능한 용량 있는 경우에만 메시지가 $throttle 단계를 통과할 수 있습니다. 두 필드를 모두 설정하다 경우 두 제한 모두 용량 있어야 합니다. 둘 이상의 제한이 모두 소진되면 단계는 가장 많이 소진된 제한에 필요한 시간 동안 대기합니다. 해당 대기 시간 동안 모든 제한이 용량 회복하므로 대기 시간이 종료되기 전에 부족한 용량이 더 작은 제한이 복구됩니다.

이 단계는 더 많은 데이터를 전달하기 전에 1초도 기다리지 않습니다. 구성된 모든 제한의 용량 충분해지는 즉시 다음 메시지를 처리합니다.

단일 문서 bytesPerSec 제한보다 큰 경우, 스테이지는 충분한 용량 누적될 때까지 문서 보관하지 않습니다. 대신 이 단계에서는 문서 다음 단계로 전달한 후 1초 동안 기다렸다가 추가 데이터를 전달합니다.

단일 파이프라인 에서 둘 이상의 $throttle 단계를 사용할 수 있습니다. 각 단계는 이를 통과하는 데이터만 제한하므로 파이프라인 의 여러 부분에 서로 다른 제한을 적용 할 수 있습니다.

$throttle는 메인 파이프라인 에서만 사용할 수 있습니다. 다른 단계 내에 $throttle을(를) 중첩할 수 없습니다.

배압

스로틀링은 파이프라인 통한 데이터 흐름을 느리게 하여 $throttle 단계 이전 단계에 역압을 생성합니다. 소스가 스로틀 제한에서 허용하는 것보다 빠르게 데이터를 생성하는 경우 프로세서는 소스보다 뒤처지고 시간이 지남에 따라 지연이 커집니다.

무제한 지연을 방지하려면 최대값이 아닌 소스의 지속적인 처리량 에 일치하는 제한을 설정하다 . stats.changeStreamTimeDifferenceSecs를 모니터링하고 모니터링에 설명된 스로틀 통계를 통해 프로세서가 소스를 따라잡는지 확인합니다.

모니터링

Atlas Stream Processing $throttle을 사용하는 스트림 프로세서에 대해 다음 통계를 보고합니다.

통계	설명
`throttle.throttledTimeMs`	프로세서가 적극적으로 스로틀링하는 데 소비한 누적 시간(단위: 밀리초)입니다.
`throttle.throttleEvents`	구성된 속도 제한 내에서 메시지를 지연시킨 횟수입니다.

Atlas Stream Processing 프로세서 수준과 각 단계에 대한 통계를 모두 보고합니다. 파이프라인 $throttle 단계가 두 개 이상 포함된 경우 프로세서 수준 값은 단계별 값의 합계입니다. 단계별 통계는 stats.operatorStats.throttle.throttledTimeMs 및 stats.operatorStats.throttle.throttleEvents입니다.

스트림 프로세서 통계에 대해 자세히 학습 Atlas Stream Processing 모니터링 및 지표를 참조하세요.

예시

다음 파이프라인 스트림 프로세서가 Kafka 주제 에 쓰는 데이터를 초당 5 MiB 및 50 메시지로 제한합니다.

{
  $throttle: {
    bytesPerSec: 5242880,
    messagesPerSec: 50
  }
},
{
  $emit: {
    connectionName: "ordersTopic"
  }
}

총 2 MiB에 해당하는 20 문서 배치 생각해 보세요. 최근 트래픽이 두 제한의 용량 중 일부를 사용했습니다.

Limit	필요한 용량	사용 가능한 용량	부족	복구 속도	대기 필요
`messagesPerSec`	20	15	5	50/초	100ms
`bytesPerSec`	2,097,152	1,048,576	1,048,576	5,242,880/초	200ms

두 한도의 용량 충분할 때까지 이 단계를 진행할 수 없으므로 두 대기 시간 중 더 긴 200ms 동안 대기합니다. 대기하는 동안 두 제한 모두 용량 회복하므로 100ms에 불과했던 messagesPerSec 제한은 대기가 종료될 때 이미 복구된 상태입니다.

돌아가기

$setStreamMeta

$emit

$throttle 애그리게이션 단계(스트림 처리)

정의

구문

행동

배압

모니터링

예시

`$throttle` 애그리게이션 단계(스트림 처리)