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

項目一覧

前提条件

Considerations
ストリームプロセッサを対話的に作成する
Atlas Stream Processing インスタンスに接続します。
パイプラインを定義します。
ストリームプロセッサを作成します。
ストリームプロセッサを作成する
ストリームプロセッサを起動する
ストリームプロセッサの停止
ストリームプロセッサを変更する
制限
ストリームプロセッサを変更するには、次の手順に従います。
ストリームプロセッサの削除
使用可能なストリームプロセッサを一覧表示する
ストリームプロセッサからのサンプル
ストリームプロセッサの統計情報を表示する

Atlas Stream Processing ストリームプロセッサは、一意の名前を持つストリーム集計パイプラインのロジックをストリーミングデータに適用します。 Atlas Stream Processing は、各ストリームプロセッサの定義を永続ストレージに保存し、再利用できるようにします。特定のストリームプロセッサは、その定義が保存されているストリームプロセシングインスタンス内でのみ使用できます。 Atlas Stream Processing は、ワーカーごとに最大4ストリームプロセッサをサポートします。この制限を超える追加のプロセッサには、Atlas Stream Processing は新しいリソースを割り当てます。

前提条件

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

Atlas Stream Processing インスタンス
mongosh バージョン2.0以上
ストリームプロセッサを作成および実行するatlasAdminロールを持つデータベースユーザー
Atlas クラスター

Considerations

多くのストリームプロセッサコマンドでは、メソッド呼び出しで関連するストリームプロセッサの名前を指定する必要があります。次のセクションで説明される構文では、厳密に英数字の名前を使用することを前提としています。ストリームプロセッサの名前にハイフン（-）や完全停止（.）など、英数字以外の文字が含まれている場合は、名前を角括弧（[]）とdouble引用符（""）で囲む必要があります。メソッド呼び出し（sp.["special-name-stream"].stats() と同様）。

ストリームプロセッサを対話的に作成する

sp.process()メソッドを使用して、ストリームプロセッサを対話的に作成できます。対話的に作成したストリームプロセッサは、次の動作を示します。

出力とデッドレターキューのドキュメントを shell に書込む
作成後すぐに実行を開始
10分間、またはユーザーがそれらを停止するまで実行します
停止した後に永続化しない

対話的に作成するストリームプロセッサは、プロトタイプ作成を目的としています。永続的なストリームプロセッサを作成するには、「ストリームプロセッサの作成」を参照してください。

sp.process() の構文は次のとおりです。

sp.process(<pipeline>)

フィールド	タイプ	必要性	説明
`pipeline`	配列	必須	データのストリーミング配信に適用するストリーム集約パイプライン。

Atlas Stream Processing インスタンスに接続します。

Atlas Stream Processing インスタンスに関連付けられた接続文字列を使用して接続し、 mongosh を使用して接続します。

例

以下のコマンドは、x.059 認証を使用して、streamOwner という名前のユーザーとしてストリームプロセシングインスタンスに接続します。

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

ユーザーのパスワードの入力を求められたら、入力します。

パイプラインを定義します。

mongoshプロンプトで、 pipelineという名前の変数に適用する集計ステージを含む配列を割り当てます。

次の例では、接続レジストリ内のmyKafka接続のstuffトピックを$sourceとして使用し、 temperatureフィールドの値が46であるレコードを照合し、処理されたメッセージをoutputに出力します。接続レジストリにあるmySink接続のトピック。

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

ストリームプロセッサを作成します。

次のコマンドは、 pipelineで定義されたロジックを適用するストリームプロセッサを作成します。

sp.process(pipeline)

ストリームプロセッサを作成する

ストリームプロセッサを作成するには:

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

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

mongoshを使用して新しいストリームプロセッサを作成するには、 sp.createStreamProcessor()メソッドを使用します。構文は次のとおりです。

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

