Develop and Manage Stream Processors

You cannot create a stream processor interactively using the Atlas Administration API. To learn how to create a stream processor interactively using mongosh, select MongoDB Shell from the Interface dropdown.

You cannot create a stream processor interactively using the Atlas UI. To learn how to create a stream processor interactively using mongosh, select MongoDB Shell from the Interface dropdown.

You cannot create a stream processor interactively using the Atlas CLI. To learn how to create a stream processor interactively using mongosh, select MongoDB Shell from the Interface dropdown.

You can create a stream processor interactively with the sp.process() method in mongosh. Stream processors that you create interactively exhibit the following behavior:

• Write output and dead letter queue documents to the shell

• Begin running immediately upon creation

• Run until one of the following occurs: They return the number of documents set by the options.limit field, 10 minutes elapse, or the user stops them

• Don't persist after stopping

Stream processors that you create interactively are intended for prototyping. To create a persistent stream processor, see Create a Stream Processor.

sp.process() has the following syntax:

sp.process(<pipeline>, <options>)

To create a stream processor interactively:

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

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

sp.process(pipeline)

The Atlas Administration API provides an endpoint for creating a stream processor.

Create One Stream Processor

To create a stream processor in the Atlas UI, go to the Stream Processing page for your Atlas project and click Manage in the pane for your stream processing workspace.

You can choose between using the Visual Builder or the JSON editor to configure your stream processor:

{
  "$match": {
    "obs.watts": { "$gt": 300 }
  }
}

{
  "$merge": {
    "into": {
      "connectionName": "demoConnection",
      "db": "demoDb",
      "coll": "demoColl"
    }
  }
}

{
  "name": "solarDemo",
  "pipeline": [
    {
      "$source": {
        "connectionName": "sample_stream_solar"
      }
    },
    {
      "$tumblingWindow": {
        "interval": {
          "size": 10,
          "unit": "second"
        },
        "pipeline": [
          {
            "$group": {
              "_id": "$group_id",
              "max_watts": { "$max": "$obs.watts" },
              "min_watts": { "$min": "$obs.watts" }
            }
          }
        ]
      }
    },
    {
      "$merge": {
        "into": {
          "connectionName": "mongodb1",
          "db": "solarDb",
          "coll": "solarColl"
        }
      }
    }
  ]
}

To create one stream processor within the stream processing workspace you specify using the Atlas CLI, run the following command:

atlas api streams createStreamProcessor [options]

To learn more about the command syntax and parameters, see the Atlas CLI documentation for atlas api streams createStreamProcessor.

To create a new stream processor with mongosh, use the sp.createStreamProcessor() method. It has the following syntax:

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

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

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

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

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

The Atlas Administration API provides the Start One Stream Processor and Start One Stream Processor with Options endpoints for starting a stream processor.

• To start a stream processor without options, use Start One Stream Processor

• To start a stream processor and modify properties of the $source stage, use Start One Stream Processor with Options This endpoint supports modifying either the startAfter or startAtOperationTime properties.

To start a stream processor in the Atlas UI, go to the Stream Processing page for your Atlas project and click Manage in the pane for your stream processing workspace to view the list of stream processors defined for it.

Then, click the Start icon for your stream processor.

To start one stream processor within the stream processing workspace you specify using the Atlas CLI, run the following command:

atlas api streams startStreamProcessor [options]

To learn more about the command syntax and parameters, see the Atlas CLI documentation for atlas api streams startStreamProcessor.

To start a stream processor with mongosh, use the sp.processor.start() method. It has the following syntax:

sp.processor.start(<options>)

Where <options> is a generic, optional document whose fields are passed to the underlying start command. To learn more about the command options, see sp.createStreamProcessor(). For example, to start a stream processor named proc01, run the following command:

sp.proc01.start()

{ "ok" : 1 }

This method returns { "ok": 1 } if the stream processor exists and isn't currently running. If you invoke sp.processor.start() for a stream processor that is not STOPPED, mongosh returns an error.

The Atlas Administration API provides an endpoint for stopping a stream processor.

Stop One Stream Processor

