ストリーム プロセッサの開発

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

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

Atlas UI でストリーム プロセッサを作成するには、Atlas プロジェクトのStream Processingページに移動し、Stream Processing ワークスペースのペインでManageをクリックします。

ストリーム プロセッサを構成するには、Visual Builder または JSONエディターのどちらかを使用するかを選択できます。

{
  "$source": {
    "connectionName": "sample_stream_solar"
  }
}

{
  "$match": {
    "obs.watts": { "$gt": 300 }
  }
}

{
  "$merge": {
    "into": {
      "connectionName": "demoConnection",
      "db": "demoDb",
      "coll": "demoColl"
    }
  }
}

{
  "name": "solarDemo",
  "pipeline": [
    {
      "$source": {
        "connectionName": "sample_stream_solar"
      }
    },
    {
      "$tumblingWindow": {
        "interval": {
          "size": 10,
          "unit": "second"
        },
        "pipeline": [
          {
            "$group": {
              "_id": "$group_id",
              "max_watts": { "$max": "$obs.watts" },
              "min_watts": { "$min": "$obs.watts" }
            }
          }
        ]
      }
    },
    {
      "$merge": {
        "into": {
          "connectionName": "mongodb1",
          "db": "solarDb",
          "coll": "solarColl"
        }
      }
    }
  ]
}

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 つのストリーム プロセッサを起動 および オプションを使用して 1 つのストリーム プロセッサを起動 エンドポイントとなる接続されたデバイスを提供します。

• オプションなしでストリーム プロセッサを起動するには、1 つのストリーム プロセッサを起動を使用してください。

• $sourceステージのプロパティを変更するには、ストリーム プロセッサを起動し、オプション付きで1つのストリーム プロセッサを起動を使用します。このエンドポイントとなる接続されたデバイスは、startAfter または startAtOperationTime プロパティの変更をサポートしています。

Atlas UI でストリーム プロセッサを開始するには、 Atlas プロジェクトのStream Processingページに移動し、Stream Processing ワークスペースのペインでManageをクリックして、それに定義されているストリーム プロセッサのリストを表示します。

次に、ストリーム プロセッサの Start アイコンをクリックします。

ストリーム プロセッサを起動するには、mongosh で sp.processor.start() メソッドを使用します。構文は次のとおりです。

sp.processor.start(<options>)

<options> は汎用の任意ドキュメントで、フィールドは基礎となる開始コマンドに渡されます。これらのフィールドは、次のとおりになります。

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

sp.proc01.start()

{ "ok" : 1 }

このメソッドは、ストリーム プロセッサが存在し、現在を実行中いない場合は { "ok": 1 } を返します。sp.processor.start()ではないストリーム プロセッサに対してSTOPPED を呼び出すと、mongosh はエラーを返します。

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

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

Atlas UI でストリーム プロセッサを一時停止するには、Atlas プロジェクトの Stream Processing ページに移動し、Stream Processing ワークスペースのペインで Manage をクリックして、それに定義されているストリーム プロセッサのリストを表示します。

次に、ストリーム プロセッサの Pause アイコンをクリックします。

を使用して既存のストリーム プロセッサを停止するには、mongosh sp.processor.stop()メソッドを使用します。構文は次のとおりです。

sp.processor.stop(<options>)

<options> は汎用の任意ドキュメントで、そのフィールドは基になる停止コマンドに渡されます。

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

sp.proc01.stop()

{ "ok" : 1 }

このメソッドは、ストリーム プロセッサが存在し、現在を実行中場合は { "ok": 1 } を返します。sp.processor.stop()ではないストリーム プロセッサに対してrunning を呼び出すと、mongosh はエラーを返します。

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

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

Atlas UI でストリーム プロセッサを削除するには、Atlas プロジェクトの Stream Processing ページに移動し、Stream Processing ワークスペースのペインで Manage をクリックして、それに定義されたストリームプロセッサのリストを表示します。