Argument	タイプ	必要性	説明
`name`	string	必須	ストリームプロセッサの論理名。これは、ストリームプロセシングインスタンス内で一意である必要があります。この名前には、英数字のみを含める必要があります。
`pipeline`	配列	必須	データのストリーミング配信に適用するストリーム集約パイプライン。
`options`	オブジェクト	任意	ストリームプロセッサのさまざまなオプション設定を定義するオブジェクト。
`options.dlq`	オブジェクト	条件付き	ストリームプロセシングインスタンスにデッドレターキューを割り当てるオブジェクト。このフィールドは、 `options`フィールドを定義する場合に必要です。
`options.dlq.connectionName`	string	条件付き	接続レジストリ内の接続を識別する、人間が判読可能なラベル。この接続は Atlas クラスターを参照する必要があります。このフィールドは、 `options.dlq`フィールドを定義する場合に必要です。
`options.dlq.db`	string	条件付き	`options.dlq.connectionName`で指定されたクラスター上の Atlas データベースの名前。このフィールドは、 `options.dlq`フィールドを定義する場合に必要です。
`options.dlq.coll`	string	条件付き	`options.dlq.db`で指定されるデータベース内のコレクションの名前。このフィールドは、 `options.dlq`フィールドを定義する場合に必要です。

Atlas Stream Processing インスタンスに接続します。

Atlas Stream Processing インスタンスに関連付けられた接続文字列を使用して接続し、 mongosh を使用して接続します。

例

以下のコマンドは、x.059 認証を使用して、streamOwner という名前のユーザーとしてストリームプロセシングインスタンスに接続します。

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

ユーザーのパスワードの入力を求められたら、入力します。

パイプラインを定義します。

mongoshプロンプトで、 pipelineという名前の変数に適用する集計ステージを含む配列を割り当てます。

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

（任意）DLQを定義します。

mongoshプロンプトで、DLQ の次のプロパティを含むオブジェクトを割り当てます。

connectionName
databaseName
コレクション名

次の例では、 metadata.dlqデータベースコレクション内のcluster01接続を介したDLQ を定義しています。

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

ストリームプロセッサを作成します。

次のコマンドは、 pipelineで定義されたロジックを適用するproc01という名前のストリームプロセッサを作成します。処理でエラーをスローするドキュメントは、 deadLetterで定義されたDLQ に書き込まれます。

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

ストリームプロセッサを起動する

注意

Atlas Stream Processing は、45 日以上で stopped であったストリームプロセッサの内部状態を破棄します。このようなプロセッサを起動すると、最初の実行と同じように動作し、統計情報を報告します。

ストリームプロセッサを起動するには:

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 ストリームプロセッサが存在しない場合、またはストリームプロセッサが現在実行中でない場合はになります。

ストリームプロセッサを変更する

既存のストリームプロセッサの次の要素を変更できます。

ストリームプロセッサを変更するには、次の手順を行います。

ストリームプロセッサを停止します。
ストリームプロセッサに更新を適用します。
ストリームプロセッサを再起動します。

デフォルトでは、変更されたプロセッサは最後のチェックポイントから復元します。あるいは、resumeFromCheckpoint=false を設定することもできます。この場合、プロセッサは概要統計のみを保持します。オープンウィンドウを使用してプロセッサを変更すると、アップデートされたパイプライン上でウィンドウが完全に再計算されます。

制限

デフォルト設定 resumeFromCheckpoint=true が有効になっている場合、次の制限が適用されます。

$source ステージは変更できません。
ウィンドウの間隔は変更できません。
ウィンドウを削除することはできません。
ウィンドウを使用してパイプラインを変更できるのは、そのウィンドウが内部パイプラインに $group または $sort ステージのいずれかがある場合のみです。
既存のウィンドウタイプは変更できません。例では、$tumblingWindow から $hoppingWindow へ、またはその逆に変更することはできません。
Windows を使用するプロセッサでは、ウィンドウの再計算の結果として一部のデータを再処理する場合があります。

ストリームプロセッサを変更するには、次の手順に従います。

mongoshv2.3.4 + が必要です。

既存のストリームプロセッサを変更するには、sp.<streamprocessor>.modify() コマンドを使用します。 <streamprocessor> は、現在のストリーム処理インスタンスに対して定義されている停止したストリームプロセッサの名前である必要があります。

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

 sp.proc1.modify(<pipeline>, {
    resumeFromCheckpoint: bool, // optional
    name: string, // optional
    dlq: string, // optional
}})

既存のパイプラインへのステージの追加

 sp.createStreamProcessor("foo", [
   {$source: {
      connectionName: "StreamsAtlasConnection",
      db: "test",
      coll: "test"
  }},
  {$merge: {
      into: {
          connectionName: "StreamsAtlasConnection",
          db: "testout",
          coll: "testout"
      }
  }}
])
sp.foo.start();

sp.foo.stop();
sp.foo.modify([
  {$source: {
    connectionName: "StreamsAtlasConnection",
    db: "test",
    coll: "test"
  }},
  {$match: {
    operationType: "insert"
  }},
  {$merge: {
     into: {
     connectionName: "StreamsAtlasConnection",
     db: "testout",
     coll: "testout2"
     }
  }}
]);
sp.foo.start();