To stop a stream processor in the Atlas UI, go to the Stream Processing page for your Atlas project and click Manage in the pane for your stream processing workspace to view the list of stream processors defined for it.

Then, click the Stop icon for your stream processor.

To stop one stream processor within the stream processing workspace you specify using the Atlas CLI, run the following command:

atlas api streams stopStreamProcessor [options]

To learn more about the command syntax and parameters, see the Atlas CLI documentation for atlas api streams stopStreamProcessor.

To stop an existing stream processor with mongosh, use the sp.processor.stop() method. It has the following syntax:

sp.processor.stop(<options>)

Where <options> is a generic, optional document whose fields are passed to the underlying stop command.

For example, to stop a stream processor named proc01, run the following command:

sp.proc01.stop()

{ "ok" : 1 }

This method returns { "ok": 1 } if the stream processor exists and is currently running. If you invoke sp.processor.stop() for a stream processor that is not running, mongosh returns an error.

The Atlas Administration API provides an endpoint for modifying a stream processor.

Modify One Stream Processor

To modify one existing stream processor within the stream processing workspace you specify using the Atlas CLI, run the following command:

atlas api streams updateStreamProcessor [options]

To learn more about the command syntax and parameters, see the Atlas CLI documentation for atlas api streams updateStreamProcessor.

Requires mongosh v2.3.4+.

Use the sp.<streamprocessor>.modify() command to modify an existing stream processor. <streamprocessor> must be the name of a stopped stream processor defined for the current stream processing workspace.

For example, to modify a stream processor named proc01, run the following command:

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

The failover option controls whether the processor runs in the failover regions configured for your stream processing workspace:

• Set failover to true to read the processor definition from the default region and create that processor in every failover region that doesn't already have it.

• Set failover to false to disable failover for a processor that has failover regions configured and revert it to running in a single region.

 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();

To initiate failover of an individual stream processor, use the Start One Stream Processor With Options endpoint. In the failover field of the request body, provide the name of the cloud region in which the target failover processor lives. Initializing the failover processor will also deactivate the primary processor.

To start one stream processor with optional parameters set using the Atlas CLI, run the following command:

atlas api streams startStreamProcessorWith [options]

To learn more about the command syntax and parameters, see the Atlas CLI documentation for atlas api streams startStreamProcessorWith.

To initiate failover for an individual stream processor with mongosh, use the sp.processor.start() method and pass the failover option. Starting the failover processor also stops the processor in the currently active region.

The failover option has the following syntax:

sp.<streamprocessor>.start({
  failover: {
    region: "<region>",
    mode: "<mode>",
    dryRun: <boolean>
  }
})

GRACEFUL mode takes a checkpoint and confirms that the active processor has stopped before it initiates failover. FORCED mode also attempts to stop the active processor, but initiates failover even if the processor doesn't stop. By default, a FORCED failover starts the processor in the target region using the most recent checkpoint.

To start the target processor without recovering from a checkpoint for a FORCED failover, set clearCheckpoints to true:

sp.<streamprocessor>.start({
  failover: {
    region: "<region>",
    mode: "FORCED"
  },
  clearCheckpoints: true
})

When you clear checkpoints, the target processor starts from the current position in the stream instead of recovering its most recent checkpoint. Clearing checkpoints might cause duplicate or missing records in the output.

For example, to initiate a graceful failover of a stream processor named proc01 to the us-west-2 region, run the following command:

sp.proc01.start({
  failover: {
    region: "us-west-2",
    mode: "GRACEFUL",
    dryRun: false
  }
})

To stop a stream processor, use the sp.processor.stop() method. You don't have to stop the active processor before you initiate failover.

The Atlas Administration API provides an endpoint for deleting a stream processor.

Delete One Stream Processor

To delete one stream processor within the stream processing workspace you specify using the Atlas CLI, run the following command:

atlas api streams deleteStreamProcessor [options]

To learn more about the command syntax and parameters, see the Atlas CLI documentation for atlas api streams deleteStreamProcessor.

To delete an existing stream processor with mongosh, use the sp.processor.drop() method. It has the following syntax:

sp.processor.drop(<options>)

