스트림 프로세서 관리

전제 조건

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

• 스트림 처리 인스턴스

• mongosh 버전 2.0 이상

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

• Atlas 클러스터

고려 사항

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

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

sp.process() 메서드를 사용하여 대화형으로 스트림 프로세서를 만들 수 있습니다. 대화형으로 생성하는 스트림 프로세서는 다음과 같은 동작을 나타냅니다.

• 출력 및 데드 레터 큐 문서를 shell에 쓰기

• 생성 즉시 실행 시작

• 10 분 동안 또는 사용자가 중지할 때까지 실행합니다.

• 중지한 후 지속하지 않음

대화형으로 생성하는 스트림 프로세서는 프로토타입 제작을 위한 것입니다. 영구 스트림 프로세서를 만들려면 스트림 프로세서 만들기를 참조하세요.

sp.process() 의 구문은 다음과 같습니다:

sp.process(<pipeline>)

mongosh "mongodb://atlas-stream-78xq9gi1v4b4288x06a73e9f-zi30g.virginia-usa.a.query.mongodb-qa.net/?authSource=%24external&authMechanism=MONGODB-X509" \\
--tls --authenticationDatabase admin --username streamOwner

pipeline = [
 {$source: {"connectionName": "myKafka", "topic": "stuff"}},
 {$match: { temperature: 46 }},
 {
   "$emit": {
     "connectionName": "mySink",
     "topic" : "output",
   }
 }
]

sp.process(pipeline)

스트림 프로세서 생성

mongosh 을 사용하여 새 스트림 프로세서를 만들려면 sp.createStreamProcessor() 메서드를 사용합니다. 다음과 같은 구문을 가집니다.

sp.createStreamProcessor(<name>, <pipeline>, <options>)

mongosh "mongodb://atlas-stream-78xq9gi1v4b4288x06a73e9f-zi30g.virginia-usa.a.query.mongodb-qa.net/?authSource=%24external&authMechanism=MONGODB-X509" \\
--tls --authenticationDatabase admin --username streamOwner

pipeline = [
 {$source: {"connectionName": "myKafka", "topic": "stuff"}},
 {$match: { temperature: 46 }},
 {
   "$emit": {
     "connectionName": "mySink",
     "topic" : "output",
   }
 }
]

deadLetter = {
  dlq: {
    connectionName: "cluster01",
    db: "metadata",
    coll: "dlq"
  }
}

sp.createStreamProcessor("proc01", pipeline, deadLetter)

스트림 프로세서 시작

mongosh 으로 기존 스트림 프로세서를 시작하려면 sp.<streamprocessor>.start() 메서드를 사용합니다. <streamprocessor> 는 현재 스트림 처리 인스턴스에 대해 정의된 스트림 프로세서의 이름이어야 합니다.

예를 들어 proc01 이라는 스트림 프로세서를 시작하려면 다음 명령을 실행합니다.

sp.proc01.start()

이 메서드는 다음을 반환합니다.

• true 스트림 프로세서가 존재하지만 현재 실행 중이 아닌 경우.

• false 존재하지 않거나 존재하지만 현재 실행 중인 스트림 프로세서를 시작하려고 하는 경우

스트림 프로세서 중지

mongosh 으로 기존 스트림 프로세서를 중지하려면 sp.<streamprocessor>.stop() 메서드를 사용합니다. <streamprocessor> 는 현재 스트림 처리 인스턴스에 대해 정의된 현재 실행 중인 스트림 프로세서의 이름이어야 합니다.

예를 들어 proc01 이라는 스트림 프로세서를 중지하려면 다음 명령을 실행합니다.

sp.proc01.stop()

이 메서드는 다음을 반환합니다.

• true 스트림 프로세서가 존재하고 현재 실행 중인 경우

• false 스트림 프로세서가 존재하지 않거나 스트림 프로세서가 현재 실행 중이 아닌 경우.

스트림 프로세서 삭제

mongosh 으로 기존 스트림 프로세서를 삭제하려면 sp.<streamprocessor>.drop() 메서드를 사용합니다. <streamprocessor> 는 현재 스트림 처리 인스턴스에 대해 정의된 스트림 프로세서의 이름이어야 합니다.

예를 들어 proc01 이라는 스트림 프로세서를 삭제하려면 다음 명령을 실행합니다.

sp.proc01.drop()

이 메서드는 다음을 반환합니다.

• true 스트림 프로세서가 존재하는 경우.

• false 스트림 프로세서가 존재하지 않는 경우.

스트림 프로세서를 삭제하면 Atlas Stream Processing이 이에 대해 프로비저닝한 모든 리소스와 저장된 모든 상태가 함께 폐기됩니다.

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