ストリームプロセッサの入力ソースの変更

sp.foo.modify([
  {$source: {
    connectionName: "StreamsAtlasConnection",
    db: "test",
    coll: "test",
    config: {
      startAtOperationTime: new Date(now.getTime() - 5 * 60 * 1000)
    }
  }},
  {$match: {
    operationType: "insert"
  }},
  {$merge: {
    into: {
      connectionName: "StreamsAtlasConnection",
      db: "testout",
      coll: "testout2"
    }
  }}
], {resumeFromCheckpoint: false});

ストリームプロセッサからのデッド文字キューの削除

sp.foo.stop();
sp.foo.modify({dlq: {}})
sp.foo.start();

ウィンドウの使用によるストリームプロセッサの変更

sp.foo.stop();
sp.foo.modify([
  {$source: {
    connectionName: "StreamsAtlasConnection",
    db: "test",
    coll: "test"
  }},
  {$replaceRoot: {newRoot: "$fullDocument"}},
  {$match: {cost: {$gt: 500}}},
  {$tumblingWindow: {
    interval: {unit: "day", size: 1},
    pipeline: [
     {$group: {_id: "$customerId", sum: {$sum: "$cost"}, avg: {$avg: "$cost"}}}
    ]
  }},
  {$merge: {
    into: {
    connectionName: "StreamsAtlasConnection",
    db: "testout",
    coll: "testout"
   }
  }}
], {resumeFromCheckpoint: false});
sp.foo.start();

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

1 つのストリームプロセッサを変更

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

ストリームプロセッサを削除するには:

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

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

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 }

同じストリームプロセシングインスタンスでコマンドを再度実行し、 "running"の"state"をフィルタリングすると、次の出力が表示されます。

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>',
     instanceName: '<instanceName>',
     dlqTime: ISODate('2024-09-19T20:04:34.263+00:00')
   }
}

これらのメッセージを使用して、デッドレターキューコレクションを定義せずにデータのクリーンアップの問題を診断できます。

ストリームプロセッサの統計情報を表示する

注意

ストリームプロセッサの統計を表示するには:

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

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

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

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

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

フィールド	タイプ	説明
`scale`	integer	出力内の項目のサイズに使用する単位。デフォルトでは、Atlas Stream Processing はアイテムのサイズをバイト単位で表示します。 KB 単位で表示するには、 `1024`の`scale`を指定します。
`verbose`	ブール値	出力ドキュメントの冗長レベルを指定するフラグ。 `true`に設定されている場合、出力ドキュメントには、パイプライン内の各演算子の統計を報告するサブドキュメントが含まれます。デフォルトは`false`です。

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

