스트림 프로세서 관리

Atlas Administration API는 스트림 프로세서를 생성하기 위한 엔드포인트를 제공합니다.

단일 스트림 프로세서 생성

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)

Atlas 관리 API는 스트림 프로세서를 시작하기 위한 엔드포인트를 제공합니다.

단일 스트림 프로세서 시작

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

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

sp.proc01.start()

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

• true 스트림 프로세서가 존재하고 현재 실행 중이 아닐 때

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

Atlas Administration API는 스트림 프로세서를 중지하기 위한 엔드포인트를 제공합니다.

단일 스트림 프로세서 중지

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

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

sp.proc01.stop()

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

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

• false 스트림 프로세서가 존재하지 않거나 스트림 프로세서가 현재 실행 중이 아닐 때

Atlas 관리 API는 스트림 프로세서를 삭제하기 위한 엔드포인트를 제공합니다.

단일 스트림 프로세서 삭제

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

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

sp.proc01.drop()

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

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

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

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

Atlas Administration API는 사용 가능한 모든 스트림 프로세서를 나열하는 엔드포인트를 제공합니다.

단일 스트림 프로세서 삭제

현재 스트림 처리 인스턴스에서 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
}

Atlas Administration API는 스트림 프로세서의 통계를 볼 수 있는 엔드포인트를 제공합니다.

단일 스트림 프로세서 가져오기

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
  },
}