sp.processor.modify() (método mongosh)

Definición

sp.processor.modify()

Modifica un nombre Stream Processor en el Stream Processing Workspaceactual.

Compatibilidad

Este método es compatible en espacios de trabajo de Atlas Stream Processing.

Sintaxis

El método sp.processor.modify() tiene la siguiente sintaxis:

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

Campos de comandos

sp.processor.modify() requiere los siguientes campos:

Campo	Tipo	Necesidad	Descripción
`pipeline`	arreglo	Opcional	Arreglo de etapas de agregación para aplicar a sus datos de transmisión donde la última etapa debe ser una etapa de sumidero. Para obtener más información, consulte Agregación de Stream Processing.
`name`	string	Opcional	Nuevo nombre para el procesador de streams.
`dlq`	Objeto	Opcional	Objeto que establece un fila de letra muerta para el procesador de flujos. Para remover una fila de letra muerta existente, pasa un objeto vacío (`{}`).
`dlq.connectionName`	string	Condicional	Etiqueta que identifica una conexión en el registro de conexiones. Esta conexión debe hacer referencia a un clúster de MongoDB Atlas. Obligatorio cuando se establece una fila de letra muerta.
`dlq.db`	string	Condicional	Nombre de una base de datos de MongoDB Atlas en el clúster especificado en `dlq.connectionName`. Obligatorio cuando se configura una fila de letra muerta.
`dlq.coll`	string	Condicional	Nombre de una colección en la base de datos especificada en `dlq.db`. Obligatorio cuando se configura una fila de letra muerta.
`resumeFromCheckpoint`	booleano	Opcional	Marca que especifica si el procesador de flujo modificado se reanuda desde su último punto de control. Por defecto, este campo es `true`. Cuando se establece en `false`, el procesador solo conserva las estadísticas de resumen.
`tier`	string	Opcional	El nivel que se asignará al procesador de transmisión. Si no declara esta opción, el procesador conserva su nivel actual. Debe ser uno de los siguientes valores: `SP2` `SP5` `SP10` `SP30` `SP50` Para obtener más información, consulta Niveles.

Comportamiento

El procesador de transmisión debe estar en estado STOPPED antes de invocar este método. El argumento pipeline reemplaza todo el pipeline existente del procesador, incluidas las etapas que no cambie.

Por defecto, el procesador modificado se reanuda desde su último punto de control. Si establece resumeFromCheckpoint en false, el procesador modificado solo retiene las estadísticas de resumen. Cuando se modifica un procesador con ventanas abiertas, Atlas Stream Processing vuelve a calcular esas ventanas en el pipeline actualizado.

Para conocer las limitaciones que se aplican al modificar los procesadores de transmisión, consulte Modificar un procesador de transmisión.

Control de acceso

Para ejecutar sp.processor.modify(), debe tener el rol atlasAdmin.

Ejemplo

El ejemplo cambia un procesador de transmisión detenido llamado solarDemo para agregar una etapa $match, cambiarle el nombre, actualizar su nivel y configurar una fila de letra muerta:

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 }

Inicie el procesador renombrado y, a continuación, ejecute sp.listStreamProcessors() para verificar los cambios de nombre, nivel y fila de letra muerta:

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'
    }
  }
]

Ejecuta sp.processor.sample() para verificar el cambio en el pipeline:

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
  }
}

Obtén más información

Gestionar procesadores de flujo

Volver

sp.processor.descartar

sp.processor.muestra