フィールド	タイプ	説明
`ns`	string	ストリームプロセッサが定義されている名前空間。
`stats`	オブジェクト	ストリームプロセッサの動作状態を説明するドキュメント。
`stats.name`	string	ストリームプロセッサの名前。
`stats.status`	string	ストリームプロセッサの状態。このフィールドには、次の値を指定できます。 `starting` `running` `error` `stopping`
`stats.scaleFactor`	integer	サイズフィールドが表示される単位。 `1`に設定されている場合、サイズはバイト単位で表示されます。 `1024`に設定されている場合、サイズはキロバイト単位で表示されます。
`stats.inputMessageCount`	integer	ストリームに公開されたドキュメントの数。ドキュメントは、パイプライン全体を通過する場合ではなく、 `$source`ステージを通過するとストリームに「公開」されたと見なされます。
`stats.inputMessageSize`	integer	ストリームに公開されたバイト数またはキロバイト数。バイトは、パイプライン全体を通過する場合ではなく、 `$source`ステージを通過するとストリームに「公開」と見なされます。
`stats.outputMessageCount`	integer	ストリームによって処理されたドキュメントの数。ドキュメントはストリームによってパイプライン全体を通過すると、ストリームによって「処理済み」と見なされます。
`stats.outputMessageSize`	integer	ストリームによって処理されたバイト数またはキロバイト数。バイトはパイプライン全体を通過すると、ストリームによって「処理済み」と見なされます。
`stats.dlqMessageCount`	integer	デッド文字キューに送信されたドキュメントの数。
`stats.dlqMessageSize`	integer	デッド文字キューに送信されたバイト数またはキロバイト数。
`stats.changeStreamTimeDifferenceSecs`	integer	最新の変更ストリーム再開トークンによって表されるイベント時間とoplog 内の最新のイベントとの差（秒単位）。
`stats.changeStreamState`	token	最新の変更ストリーム再開トークン。変更ストリームソースを持つストリームプロセッサにのみ適用されます。
`stats.stateSize`	integer	Windowsがプロセッサの状態を保存するために使用するバイト数。
`stats.watermark`	integer	現在の浮動小数のタイムスタンプ。
`stats.operatorStats`	配列	プロセッサパイプライン内の各演算子の統計情報。 Atlas Stream Processing では、 `verbose`オプションで渡された場合にのみこのフィールドが返されます。 `stats.operatorStats` は、多数のコア`stats`フィールドの演算子ごとのバージョンを提供します。 `stats.operatorStats.name` `stats.operatorStats.inputMessageCount` `stats.operatorStats.inputMessageSize` `stats.operatorStats.outputMessageCount` `stats.operatorStats.outputMessageSize` `stats.operatorStats.dlqMessageCount` `stats.operatorStats.dlqMessageSize` `stats.operatorStats.stateSize` `stats.operatorStats` includes the following unique fields: `stats.operatorStats.maxMemoryUsage` `stats.operatorStats.executionTime` `stats.operatorStats` also includes the following fields given that you have passed in the `verbose` option and your processor includes a window stage: `stats.minOpenWindowStartTime` `stats.maxOpenWindowStartTime`
`stats.operatorStats.maxMemoryUsage`	integer	演算子の最大メモリ使用量（バイトまたはキロバイト単位）。
`stats.operatorStats.executionTime`	integer	演算子の合計実行時間（秒単位）。
`stats.minOpenWindowStartTime`	date	The start time of the minimum open window. This value is optional.
`stats.maxOpenWindowStartTime`	date	The start time of the maximum open window. This value is optional.
`stats.kafkaPartitions`	配列	Apache Kafka のオフセット情報プロバイダーのパーティション。は`kafkaPartitions` Apache Kafka を使用する接続にのみ適用されますソース。
`stats.kafkaPartitions.partition`	integer	Apache Kafka トピックのパーティション番号。
`stats.kafkaPartitions.currentOffset`	integer	指定されたパーティションに対するストリームプロセッサのオフセット。この値は、ストリームプロセッサが処理した以前のオフセットに`1`を加えた値に等しくなります。
`stats.kafkaPartitions.checkpointOffset`	integer	ストリームプロセッサが Apache Kafka に最後にコミットしたオフセット指定されたパーティションのプロバイダーとチェックポイント。このオフセットを超えるすべてのメッセージは最後のチェックポイントに記録されます。
`stats.kafkaPartitions.isIdle`	ブール値	The flag that indicates whether the partition is idle. This value defaults to `false`.

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

戻る

接続の管理

VPCピアリング接続の管理

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	}

前提条件

Considerations

ストリーム プロセッサを対話的に作成する

Atlas Stream Processing インスタンスに接続します。

例

パイプラインを定義します。

ストリーム プロセッサを作成します。

ストリーム プロセッサを作成する

Atlas Stream Processing インスタンスに接続します。

例

パイプラインを定義します。

（任意）DLQを定義します。

ストリーム プロセッサを作成します。

ストリーム プロセッサを起動する

注意

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

注意

ストリーム プロセッサを変更する

制限

ストリーム プロセッサを変更するには、次の手順に従います。

既存のパイプラインへのステージの追加

ストリーム プロセッサの入力ソースの変更

ストリーム プロセッサからのデッド文字キューの削除

ウィンドウの使用によるストリーム プロセッサの変更

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

使用可能なストリーム プロセッサを一覧表示する

例

ストリーム プロセッサからのサンプル

ストリーム プロセッサの統計情報を表示する

注意

ストリームプロセッサを対話的に作成する

ストリームプロセッサを作成します。

ストリームプロセッサを作成する

ストリームプロセッサを作成します。

ストリームプロセッサを起動する

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

ストリームプロセッサを変更する

ストリームプロセッサを変更するには、次の手順に従います。

ストリームプロセッサの入力ソースの変更

ストリームプロセッサからのデッド文字キューの削除

ウィンドウの使用によるストリームプロセッサの変更

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

使用可能なストリームプロセッサを一覧表示する

ストリームプロセッサからのサンプル

ストリームプロセッサの統計情報を表示する