ストリーム プロセッサの管理

Atlas管理APIは、ストリームプロセッサを作成するためのエンドポイントを提供します。

1 つのストリーム プロセッサを作成

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は、ストリームプロセッサを起動するためのエンドポイントを提供します。

1 つのストリーム プロセッサを起動

mongoshを使用して既存のストリーム プロセッサを起動するには、 sp.<streamprocessor>.start()メソッドを使用します。 <streamprocessor>は、現在のストリーム プロセシング インスタンスに定義されているストリーム プロセッサの名前である必要があります。

たとえば、 proc01という名前のストリーム プロセッサを起動するには、次のコマンドを実行します。

sp.proc01.start()

このメソッドは以下を返します。

• true ストリーム プロセッサが存在し、現在実行中でない場合。

• false 存在しない、または存在し、現在実行中のストリーム プロセッサを起動しようとした場合。

Atlas Administration APIは、ストリーム プロセッサを停止するためのエンドポイントを提供します。

1 つのストリーム プロセッサを停止

mongoshで既存のストリーム プロセッサを停止するには、 sp.<streamprocessor>.stop()メソッドを使用します。 <streamprocessor>は、現在のストリーム プロセシング インスタンスに定義されている現在実行中のストリーム プロセッサの名前である必要があります。

たとえば、 proc01という名前のストリーム プロセッサを停止するには、次のコマンドを実行します。

sp.proc01.stop()

このメソッドは以下を返します。

• true ストリーム プロセッサが存在し、現在実行中の場合。

• false ストリーム プロセッサが存在しない場合、またはストリーム プロセッサが現在実行中でない場合は になります。

Atlas管理APIは、ストリームプロセッサを削除するためのエンドポイントを提供します。

1 つのストリーム プロセッサを削除

mongoshを使用して既存のストリーム プロセッサを削除するには、 sp.<streamprocessor>.drop()メソッドを使用します。 <streamprocessor>は、現在のストリーム プロセシング インスタンスに定義されているストリーム プロセッサの名前である必要があります。

たとえば、 proc01という名前のストリーム プロセッサを削除するには、次のコマンドを実行します。

sp.proc01.drop()

このメソッドは以下を返します。

• true ストリーム プロセッサが存在する場合。

• false ストリーム プロセッサが存在しない場合。

ストリーム プロセッサを削除すると、Atlas Stream Processing がそのストリーム用にプロビジョニングしたすべてのリソースと、保存されたすべての状態が破棄されます。

Atlas Administration APIには、利用可能なすべてのストリーム プロセッサを一覧表示するためのエンドポイントが用意されています。

1 つのストリーム プロセッサを削除

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は、ストリーム プロセッサの統計を表示するためのエンドポイントを提供します。

1 つのストリーム プロセッサを取得

mongoshを使用して既存のストリーム プロセッサの現在のステータスを要約するドキュメントを返すには、 sp.<streamprocessor>.stats()メソッドを使用します。 streamprocessorは、現在のストリーム プロセシング インスタンスに定義されている現在実行中のストリーム プロセッサの名前である必要があります。 構文は次のとおりです。

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

ここで、 optionsは次のフィールドを持つ任意のドキュメントです。

出力ドキュメントには、次のフィールドがあります。

たとえば、以下は、 inst01という名前のストリーム プロセシング インスタンス上のproc01という名前のストリーム プロセッサのステータスを示しており、アイテムサイズは KB 単位で表示されています。

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