SP.createStreamProcessor()（mongoshメソッド）

定義

sp.createStreamProcessor()

バージョン 7.0 の新機能:

互換性

This method is supported in Atlas Stream Processing Workspaces.

構文

The sp.createStreamProcessor() method has the following syntax:

sp.createStreamProcessor(
  <name>,
  [
    <pipeline>
  ],
  {
    <options>
  }
)

コマンドフィールド

sp.createStreamProcessor() 次のフィールドを取ります。

フィールド	タイプ	必要性	説明
`name`	string	必須	ストリームプロセッサの論理名。これは、ストリーム処理ワークスペース内で一意である必要があります。
`pipeline`	配列	必須	データのストリーミング配信に適用するストリーム集約パイプライン。
`options`	オブジェクト	任意	ストリームプロセッサのさまざまなオプション設定を定義するオブジェクト。
`options.dlq`	オブジェクト	条件付き	ストリーム処理ワークスペース用のデッドレターキュー (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`フィールドを定義する場合に必要です。
`options.tier`	string	任意	Atlas Stream Processing がプロセッサを割り当てる階層。このオプションを宣言しない場合、Atlas Stream Processing はプロセッサを Stream Processing ワークスペースの階層に割り当てます。次のいずれかである必要があります。 `SP2` `SP5` `SP10` `SP30` `SP50` 詳細については、階層を参照してください。

動作

sp.createStreamProcessor() は、現在のストリーム処理ワークスペースに永続的な名前付きストリームプロセッサを作成します。このストリームプロセッサは sp.processor.start() で初期化できます。既存のストリームプロセッサと同じ名前でストリームプロセッサを作成しようとすると、mongosh はエラーを返します。

アクセス制御

sp.createStreamProcessor()を実行中ユーザーには atlasAdmin ロールが必要です。

例

The following example creates a stream processor named solarDemo which ingests data from the sample_stream_solar connection. The processor excludes all documents where the value of the device_id field is device_8, passing the rest to a tumbling window with a 10-second duration. Each window groups the documents it receives, then returns various useful statistics of each group. The stream processor then merges these records to solar_db.solar_coll over the mongodb1 connection.

sp.createStreamProcessor(
  'solarDemo',
  [
    {
      $source: {
        connectionName: 'sample_stream_solar',
        timeField: {
          $dateFromString: {
            dateString: '$timestamp'
          }
        }
      }
    },
    {
      $match: {
        $expr: {
          $ne: [
            "$device_id",
            "device_8"
          ]
        }
      }
    },
    {
      $tumblingWindow: {
        interval: {
          size: Int32(10),
          unit: "second"
        },
        "pipeline": [
          {
            $group: {
              "_id": {  "device_id": "$device_id" },
              "max_temp": { $max: "$obs.temp" },
              "max_watts": { $max: "$obs.watts" },
              "min_watts": { $min: "$obs.watts" },
              "avg_watts": { $avg: "$obs.watts" },
              "median_watts": {
                                $median: {
                                  input: "$obs.watts",
                                  method: "approximate"
                                }
                              }
            }
          }
        ]
      }
    },
    {
      $merge: {
        into: {
          connectionName: "mongodb1",
          db: "solar_db",
          coll: "solar_coll"
        },
        on: ["_id"]
      }
    }
  ]
)

詳細

戻る

db.collection.updateSearchIndex

SP.listConnections