Get Started with Atlas Stream Processing

To create a stream processor in the Atlas UI, go to the Stream Processing page for your Atlas project and click Configure in the pane for your stream processing instance. Then choose between using the visual builder or the JSON editor to configure a stream processor named solarDemo:

Run the following commands in mongosh to create a persistent stream processor named solarDemo:

1. Connect to your stream processing instance.

   Use the connection string associated with your stream processing instance to connect using mongosh.

   1. In the pane for your stream processing instance, click Connect.

   2. In the Connect to your instance dialog, select the Shell tab.

   3. Copy the connection string displayed in the dialog. It has the following format, where <atlas-stream-processing-url> is the URL of your stream processing instance and <username> is the username of a database user with the atlasAdmin role:

      mongosh "mongodb://<atlas-stream-processing-url>/"
      --tls --authenticationDatabase admin --username <username>
      --password <password>

   4. Paste the connection string into your terminal and replace the <password> placeholder with the credentials for the user.

      Press Enter to run it and connect to your stream processing instance.

2. Configure a $source stage.

   Define a variable for a $source stage that ingests data from the sample_stream_solar source.

   let s = {
     $source: {
       connectionName: "sample_stream_solar"
     }
   }

3. Configure a $group stage.

   Define a variable for a $group stage that derives the maximum temperature and the average, maximum, and minimum wattages of each solar device according to its group_id.

   let g = {
     $group: {
       _id: "$group_id",
       max_temp: {
         $max: "$obs.temp"
       },
       avg_watts: {
          $avg: "$obs.watts"
       },
       max_watts: {
         $max: "$obs.watts"
       },
       min_watts: {
         $min: "$obs.watts"
       }
     }
   }

4. Configure a $tumblingWindow stage.

   In order to perform accumulations such as $group on streaming data, Atlas Stream Processing uses windows to bound the data set. Define a variable for a $tumblingWindow stage that separates the stream into consecutive 10-second intervals.

   This means, for example, that when the $group stage computes a value for max_watts, it extracts the maximum value from the obs.watts values for all documents with a given group_id ingested in the previous 10 seconds.

    let t = {
      $tumblingWindow: {
        interval: {
          size: NumberInt(10),
          unit: "second"
        },
        pipeline: [g]
     }
   }

5. Configure a $merge stage.

   Define a variable for a $merge stage that writes the processed streaming data to a collection named solarColl in the solarDb database of your connected Atlas cluster.

   let m = {
     $merge: {
       into: {
         connectionName: "mongodb1",
         db: "solarDb",
         coll: "solarColl"
       }
     }
   }

6. Create the stream processor.

   Use the sp.createStreamProcessor() method to assign a name to your new stream processor and declare its aggregation pipeline. The $group stage belongs to the nested pipeline of the $tumblingWindow, and you must not include it in the processor pipeline definition.

   sp.createStreamProcessor("solarDemo", [s, t, m])

   This creates a stream processor named solarDemo that applies the previously defined query and writes the processed data to the solarColl collection of the solarDb database on the cluster you connected to. It returns various measurements derived from 10-second intervals of observations from your solar devices.

   To learn more about how Atlas Stream Processing writes to at-rest databases, see $merge (Stream Processing).

In the list of stream processors for your stream processing instance, click the Start icon for your stream processor.

Use the sp.processor.start() method in mongosh:

sp.solarDemo.start()

To verify that the stream processor is writing data to your Atlas cluster:

1. In Atlas, go to the Clusters page for your project.

   1. If it's not already displayed, select the organization that contains your desired project from the Organizations menu in the navigation bar.

   2. If it's not already displayed, select your desired project from the Projects menu in the navigation bar.

   3. In the sidebar, click Clusters under the Database heading.

      The Clusters page displays.

2. In Atlas, go to the Data Explorer page for your project.

   1. If it's not already displayed, select the organization that contains your project from the Organizations menu in the navigation bar.

   2. If it's not already displayed, select your project from the Projects menu in the navigation bar.

   3. In the sidebar, click Data Explorer under the Database heading.

      The Data Explorer displays.

   Note

   You can also go to the Clusters page, and click Data Explorer under the Shortcuts heading.

3. View the MySolar collection.

To verify that the processor is active, use the sp.processor.stats() method in mongosh:

sp.solarDemo.stats()

This method reports operational statistics of the solarDemo stream processor.

You can also use the sp.processor.sample() method in mongosh to return a sampling of processed documents in the terminal.

sp.solarDemo.sample()

{
  _id: 10,
  max_temp: 16,
  avg_watts: 232,
  max_watts: 414,
  min_watts: 73
}

In the list of stream processors for your stream processing instance, 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.

Use the sp.processor.drop() method in mongosh to drop solarDemo:

sp.solarDemo.drop()

To confirm that you have dropped solarDemo, use the sp.listStreamProcessors() method to list all your available stream processors:

sp.listStreamProcessors()