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

前提条件

ストリーム プロセッサを作成および管理するには、次のものが必要です。

• Atlas Stream Processing インスタンス

• mongosh バージョン2.0以上

• ストリーム プロセッサを作成および実行するatlasAdminロールを持つデータベースユーザー

• Atlas クラスター

Considerations

多くのストリーム プロセッサ コマンドでは、メソッド呼び出しで関連するストリーム プロセッサの名前を指定する必要があります。 次のセクションで説明される構文では、厳密に英数字の名前を使用することを前提としています。 ストリーム プロセッサの名前にハイフン（-）や完全停止（.）など、英数字以外の文字が含まれている場合は、名前を角括弧（[]）とdouble引用符（""）で囲む必要があります。メソッド呼び出し（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は次のフィールドを持つ任意のドキュメントです。

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

たとえば、以下は、 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
  },
}