Robert Walters

> Following this tutorial requires private preview access to [Atlas Stream Processing](https://www.mongodb.com/products/platform/atlas-stream-processing). If you do not already have access, please [request it](https://cloud.mongodb.com/go?l=https%3A%2F%2Fcloud.mongodb.com%2Fv2%2F%23%2FstreamProcessing%2FrequestAccess%3FopenSurvey%3Dtrue) or learn more about Atlas Stream Processing in our [announcement blog](https://www.mongodb.com/blog/post/introducing-atlas-stream-processing-simplifying-path-reactive-responsive-even-driven-apps).

[Apache Kafka](https://kafka.apache.org/) is a massively popular streaming platform today. It is available in the open-source community and also as software (e.g., Confluent Platform) for self-managing. Plus, you can get a hosted Kafka (or Kafka-compatible) service from a number of providers, including AWS Managed Streaming for Apache Kafka (MSK), RedPanda Cloud, and Confluent Cloud, to name a few.

In this tutorial, we will configure network connectivity between MongoDB Atlas Stream Processing instances and a topic within the [Confluent Cloud.](https://www.confluent.io/confluent-cloud/) By the end of this tutorial, you will be able to process stream events from Confluent Cloud topics and emit the results back into a Confluent Cloud topic. 

Confluent Cloud dedicated clusters support connectivity through secure public internet endpoints with their Basic and Standard clusters. Private network connectivity options such as Private Link connections, VPC/VNet peering, and AWS Transit Gateway are available in the Enterprise and Dedicated cluster tiers. 

**Note:** At the time of this writing, Atlas Stream Processing is in preview and only supports internet-facing Basic and Standard Confluent Cloud clusters. This post will be updated to accommodate Enterprise and Dedicated clusters when support is provided for private networks.

The easiest way to get started with connectivity between Confluent Cloud and MongoDB Atlas is by using public internet endpoints. Public internet connectivity is the only option for Basic and Standard Confluent clusters. Rest assured that Confluent Cloud clusters with internet endpoints are protected by a proxy layer that prevents types of DoS, DDoS, SYN flooding, and other network-level attacks. We will also use authentication API keys with the SASL_SSL authentication method for secure credential exchange.

In this tutorial, we will set up and configure Confluent Cloud and MongoDB Atlas for network connectivity and then work through a simple example that uses a sample data generator to stream data between MongoDB Atlas and Confluent Cloud.

## Tutorial prerequisites

This is what you’ll need to follow along:

- An Atlas project (free or paid tier)
- An [Atlas database user](https://www.mongodb.com/docs/atlas/tutorial/create-mongodb-user-for-cluster/) with atlasAdmin permission 
    - For the purposes of this tutorial, we’ll have the user “tutorialuser.”
- MongoDB shell (Mongosh) version 2.0+
- Confluent Cloud cluster (any configuration)

## Configure Confluent Cloud

For this tutorial, you need a Confluent Cloud cluster created with a topic, “solardata,” and an API access key created. If you already have this, you may skip to Step 2.

To create a Confluent Cloud cluster, log into the Confluent Cloud portal, select or create an environment for your cluster, and then click the “Add Cluster” button. 

In this tutorial, we can use a **Basic** cluster type.

![Confluent Cloud create cluster][1]

Once your cluster is created, create an API key by clicking on the “API Keys” menu under the Cluster Overview on the left side of the page.

![Confluent Cloud API keys dialog][2]

Click on “Create Key” and provide a description for your key pair as shown below.

![Confluent Cloud Create key dialog][3]

Make a note of the API key and password before you download and continue. You will need these when creating the connection in Atlas Stream Processing. Note that Confluent OAuth and Confluent Single Sign-on are not supported as authentication methods in Atlas Stream Processing.

Next, create a topic by clicking on the “Topics” menu item and then the “Add topic” button. Accept the default settings and give the topic a name: “solardata.” We are now ready to configure MongoDB Atlas Stream Processing.

## Configure Atlas Stream Processing Connection Registry

In [MongoDB Atlas](http://cloud.mongodb.com), click on “Stream Processing” from the Services menu. Next, click on the “Create Instance” button. Provide a name, cloud provider, and region. Note: For a lower network cost, choose the cloud provider and region that matches your Confluent Cloud cluster. In this tutorial, we will use AWS us-east-1 for both Confluent Cloud and MongoDB Atlas.

![The Atlas Stream Processing Stream Processor Instance dashboard][4]

Once the Stream Processing Instance (SPI) is created, we can create our connection to the Confluent Cloud using the Connection Registry. Click on “Configure,” and then click on the “Connection Registry” tab as shown below.

![Configure Atlas Stream Processor Instance dashboard][5]

To create the connection to the Confluent Cloud, click on “Add Connection.”

Select, “Kafka” and enter “confluentcloud” for the connection name. Fill out the following details from the information in your Confluent Cloud cluster.

- Bootstrap server: Provided in Confluent Cloud under Cluster Settings/Endpoints
- Security Protocol: SASL_SSL
- SASL Mechanism: PLAIN
- Username: Paste in the API KEY
- Password: Paste in the API SECRET

An example of add connection dialog is shown below.

![Add connection registry connection dialog][6]

Click on “Add Connection” and your new connection to the Confluent Cloud will show up in the list.

![Connection Registry connection dashboard][7]

Next, create another connection by clicking on the “Add Connection” button. This time, we will select “Sample Stream” and “sample_stream_solar” in the drop-down as shown below.

![Add sample stream connection][8]

This will make a sample data generator called “sample_stream_solar” available in our SPI.

Next, let’s test the connectivity to Confluent and run our first Atlas Stream Processor with data from Confluent Cloud.

## Create the stream processor in Atlas

Note: To connect to the SPI, you will need a database user with Atlas Admin permissions or a member of the Project Owner role. If you do not already have this, [create it now](https://www.mongodb.com/docs/atlas/tutorial/create-mongodb-user-for-cluster/) before continuing this tutorial.

Connection information can be found by clicking on the “Connect” button on your SPI. The connect dialog is similar to the connect dialog when connecting to an Atlas cluster. To connect to the SPI, you will need to use the **mongosh** command line tool.

![Connect dialog for the stream processor instance][9]

To connect to the SPI, use the connection string provided in the connect dialog.

Once connected, you can enumerate the available connections using the `sp.listConnections()` command.

```
AtlasStreamProcessing> sp.listConnections()
[
    { name: 'sample_stream_solar', type: 'inmemory' },
    { name: 'confluentcloud', type: 'kafka' }
]
```

Now that we have confirmed both our ‘sample_stream_solar’ sample data and our ‘confluentcloud’ Kafka topic are available, let’s use the solar sample source to create a streaming query that calculates the average power output and writes this to a Kafka topic “solardata”.

```
s_solar={$source: { connectionName: "sample_stream_solar"}}

Twindow = { $tumblingWindow: { interval: { size: NumberInt(30), unit: "second" }, pipeline: [ { $group: { _id: "$device_id", max: { $max: "$obs.watts" }, avg: { $avg: "$obs.watts" } } }] } }

write_kafka={ $emit: { "connectionName": "confluentcloud", "topic" : "solardata"}}
```

Now that we have our pipeline variables defined, let’s use the `.process` to run this stream processor in the foreground.

```
AtlasStreamProcessing> sp.process([s_solar,Twindow,write_kafka])
```

To read the topic data, open another terminal window and connect to the SPI. Define a variable for the Kafka topic as shown below.

```
s_kafka={$source: { connectionName: "confluentcloud", "topic": "solardata"}}
```

Next, use the `.process()` command to read the data from the ‘solardata’ topic.

```
AtlasStreamProcessing> sp.process([s_solar])
```

After about 30 seconds, you will see data output from the ‘solardata’ topic.

## Wrapping up

In this tutorial, we used Atlas Stream Processing to create a stream processor with sample data and wrote the aggregation results to a Kafka topic in Confluent Cloud. We also streamed data from Confluent Cloud into Atlas Stream Processing and confirmed that the transformed data was written to the topic. This tutorial was done without any extra network configuration.

You might recall that by default, no network connections are allowed into Atlas. Users need to either open their cluster to the world via adding 0.0.0.0 or specify specific IP ranges. What is important to note is connections from Atlas Stream Processing originate within Atlas and connect out to the Confluent Cloud. Thus, there is no network access IP that needs to be opened or IP allowlisted. 

In Confluent Cloud, there is no concept of IP filtering or IP allowlisting. For this reason, there is nothing extra to perform on the Confluent Cloud side with respect to networking configuration. At the time of this writing, September 2023, private networking options available in Confluent Cloud such as PrivateLink are not supported in Atlas Stream Processing. This tutorial will be updated when these private networking options are supported.

> [Request access](https://cloud.mongodb.com/go?l=https%3A%2F%2Fcloud.mongodb.com%2Fv2%2F%23%2FstreamProcessing%2FrequestAccess%3FopenSurvey%3Dtrue&_ga=2.15333685.2035915229.1695608204-1192886547.1694190563) to participate in the private preview and check out more Atlas Stream Processing tutorials and content on the MongoDB [Developer Center](https://www.mongodb.com/developer/products/atlas/stream-processing/)!


  [1]: https://images.contentstack.io/v3/assets/blt39790b633ee0d5a7/bltcfb9c8a1f971ace1/652994177aecdf27ae595bf9/image24.png
  [2]: https://images.contentstack.io/v3/assets/blt39790b633ee0d5a7/blt63a22c62ae627895/652994381e33730b6478f0d1/image5.png
  [3]: https://images.contentstack.io/v3/assets/blt39790b633ee0d5a7/blte3f1138a6294748f/65299459382be57ed901d434/image21.png
  [4]: https://images.contentstack.io/v3/assets/blt39790b633ee0d5a7/blt3ccf2827c99f1c83/6529951a56a56b7388898ede/image19.png
  [5]: https://images.contentstack.io/v3/assets/blt39790b633ee0d5a7/bltaea830d5730e5f51/652995402e91e47b2b547e12/image20.png
  [6]: https://images.contentstack.io/v3/assets/blt39790b633ee0d5a7/blt9c425a65bb77f282/652995c0451768c2b6719c5f/image13.png
  [7]: https://images.contentstack.io/v3/assets/blt39790b633ee0d5a7/blt2748832416fdcf8e/652996cd24aaaa5cb2e56799/image15.png
  [8]: https://images.contentstack.io/v3/assets/blt39790b633ee0d5a7/blt9010c25a76edb010/652996f401c1899afe4a465b/image7.png
  [9]: https://images.contentstack.io/v3/assets/blt39790b633ee0d5a7/blt27b3762b12b6b871/652997508adde5d1c8f78a54/image3.png

Learn how to configure network connectivity between Confluent Cloud and MongoDB Atlas Stream Processing.

Using the Confluent Cloud with Atlas Stream Processing

> Following this tutorial requires private preview access to [Atlas Stream Processing](https://www.mongodb.com/products/platform/atlas-stream-processing). If you do not already have access, please request it [here](https://cloud.mongodb.com/go?l=https%3A%2F%2Fcloud.mongodb.com%2Fv2%2F%3Cproject%3E%23%2FstreamProcessing%2FrequestAccess%3FopenSurvey%3Dtrue&_ga=2.177827937.53133559.1696007795-1800933486.1688662045) or learn more about Atlas Stream Processing in our [announcement blog](https://www.mongodb.com/blog/post/introducing-atlas-stream-processing-simplifying-path-reactive-responsive-even-driven-apps).

In our [previous post](https://dev.to/mongodb/aggregate-streaming-data-within-windows-of-time-using-atlas-stream-processing-1lm1) on windowing, we introduced window operators available in Atlas Stream Processing. Window operators are one of the most commonly used operations to effectively process streaming data. [Atlas Stream Processing](https://www.mongodb.com/docs/atlas/atlas-sp/overview/) provides two window operators: [$tumblingWindow](https://www.mongodb.com/docs/atlas/atlas-sp/overview/#tumbling-windows) and [$hoppingWindow](https://www.mongodb.com/docs/atlas/atlas-sp/overview/#hopping-windows). In this tutorial, we will explore both of these operators using the sample solar data generator provided within Atlas Stream Processing.

## Getting started

Before we begin creating stream processors, make sure you have a [database user](https://www.mongodb.com/docs/atlas/security-add-mongodb-users/#built-in-roles) who has “atlasAdmin” access to the Atlas Project. Also, if you do not already have a Stream Processing Instance created with a connection to the sample_stream_solar data generator, please follow the instructions in [Get Started with Atlas Stream Processing: Creating Your First Stream Processor](https://www.mongodb.com/developer/products/atlas/create-first-stream-processor) and then continue on.

## View the solar stream sample data

For this tutorial, we will be using the [MongoDB shell](https://www.mongodb.com/docs/mongodb-shell/). 

First, confirm sample_stream_solar is added as a connection by issuing `sp.listConnections()`.

```
AtlasStreamProcessing> sp.listConnections()
{
    ok: 1,
    connections: [
    	{
            name: 'sample_stream_solar',
            type: 'inmemory',
            createdAt: ISODate("2023-08-26T18:42:48.357Z")
        } 
    ]
}
```

Next, let’s define a **$source** stage to describe where Atlas Stream Processing will read the stream data from.

```
var solarstream={ $source: { "connectionName": "sample_stream_solar" } }
```

Then, issue a **.process** command to view the contents of the stream on the console.

```
sp.process([solarstream])
```

You will see the stream of solar data printed on the console. A sample of this data is as follows:

```json
{
    device_id: 'device_2',
    group_id: 3,
    timestamp: '2023-08-27T13:51:53.375+00:00',
    max_watts: 250,
    event_type: 0,
    obs: {
        watts: 168,
        temp: 15
    },
    _ts: ISODate("2023-08-27T13:51:53.375Z"),
    _stream_meta: {
        sourceType: 'sampleData',
        timestamp: ISODate("2023-08-27T13:51:53.375Z")
    }
}
```

## Create a tumbling window query

A tumbling window is a fixed-size window that moves forward in time at regular intervals. In Atlas Stream Processing, you use the [$tumblingWindow](https://www.mongodb.com/docs/atlas/atlas-sp/overview/#tumbling-windows) operator. In this example, let’s use the operator to compute the average watts over one-minute intervals.

Refer back to the schema from the sample stream solar data. To create a tumbling window, let’s create a variable and define our tumbling window stage. 

```javascript
var Twindow= { 
    $tumblingWindow: { 
         interval: { size: NumberInt(1), unit: "minute" },
         pipeline: [ 
             { 
                 $group: { 	
                     _id: "$device_id", 
                    max: { $max: "$obs.watts" }, 
                    avg: { $avg: "$obs.watts" } 
                 }
             }
         ]
     } 
}
```

We are calculating the maximum value and average over the span of one-minute, non-overlapping intervals. Let’s use the `.process` command to run the streaming query in the foreground and view our results in the console.

```
sp.process([solarstream,Twindow])
```

Here is an example output of the statement:

```json
{
    _id: 'device_4',
    max: 236,
    avg: 95,
    _stream_meta: {
        sourceType: 'sampleData',
        windowStartTimestamp: ISODate("2023-08-27T13:59:00.000Z"),
        windowEndTimestamp: ISODate("2023-08-27T14:00:00.000Z")
    }
}
{
    _id: 'device_2',
    max: 211,
    avg: 117.25,
    _stream_meta: {
        sourceType: 'sampleData',
        windowStartTimestamp: ISODate("2023-08-27T13:59:00.000Z"),
        windowEndTimestamp: ISODate("2023-08-27T14:00:00.000Z")
    }
}
```

## Exploring the window operator pipeline

The pipeline that is used within a window function can include blocking stages and non-blocking stages.  

[Accumulator operators](https://www.mongodb.com/docs/manual/reference/operator/aggregation/#accumulators---group---bucket---bucketauto---setwindowfields-) such as `$avg`, `$count`, `$sort`, and `$limit` can be used within blocking stages. Meaningful data returned from these operators are obtained when run over a series of data versus a single data point. This is why they are considered blocking. 

Non-blocking stages do not require multiple data points to be meaningful, and they include operators such as `$addFields`, `$match`, `$project`, `$set`, `$unset`, and `$unwind`, to name a few. You can use non-blocking before, after, or within the blocking stages. To illustrate this, let’s create a query that shows the average, maximum, and delta (the difference between the maximum and average). We will use a non-blocking **$match** to show only the results from device_1, calculate the tumblingWindow showing maximum and average, and then include another non-blocking `$addFields`. 

```
var m= { '$match': { device_id: 'device_1' } }
```

```javascript
var Twindow= {
    '$tumblingWindow': {
        interval: { size: Int32(1), unit: 'minute' },
        pipeline: [
            {
                '$group': {
                    _id: '$device_id',
                    max: { '$max': '$obs.watts' },
                    avg: { '$avg': '$obs.watts' }
                }
            }
        ]
    }
}

var delta = { '$addFields': { delta: { '$subtract': ['$max', '$avg'] } } }
```

Now we can use the .process command to run the stream processor in the foreground and view our results in the console.

```
sp.process([solarstream,m,Twindow,delta])
```

The results of this query will be similar to the following:

```json
{
    _id: 'device_1',
    max: 238,
    avg: 75.3,
    _stream_meta: {
        sourceType: 'sampleData',
        windowStartTimestamp: ISODate("2023-08-27T19:11:00.000Z"),
        windowEndTimestamp: ISODate("2023-08-27T19:12:00.000Z")
    },
    delta: 162.7
}
{
    _id: 'device_1',
    max: 220,
    avg: 125.08333333333333,
    _stream_meta: {
        sourceType: 'sampleData',
        windowStartTimestamp: ISODate("2023-08-27T19:12:00.000Z"),
        windowEndTimestamp: ISODate("2023-08-27T19:13:00.000Z")
    },
    delta: 94.91666666666667
}
{
    _id: 'device_1',
    max: 238,
    avg: 119.91666666666667,
    _stream_meta: {
        sourceType: 'sampleData',
        windowStartTimestamp: ISODate("2023-08-27T19:13:00.000Z"),
        windowEndTimestamp: ISODate("2023-08-27T19:14:00.000Z")
    },
    delta: 118.08333333333333
}
```

Notice the time segments and how they align on the minute. 

![Time segments aligned on the minute][1]

Additionally, notice that the output includes the difference between the calculated values of maximum and average for each window.

## Create a hopping window

A hopping window, sometimes referred to as a sliding window, is a fixed-size window that moves forward in time at overlapping intervals. In Atlas Stream Processing, you use the `$hoppingWindow` operator. In this example, let’s use the operator to see the average.

```javascript
var Hwindow = {
    '$hoppingWindow': {
        interval: { size: 1, unit: 'minute' },
        hopSize: { size: 30, unit: 'second' },
        pipeline: [
            {
                '$group': {
                    _id: '$device_id',
                    max: { '$max': '$obs.watts' },
                    avg: { '$avg': '$obs.watts' }
                }
            }
        ]
    }
}
```

To help illustrate the start and end time segments, let's create a filter to only return device_1.

```
var m = { '$match': { device_id: 'device_1' } }
```

Now let’s issue the `.process` command to view the results in the console.

```
sp.process([solarstream,m,Hwindow])
```

An example result is as follows:

```json
{
    _id: 'device_1',
    max: 238,
    avg: 76.625,
    _stream_meta: {
        sourceType: 'sampleData',
        windowStartTimestamp: ISODate("2023-08-27T19:37:30.000Z"),
        windowEndTimestamp: ISODate("2023-08-27T19:38:30.000Z")
    }
}
{
    _id: 'device_1',
    max: 238,
    avg: 82.71428571428571,
    _stream_meta: {
        sourceType: 'sampleData',
        windowStartTimestamp: ISODate("2023-08-27T19:38:00.000Z"),
        windowEndTimestamp: ISODate("2023-08-27T19:39:00.000Z")
    }
}
{
    _id: 'device_1',
    max: 220,
    avg: 105.54545454545455,
    _stream_meta: {
        sourceType: 'sampleData',
        windowStartTimestamp: ISODate("2023-08-27T19:38:30.000Z"),
        windowEndTimestamp: ISODate("2023-08-27T19:39:30.000Z")
    }
}
```

Notice the time segments.

![Overlapping time segments][2]

The time segments are overlapping by 30 seconds as was defined by the hopSize option. Hopping windows are useful to capture short-term patterns in data. 

## Summary

By continuously processing data within time windows, you can generate real-time insights and metrics, which can be crucial for applications like monitoring, fraud detection, and operational analytics. Atlas Stream Processing provides both tumbling and hopping window operators. Together these operators enable you to perform various aggregation operations such as sum, average, min, and max over a specific window of data. In this tutorial, you learned how to use both of these operators with solar sample data.  

> [Request access](https://cloud.mongodb.com/go?l=https%3A%2F%2Fcloud.mongodb.com%2Fv2%2F%3Cproject%3E%23%2FstreamProcessing%2FrequestAccess%3FopenSurvey%3Dtrue&_ga=2.177827937.53133559.1696007795-1800933486.1688662045) to participate in the private preview and check out more Atlas Stream Processing tutorials and content on the MongoDB [Developer Center](https://www.mongodb.com/developer/products/atlas/stream-processing/)!


  [1]: https://images.contentstack.io/v3/assets/blt39790b633ee0d5a7/blt73ff54f0367cad3b/650da3ef69060a5678fc1242/image1.jpg
  [2]: https://images.contentstack.io/v3/assets/blt39790b633ee0d5a7/blt833bc1a824472d14/650da41aa5f15dea3afc5b55/image3.jpg

Learn how to use the various window operators such as tumbling window and hopping window with MongoDB Atlas Stream Processing.

Exploring Window Operators in Atlas Stream Processing

>Following this tutorial requires private preview access to [Atlas Stream Processing](https://www.mongodb.com/products/platform/atlas-stream-processing). If you do not already have access, please request it [here](https://cloud.mongodb.com/go?l=https%3A%2F%2Fcloud.mongodb.com%2Fv2%2F%3Cproject%3E%23%2FstreamProcessing%2FrequestAccess%3FopenSurvey%3Dtrue) or learn more about Atlas Stream Processing in our [announcement blog](https://www.mongodb.com/blog/post/introducing-atlas-stream-processing-simplifying-path-reactive-responsive-even-driven-apps).

If you're not already familiar, Atlas Stream Processing enables processing high-velocity streams of complex data using the same [data model](https://www.mongodb.com/docs/manual/core/data-modeling-introduction/#document-structure) and [Query API](https://www.mongodb.com/docs/manual/query-api/) that's used in MongoDB Atlas databases. Streaming data is increasingly critical to building responsive, [event-driven experiences](https://www.mongodb.com/collateral/building-event-driven-applications-with-mongodb) for your customers. Stream processing is a fundamental building block powering these applications, by helping to tame the firehouse of data coming from many sources, by finding important events in a stream, or by combining data in motion with data in rest. 

In this tutorial, we will create a stream processor that uses sample data included in [Atlas Stream Processing](https://www.mongodb.com/products/platform/atlas-stream-processing). By the end of the tutorial, you will have an operational [Stream Processing Instance (SPI)](https://www.mongodb.com/docs/atlas/atlas-sp/overview/#mongodb-expression-exp.Stream-Processing-Instance) configured with a stream processor. This environment can be used for further experimentation and Atlas Stream Processing tutorials in the future. 


### Tutorial Prerequisites 
This is what you'll need to follow along:
* An Atlas user with atlasAdmin permission. For the purposes of this tutorial, we'll have the user "tutorialuser". 
* [MongoDB shell](https://www.mongodb.com/docs/mongodb-shell/) (Mongosh) version 2.0+

## Create the Stream Processing Instance 

Let's first create a [Stream Processing Instance (SPI)](https://www.mongodb.com/docs/atlas/atlas-sp/overview/#mongodb-expression-exp.Stream-Processing-Instance). Think of an SPI as a logical grouping of one or more stream processors. When created, the SPI has a connection string similar to a typical [MongoDB Atlas cluster](https://www.mongodb.com/docs/atlas/create-database-deployment/#std-label-ref-deployment-types). 

Under the Services tab in the Atlas Project click, "Stream Processing". Then click the "Create Instance" button. 

![Stream Processing Instance button](https://images.contentstack.io/v3/assets/blt39790b633ee0d5a7/blt7d2f80b81220c397/650b06e6c5537c75a84d077c/streamprocess1.png)

This will launch the Create Instance dialog. 

![Create Stream Processing Instance dialog in the Atlas UI](https://images.contentstack.io/v3/assets/blt39790b633ee0d5a7/blta89ce227cfc6e87e/650b076ad8e1302aace5d259/streamprocess2.png)

Enter your desired cloud provider and region, and then click "Create". You will receive a confirmation dialog upon successful creation. 

## Configure the connection registry 

The connection registry stores connection information to the external data sources you wish to use within a stream processor. In this example, we will use a sample data generator that is available without any extra configuration, but typically you would connect to either Kafka or an Atlas database as a source. 

To manage the connection registry, click on "Configure" to navigate to the configuration screen. 

![The Stream Processing page in Atlas will be empty until the first stream processor is created](https://images.contentstack.io/v3/assets/blt39790b633ee0d5a7/blt265796325f27e371/650b0941c2a602557b759854/streamprocess3.png)

Once on the configuration screen, click on the "Connection Registry" tab. 
![Configuration screen with Atlas Stream Processing page in Atlas](https://images.contentstack.io/v3/assets/blt39790b633ee0d5a7/blt6dbfb41cc0b10d60/650b0a41b556bd11c3d3ba58/streamprocess4.png)

Next, click on the "Add Connection" button. This will launch the Add Connection dialog. 
![Add Connection screen within the Stream Processing page in Atlas](https://images.contentstack.io/v3/assets/blt39790b633ee0d5a7/blt43d8695536496c45/650b0aea75fde0cacc31c1aa/streamprocess5.png)

From here, you can add connections to Kafka, other Atlas clusters within the project, or a sample stream. In this tutorial, we will use the Sample Stream connection. Click on "Sample Stream" and select "sample_stream_solar" from the list of available sample streams. Then, click "Add Connection". 

The new "sample_stream_solar" will show up in the list of connections. 
![A list of your connection names and types is visible on the Stream Processing page in Atlas](https://images.contentstack.io/v3/assets/blt39790b633ee0d5a7/blte9c6f1eb2cad0b17/650b0c35b25642fa94dd183d/streamprocess6.png)

## Connect to the Stream Processing Instance (SPI)

Now that we have both created the SPI and configured the connection in the connection registry, we can create a stream processor. First, we need to connect to the SPI that we created previously. This can be done using the [MongoDB Shell (mongosh)](https://www.mongodb.com/docs/mongodb-shell/). 

To obtain the connection string to the SPI, return to the main Stream Processing page by clicking on the "Stream Processing" menu under the Services tab. 

![Navigate to Stream Processing within the Services pane in Atlas](https://images.contentstack.io/v3/assets/blt39790b633ee0d5a7/blt500738459fb7e394/650b0e17f0b16b92fd1d9712/streamprocess7.png)

Next, locate the "Tutorial" SPI we just created and click on the "Connect" button. This will present a connection dialog similar to what is found when connecting to MongoDB Atlas clusters. 

![In the Atlas UI, each Stream Processing Instance is visible for users to connect and configure](https://images.contentstack.io/v3/assets/blt39790b633ee0d5a7/blt4c74b3505ea4a65d/650b121fcd40632522735bb6/streamprocess9.png)

For connecting, we'll need to add a connection IP address and create a database user, if we haven't already. 

![The connection dialog for a Stream Processing Instance steps through setting up connection security, choosing a connection method, and finally, connecting to the SPI](https://images.contentstack.io/v3/assets/blt39790b633ee0d5a7/blta768dd0c0156b9e9/650b12a9a39cd6ae3eacf741/streamprocess10.png)

Then we'll choose our connection method. If you do not already have mongosh installed, install it using the instructions provided in the dialog. 

![Stream processor connection screen in Atlas](https://images.contentstack.io/v3/assets/blt39790b633ee0d5a7/bltb2214d242eb30757/650b130c7e0302670fbbede2/streamprocess11.png)

Once mongosh is installed, copy the connection string from the "I have the MongoDB Shell installed" view and run it in your terminal. 

```
Command Terminal > mongosh <<CONNECTION STRING>> --tls --authenticationDatabase admin --username tutorialuser

Enter password: *******************

Current Mongosh Log ID:	64e9e3bf025581952de31587
Connecting to:		mongodb://*****
Using MongoDB:		6.2.0
Using Mongosh:		2.0.0

For mongosh info see: https://docs.mongodb.com/mongodb-shell/

AtlasStreamProcessing>

```
To confirm your sample_stream_solar is added as a connection, issue `sp.listConnections()`. Our connection to sample_stream_solar is shown as expected.

```
AtlasStreamProcessing> sp.listConnections()
{
  ok: 1,
  connections: [
    {
      name: 'sample_stream_solar',
      type: 'inmemory',
      createdAt: ISODate("2023-08-26T18:42:48.357Z")
    } 
 ]
}
```

## Create a stream processor
If you are reading through this post as a prerequisite to another tutorial, you can return to that tutorial now to continue.

In this section, we will wrap up by creating a simple stream processor to process the sample_stream_solar source that we have used throughout this tutorial. This sample_stream_solar source represents the observed energy production of different devices (unique solar panels). Stream processing could be helpful in measuring characteristics such as panel efficiency or when replacement is required for a device that is no longer producing energy at all.

First, let's define a [$source](https://www.mongodb.com/docs/atlas/atlas-sp/stream-aggregation/#mongodb-pipeline-pipe.-source) stage to describe where Atlas Stream Processing will read the stream data from. 

```
var solarstream={$source:{"connectionName": "sample_stream_solar"}}
```
Now we will issue [.process](https://www.mongodb.com/docs/atlas/atlas-sp/tutorial/#verify-that-your-streaming-data-source-emits-messages.) to view the contents of the stream in the console. 
`sp.process([solarstream])`

.process lets us sample our source data and quickly test the stages of a stream processor to ensure that it is set up as intended. A sample of this data is as follows:

```
{
  device_id: 'device_2',
  group_id: 3,
  timestamp: '2023-08-27T13:51:53.375+00:00',
  max_watts: 250,
  event_type: 0,
  obs: {
    watts: 168,
    temp: 15
  },
  _ts: ISODate("2023-08-27T13:51:53.375Z"),
  _stream_meta: {
    sourceType: 'sampleData',
    timestamp: ISODate("2023-08-27T13:51:53.375Z")
  }
}
```
## Wrapping up

In this tutorial, we started by introducing Atlas Stream Processing and why stream processing is a building block for powering modern applications. We then walked through the basics of creating a stream processor – we created a Stream Processing Instance, configured a source in our connection registry using sample solar data (included in Atlas Stream Processing), connected to a Stream Processing Instance, and finally tested our first stream processor using .process. You are now ready to explore Atlas Stream Processing and create your own stream processors, adding advanced functionality like windowing and validation.

If you enjoyed this tutorial and would like to learn more check out the [MongoDB Atlas Stream Processing announcement](https://www.mongodb.com/blog/post/introducing-atlas-stream-processing-simplifying-path-reactive-responsive-even-driven-apps) blog post. For more on stream processors in Atlas Stream Processing, visit [our documentation](https://www.mongodb.com/docs/atlas/atlas-sp/manage-stream-processor/#manage-stream-processors). 

>[Request access](https://cloud.mongodb.com/go?l=https%3A%2F%2Fcloud.mongodb.com%2Fv2%2F%3Cproject%3E%23%2FstreamProcessing%2FrequestAccess%3FopenSurvey%3Dtrue) to participate in the private preview and check out more Atlas Stream Processing tutorials and content on the MongoDB [Developer Center](https://www.mongodb.com/developer/products/atlas/stream-processing/)!

Learn how to create a stream processor end-to-end using MongoDB Atlas Stream Processing.

Get Started with Atlas Stream Processing: Creating Your First Stream Processor

Atlas Stream Processing, continuously process complex streams of event data.

Stream Processing

All Stream Processing Content

Using the Confluent Cloud with Atlas Stream Processing

Exploring Window Operators in Atlas Stream Processing

Get Started with Atlas Stream Processing: Creating Your First Stream Processor

Related Topics