스트림 프로세서 개발

전제 조건

스트림 프로세서를 생성하고 managed 다음이 필요합니다.

• 1개의 스트림 처리 작업 공간

• 스트림 프로세서를 생성하고 실행 있는 atlasAdmin 역할 가진 데이터베이스 사용자

• Atlas 클러스터

고려 사항

대부분의 스트림 프로세서 명령은 메서드 호출에서 관련 스트림 프로세서의 이름을 지정해야 합니다. 다음 섹션에서 설명하는 구문은 엄밀히 영숫자 이름을 가정합니다. 스트림 프로세서의 이름에 하이픈(-) 또는 마침표(.)와 같은 영숫자가 아닌 문자가 포함된 경우, 이름을 대괄호([])와 큰따옴표("")로 묶어야 합니다. 메서드 호출(예: sp.["special-name-stream"].stats().

대화형으로 스트림 프로세서 만들기

스트림 프로세서 생성

스트림 프로세서 시작

스트림 프로세서 중지

스트림 프로세서 수정

기존 스트림 프로세서의 다음 요소를 수정할 수 있습니다.

• 이름

• 파이프라인
• 데드 레터 대기열

스트림 프로세서를 수정하려면 다음을 수행합니다.

기본적으로 수정된 프로세서는 마지막 체크포인트에서 복원됩니다. 또는 resumeFromCheckpoint=false를 설정할 수 있으며, 이 경우 프로세서는 요약 통계만 유지합니다. 열려 있는 창이 있는 프로세서를 수정하면, 업데이트된 파이프라인에서 창이 완전히 다시 계산됩니다.

스트림 프로세서 제거

사용 가능한 스트림 프로세서 목록

작업 공간 기본값 나열

스트림 프로세서의 샘플

기존 스트림 프로세서에서 샘플링된 결과의 배열 mongosh 사용하여 STDOUT (으)로 반환하려면 sp.processor.sample() 메서드를 사용합니다. 예시 들어, 다음 명령은 proc01라는 스트림 프로세서에서 샘플링합니다.

sp.proc01.sample()

이 명령은 CTRL-C를 사용하여 취소하거나 반환된 샘플의 크기가 누적 40MB에 도달할 때까지 계속 실행됩니다. 스트림 프로세서는 샘플의 잘못된 문서를 다음 형식의 _dlqMessage 문서로 보고합니다.

 {
   _dlqMessage: {
     errInfo: {
    reason: "<reasonForError>"
     },
     doc: {
    _id: ObjectId('<group-id>'),
    ...
     },
     processorName: '<procName>',
     workspaceName: '<workspaceName>',
     dlqTime: ISODate('2024-09-19T20:04:34.263+00:00')
   }
}

데드 레터 큐 컬렉션을 정의하지 않고도 이러한 메시지를 사용하여 데이터 위생 문제를 진단할 수 있습니다.

스트림 프로세서의 통계 보기