sp.processor.modify()（mongosh メソッド）

定義

sp.processor.modify()

現在の Stream Processing Workspace で、名前付きの Stream Processor を変更します。

互換性

このメソッドは Atlas Stream Processing ワークスペースでサポートされています。

構文

sp.processor.modify()メソッドの構文は次のとおりです。

sp.processor.modify({
  pipeline: [
    <pipeline>
  ],
  name: <name>,
  dlq: {
    connectionName: <connectionName>,
    db: <db>,
    coll: <coll>
  },
  resumeFromCheckpoint: <resumeFromCheckpoint>,
  tier: <tier>
})

コマンドフィールド

sp.processor.modify() は、次のフィールドがあります。

フィールド	タイプ	必要性	説明
`pipeline`	配列	任意	ストリーミングデータに適用する集計ステージの配列。最後のステージはシンクステージである必要があります。詳細については、Stream Processing 集計を参照してください。
`name`	string	任意	ストリームプロセッサの新しい名前。
`dlq`	オブジェクト	任意	ストリームプロセッサのデッドレターキュー (DLQ) を設定するオブジェクト。既存のデッドレターキュー (DLQ)を削除するには、空のオブジェクト (`{}`)を渡します。
`dlq.connectionName`	string	条件付き	接続レジストリ内の接続を識別するラベル。この接続は MongoDB Atlas クラスターを参照する必要があります。デッドレターキュー (DLQ)を設定する場合に必要です。
`dlq.db`	string	条件付き	`dlq.connectionName`で指定されたクラスター上のMongoDB Atlasデータベースの名前。デッドレターキュー (DLQ) を設定する場合は必須です。
`dlq.coll`	string	条件付き	`dlq.db` で指定されたデータベース内のコレクションの名前。デッドレターキュー (DLQ) を設定する場合は必須です。
`resumeFromCheckpoint`	ブール値	任意	変更されたストリームプロセッサーが最後のチェックポイントから再開するかどうかを指定するフラグ。デフォルトでは、このフィールドは `true` です。`false` に設定すると、プロセッサは要約統計のみを保持します。
`tier`	string	任意	ストリームプロセッサに割り当てる階層。このオプションを指定しない場合、プロセッサは現在の階層を維持します。次のいずれかの値でなければなりません： `SP2` `SP5` `SP10` `SP30` `SP50` 詳細については、階層を参照してください。

動作

ストリームプロセッサーは、このメソッドを呼び出す前に STOPPED 状態である必要があります。pipeline 引数は、変更しないステージを含め、プロセッサーの存在するパイプライン全体を置き換えます。

デフォルトでは、変更されたプロセッサーは最後のチェックポイントから再開します。resumeFromCheckpoint を false に設定すると、変更されたプロセッサーは要約統計のみを保持します。開いているウィンドウを持つプロセッサーを変更すると、Atlas Stream Processing は更新されたパイプラインでこれらのウィンドウを再計算します。

ストリームプロセッサーを変更する場合に適用される制限については、「ストリームプロセッサーの変更」を参照してください。

アクセス制御

sp.processor.modify()を実行するには、atlasAdminロールが必要です。

例

例では、solarDemo という名前の停止した Stream Processing プロセッサーを変更しています。変更内容は次のとおりです。$match ステージを追加し、名前を変更し、階層を更新し、デッドレターキュー (DLQ) を構成します。

sp.solarDemo.modify({
  pipeline: [
    { $source: { connectionName: "sample_stream_solar" }},
    { $match: { device_id: "device_0" }},
    { $merge: { into: {
        connectionName: "cluster0",
        db: "testout",
        coll: "testout2"
    }}}
  ],
  name: "solarDemoRenamed",
  dlq: {
    connectionName: "cluster0",
    db: "testout",
    coll: "dlq"
  },
  resumeFromCheckpoint: true,
  tier: "SP10"
})

{ ok: 1 }

名前を変更したプロセッサを起動し、sp.listStreamProcessors()を実行して名前、階層、デッドレターキュー (DLQ) の変更を確認します。

sp.solarDemoRenamed.start()
sp.listStreamProcessors()

[
  {
    id: '6a39b08e6d9040e1cef8e31f',
    name: 'solarDemoRenamed',
    lastModified: ISODate('2026-06-22T22:00:46.858Z'),
    state: 'STARTED',
    tier: 'SP10',
    errorMsg: '',
    workers: [ 'worker-5f4c5bbc9d-7hg2q' ],
    pipeline: [
      { '$source': { connectionName: 'sample_stream_solar' } },
      { '$match': { device_id: 'device_0' } },
      {
        '$merge': {
          into: {
            connectionName: 'cluster0',
            db: 'testout',
            coll: 'testout2'
          }
        }
      }
    ],
    lastStateChange: ISODate('2026-06-22T22:01:16.835Z'),
    dlq: {
      connectionName: 'cluster0',
      db: 'testout',
      coll: 'dlq'
    }
  }
]

sp.processor.sample()を実行してパイプラインの変更を確認します。

sp.solarDemoRenamed.sample()

{
  device_id: 'device_0',
  group_id: 9,
  timestamp: '2026-06-22T22:01:25.828+00:00',
  max_watts: 450,
  event_type: 0,
  obs: {
    watts: 122,
    temp: 18
  }
}
{
  device_id: 'device_0',
  group_id: 3,
  timestamp: '2026-06-22T22:01:26.828+00:00',
  max_watts: 450,
  event_type: 0,
  obs: {
    watts: 377,
    temp: 7
  }
}

詳細

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

戻る

SP.processor.drop

SP.processor.sample