mongosh 을 사용하여 현재 스트림 처리 인스턴스에서 사용 가능한 모든 스트림 프로세서를 나열하려면 sp.listStreamProcessors() 메서드를 사용합니다. 각 스트림 프로세서와 관련된 이름, 시작 시간, 현재 상태 및 파이프라인이 포함된 문서 목록을 반환합니다. 다음과 같은 구문을 가집니다.

sp.listStreamProcessors(<filter>)

<filter> 목록을 필터링할 필드를 지정하는 문서입니다.

sp.listStreamProcessors()

1
{
2
  id: '0135',
3
  name: "proc01",
4
  last_modified: ISODate("2023-03-20T20:15:54.601Z"),
5
  state: "RUNNING",
6
  error_msg: '',
7
  pipeline: [
8
    {
9
      $source: {
10
        connectionName: "myKafka",
11
        topic: "stuff"
12
      }
13
    },
14
    {
15
      $match: {
16
        temperature: 46
17
      }
18
    },
19
    {
20
      $emit: {
21
        connectionName: "mySink",
22
        topic: "output",
23
      }
24
    }
25
  ],
26
  lastStateChange: ISODate("2023-03-20T20:15:59.442Z")
27
},
28
{
29
  id: '0218',
30
  name: "proc02",
31
  last_modified: ISODate("2023-03-21T20:17:33.601Z"),
32
  state: "STOPPED",
33
  error_msg: '',
34
  pipeline: [
35
    {
36
      $source: {
37
        connectionName: "myKafka",
38
        topic: "things"
39
      }
40
    },
41
    {
42
      $match: {
43
        temperature: 41
44
      }
45
    },
46
    {
47
      $emit: {
48
        connectionName: "mySink",
49
        topic: "results",
50
      }
51
    }
52
  ],
53
  lastStateChange: ISODate("2023-03-21T20:18:26.139Z")
54
}

sp.listStreamProcessors({"state": "running"})

1
{
2
  id: '0135',
3
  name: "proc01",
4
  last_modified: ISODate("2023-03-20T20:15:54.601Z"),
5
  state: "RUNNING",
6
  error_msg: '',
7
  pipeline: [
8
    {
9
      $source: {
10
        connectionName: "myKafka",
11
        topic: "stuff"
12
      }
13
    },
14
    {
15
      $match: {
16
        temperature: 46
17
      }
18
    },
19
    {
20
      $emit: {
21
        connectionName: "mySink",
22
        topic: "output",
23
      }
24
    }
25
  ],
26
  lastStateChange: ISODate("2023-03-20T20:15:59.442Z")
27
}

스트림 프로세서의 샘플

기존 스트림 프로세서에서 샘플링된 결과의 배열을 mongosh 사용하여 STDOUT 으)로 반환하려면 sp.<streamprocessor>.sample() 메서드를 사용합니다. <streamprocessor> 은(는) 현재 스트림 처리 인스턴스에 대해 정의된 현재 실행 중인 스트림 프로세서의 이름이어야 합니다. 예를 들어, 다음 명령은 proc01 이라는 스트림 프로세서에서 샘플링합니다.

sp.proc01.sample()

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

 {
   _dlqMessage: {
     _stream_meta: {
       source: {
         type: "<type>"
       }
     },
     errInfo: {
       reason: "<reasonForError>"
     },
     doc: {
       _id: ObjectId('<group-id>'),
       ...
     },
     processorName: '<procName>'
   }
}

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

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

mongosh 을 사용하여 기존 스트림 프로세서의 현재 상태를 요약하는 문서를 반환하려면 sp.<streamprocessor>.stats() 메서드를 사용합니다. streamprocessor 는 현재 스트림 처리 인스턴스에 대해 정의된 현재 실행 중인 스트림 프로세서의 이름이어야 합니다. 다음과 같은 구문을 가집니다.

sp.<streamprocessor>.stats({options: {<options>}})

여기서 options 은 다음 필드가 있는 선택적 문서입니다.

출력 문서에는 다음과 같은 필드가 있습니다.

예를 들어 다음은 항목 크기가 KB 단위로 표시되는 inst01 Atlas Stream Processing 인스턴스에서 proc01 스트림 프로세서의 상태를 보여줍니다.

sp.proc01.stats(1024)
{
  ok: 1,
  ns: 'inst01',
  stats: {
    name: 'proc01',
    status: 'running',
    scaleFactor: Long("1"),
    inputMessageCount: Long("706028"),
    inputMessageSize: 958685236,
    outputMessageCount: Long("46322"),
    outputMessageSize: 85666332,
    dlqMessageCount: Long("0"),
    dlqMessageSize: Long("0"),
    stateSize: Long("2747968"),
    watermark: ISODate("2023-12-14T14:35:32.417Z"),
    ok: 1
  },
}