Manage Stream Processors

Prerequisites

To create and manage a stream processor, you must have:

• A stream processing instance

• mongosh version 2.0 or higher

• A database user with the atlasAdmin role to create and run stream processors

• An Atlas cluster

Considerations

Many stream processor commands require you to specify the name of the relevant stream processor in the method invocation. The syntax described in the following sections assumes strictly alphanumeric names. If your stream processor's name includes non-alphanumeric characters such as hyphens (-) or full stops (.), you must enclose the name in square brackets ([]) and double quotes ("") in the method invocation, as in sp.["special-name-stream"].stats().

Create a Stream Processor Interactively

You can create a stream processor interactively with the sp.process() method. 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 for either 10 minutes or until 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>)

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)

Create a Stream Processor

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)

Start a Stream Processor

To start an existing stream processor with mongosh, use the sp.<streamprocessor>.start() method. <streamprocessor> must be the name of a stream processor defined for the current stream processing instance.

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

sp.proc01.start()

This method returns:

• true if the stream processor exists and isn't currently running.

• false if you try to start a stream processor that doesn't exist, or exists and is currently running.

Stop a Stream Processor

To stop an existing stream processor with mongosh, use the sp.<streamprocessor>.stop() method. <streamprocessor> must be the name of a currently running stream processor defined for the current stream processing instance.

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

sp.proc01.stop()

This method returns:

• true if the stream processor exists and is currently running.

• false if the stream processor doesn't exist, or if the stream processor isn't currently running.

Drop a Stream Processor

To delete an existing stream processor with mongosh, use the sp.<streamprocessor>.drop() method. <streamprocessor> must be the name of a stream processor defined for the current stream processing instance.

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.

List Available Stream Processors

To list all available stream processors on the current stream processing instance with mongosh, use the sp.listStreamProcessors() method. It 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
}

Sample from a Stream Processor

To return an array of sampled results from an existing stream processor to STDOUT with mongosh, use the sp.<streamprocessor>.sample() method. <streamprocessor> must be the name of a currently running stream processor defined for the current stream processing instance. For example, the following command samples from a stream processor named proc01.

sp.proc01.sample()

This command runs continuously until you cancel it by using CTRL-C, or until the returned samples cumulatively reach 40 MB in size. The stream processor reports invalid documents in the sample in a _dlqMessage document of the following form:

 {
   _dlqMessage: {
     _stream_meta: {
       source: {
         type: "<type>"
       }
     },
     errInfo: {
       reason: "<reasonForError>"
     },
     doc: {
       _id: ObjectId('<group-id>'),
       ...
     },
     processorName: '<procName>'
   }
}

You can use these messages to diagnose data hygiene issues without defining a dead letter queue collection.

View Statistics of a Stream Processor

To return a document summarizing the current status of an existing stream processor with mongosh, use the sp.<streamprocessor>.stats() method. streamprocessor must be the name of a currently running stream processor defined for the current stream processing instance. It has the following syntax:

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

Where options is an optional document with the following fields:

The output document has the following fields:

For example, the following shows the status of a stream processor named proc01 on a stream processing instance 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"),
    ok: 1
  },
}