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

定義

sp.processor.modify()

現在のストリームプロセシングワークスペースにあるという名前の Stream プロセッサを変更します。

互換性

このメソッドは 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`	配列	任意	ストリーミングデータに適用する集計ステージの配列。最後のステージは Sink ステージである必要があります。詳しくは、「 Atlas Stream Processing の集計」を参照してください。
`name`	string	任意	ストリームプロセッサの新しい名前。
`dlq`	オブジェクト	任意	ストリームプロセッサ用にデッドレターキュー (DLQ)を設定するオブジェクト。既存のデッドレターキュー (DLQ)を削除するには、空のオブジェクト（`{}` ）を渡します。
`dlq.connectionName`	string	条件付き	接続レジストリ内の接続を識別するラベル。この接続は Atlas クラスターを参照必要があります。デッドレターキュー (DLQ)を設定する場合は必須です。
`dlq.db`	string	条件付き	`dlq.connectionName` で指定されたクラスター上の 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 という名前の停止したストリームプロセッサを変更して、$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