Develop Stream Processors

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:

{
  "$source": {
    "connectionName": "sample_stream_solar"
  }
}

{
  "$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 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 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. These fields can be:

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 is in a startable state (STOPPED, CREATED, or FAILED). Otherwise, 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 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 in the STARTED state. Otherwise, mongosh returns an error.

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

Delete One Stream Processor

To delete 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 Delete () icon for your stream processor. In the confirmation dialog that appears, type the name of the stream processor (solarDemo) to confirm that you want to delete it, and then click Delete.

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 { ok: 1 } on success and an error 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 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, 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
  lastModified: ISODate("2023-03-20T20:15:54.601Z"),
5
  state: "STARTED",
6
  errorMsg: '',
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
  lastModified: ISODate("2023-03-21T20:17:33.601Z"),
32
  state: "STOPPED",
33
  errorMsg: '',
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": "STARTED"})

1
{
2
  id: '0135',
3
  name: "proc01",
4
  lastModified: ISODate("2023-03-20T20:15:54.601Z"),
5
  state: "STARTED",
6
  errorMsg: '',
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 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 with the following fields:

For a description of all output fields, see Stream Processor Statistics Fields.

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"),
    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
    }
  ],
  ...
}