次に、ストリーム プロセッサの Delete（）アイコンをクリックします。表示される確認ダイアログで、ストリーム プロセッサの名前（solarDemo）を入力して削除することを確認し、Delete をクリックします。

を使用して既存のストリーム プロセッサを削除するには、mongosh sp.processor.drop()メソッドを使用します。構文は次のとおりです。

sp.processor.drop(<options>)

<options> は汎用の任意ドキュメントで、そのフィールドは基になる削除コマンドに渡されます。

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

sp.proc01.drop()

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

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

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

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

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

ストリーム プロセッサを一覧表示する

Atlas UI で stream processing ワークスペースに定義されているストリームプロセッサのリストを表示するには、Atlas プロジェクトの Stream Processing ページに移動し、Stream Processing ワークスペースのペインで Manage をクリックします。

ストリーム プロセッサの一覧とそのステータスが表示されます。

mongoshを使用して現在の Stream Processing ワークスペースで使用可能なストリーミング配信するプロセッサをすべて一覧表示するには、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
}

[ Stream Processing ワークスペースの詳細 ] エンドポイントには、defaultTierSize やmaxTierSize などのストリーム処理ワークスペースの詳細が一覧表示されます。

Atlas UIでストリーム処理ワークスペースの defaultTierSize と maxTierSize を表示するには、 Atlasプロジェクトの Stream Processing ページに移動します。

defaultTierSize と maxTierSize は、ストリーム処理ワークスペース カードの下部に表示されます。

を使用してストリーム処理ワークスペースのmongosh defaultTierSizeとmaxTierSize を表示するには、sp.listWorkspaceDefaults() メソッドを使用します。

sp.listWorkspaceDefaults()

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

• defaultTierSize

• maxTierSize

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

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

ストリーム プロセッサのモニタリングを表示するには、Atlasプロジェクトの Stream Processing ページにGo、 Monitoringタブを開きます。 次に、ページ左上にある Stream processor ドロップダウン リストからストリーム プロセッサを選択します。

mongosh を使用して既存のストリーム プロセッサの現在のステータスを要約するドキュメントを返すには、sp.processor.stats() メソッドを使用します。構文は次のとおりです。

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

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

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

たとえば、次の例は、inst01という名前のStream Processing ワークスペース上の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
  },
}

この例には、ストリーム プロセッサの全体的な統計が示されています。

個々の演算子がどのように動作するか、または各ターゲットが処理するトラフィックの量を確認するには、verbose オプションを使用して sp.<streamprocessor>.stats() を呼び出し、stats.operatorStats と、一部の演算子では stats.operatorStats.targetStats を検査します。

例、ソース演算子の場合、stats.operatorStats.targetStats は、すべての一意の db/coll またはすべての一意のトピックの inputMessageCount フィールドと inputMessageSize フィールドを収集します。

{
  "name" : "KafkaConsumerOperator",
  "inputMessageCount" : NumberLong(100),
  "inputMessageSize" : 100352,
  "targetStats" : [
    {
    "topic" : "outputTopic1",
    "inputMessageCount" : NumberLong(100),
    "inputMessageSize" : 100352
    }
  ],
  ...
}

また、Sink 演算子の場合、stats.operatorStats.targetStats は、すべての一意の db/coll またはすべての一意のトピックの outputMessageCount フィールドと outputMessageSize フィールドを収集します。

{
  "name" : "MergeOperator",
  "inputMessageCount" : NumberLong(10),
  "inputMessageSize" : 1744,
  "targetStats" : [
    {
      "db" : "cust1",
      "coll" : "outColl1",
      "outputMessageCount" : NumberLong(3),
      "outputMessageSize" : 1748
    },
    {
      "db" : "cust2",
      "coll" : "outColl2",
      "outputMessageCount" : NumberLong(4),
      "outputMessageSize" : 2241
    }
  ],
  ...
}