Where <options> is a generic, optional document whose fields are passed to the underlying drop command.

For example, to drop a stream processor named proc01, run the following command:

sp.proc01.drop()

This method returns:

• true if the stream processor exists.

• false if the stream processor doesn't exist.

When you drop a stream processor, all resources that Atlas Stream Processing provisioned for it are destroyed, along with all saved state.

The Atlas Administration API provides an endpoint for listing all available stream processors.

List Stream Processors

To view the list of stream processors defined for your stream processing workspace in the Atlas UI, go to the Stream Processing page for your Atlas project and click Manage in the pane for your stream processing workspace.

The list of stream processors and their statuses displays.

To return all stream processors within the stream processing workspace you specify using the Atlas CLI, run the following command:

atlas api streams getStreamProcessors [options]

To learn more about the command syntax and parameters, see the Atlas CLI documentation for atlas api streams getStreamProcessors.

To list all available stream processors on the current stream processing workspace with mongosh, use the sp.listStreamProcessors() method. This returns a list of documents containing the name, start time, current state, and pipeline associated with each stream processor. It has the following syntax:

sp.listStreamProcessors(<filter>)

<filter> is a document specifying which field(s) to filter the list by.

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
}

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
}

The List Stream Workspace Details endpoint lists the details of a stream processing workspace, including the defaultTierSize and maxTierSize.

To view the defaultTierSize and maxTierSize of a stream processing workspace in the Atlas UI, go to the Stream Processing page for your Atlas project.

The defaultTierSize and maxTierSize are listed at the bottom of the stream processing workspace card.

To return details of one stream processing workspace you specify using the Atlas CLI, run the following command:

atlas api streams getStreamWorkspace [options]

To learn more about the command syntax and parameters, see the Atlas CLI documentation for atlas api streams getStreamWorkspace.

To view the defaultTierSize and maxTierSize of a stream processing workspace using mongosh, use the sp.listWorkspaceDefaults() method.

sp.listWorkspaceDefaults()

This method returns:

• defaultTierSize

• maxTierSize

The Atlas Administration API provides an endpoint for viewing the statistics of a stream processor.

Get One Stream Processor

To view monitoring for your stream processor, go to the Stream Processing page for your Atlas project and open the Monitoring tab. Then, select your stream processor from the Stream processor drop-down list at the top left of the page.

To return a document summarizing the current status of an existing stream processor with mongosh, use the sp.processor.stats() method. It has the following syntax:

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

Where options is an optional document. To learn more about the command options and output fields, see sp.processor.stats().

For a description of every field in the output document, see Stream Processor Metrics.

For example, the following shows the status of a stream processor named proc01 on a stream processing workspace named inst01 with item sizes displayed in 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"),
    addedParallelism: 7,
    ok: 1
  },
}

The example shows the overall statistics for the stream processor.

To see how individual operators behave or how much traffic each target handles, call sp.<streamprocessor>.stats() with the verbose option and inspect stats.operatorStats and, for some operators, stats.operatorStats.targetStats.

For example, for a source operator, stats.operatorStats.targetStats collects the inputMessageCount and inputMessageSize fields for every unique db/coll or every unique topic:

{
  "name" : "KafkaConsumerOperator",
  "inputMessageCount" : NumberLong(100),
  "inputMessageSize" : 100352,
  "targetStats" : [
    {
    "topic" : "outputTopic1",
    "inputMessageCount" : NumberLong(100),
    "inputMessageSize" : 100352
    }
  ],
  ...
}

And for a sink operator, stats.operatorStats.targetStats collects the outputMessageCount and outputMessageSize fields for every unique db/coll or every unique topic:

{
  "name" : "MergeOperator",
  "inputMessageCount" : NumberLong(10),
  "inputMessageSize" : 1744,
  "targetStats" : [
    {
      "db" : "cust1",
      "coll" : "outColl1",
      "outputMessageCount" : NumberLong(3),
      "outputMessageSize" : 1748
    },
    {
      "db" : "cust2",
      "coll" : "outColl2",
      "outputMessageCount" : NumberLong(4),
      "outputMessageSize" : 2241
    }
  ],
  ...
}