Robert Walters

Senior Product Manager

The MongoDB Connector for Apache Kafka documentation now includes new [tutorials](https://www.mongodb.com/docs/kafka-connector/current/tutorials/)! These tutorials introduce you to key concepts behind the connector and by the end, you’ll have an understanding of how to move data between MongoDB and Apache Kafka. The tutorials are as follows:

* [Explore Change Streams](https://www.mongodb.com/docs/kafka-connector/current/tutorials/explore-change-streams/#std-label-kafka-tutorial-explore-change-streams)

Change streams is a MongoDB server feature that provides change data capture (CDC) capabilities for MongoDB collections. The source connector relies on change streams to move data from MongoDB to a Kafka topic. In this tutorial, you will explore creating a change stream and reading change stream events all through a Python application.

* [Getting Started with the MongoDB Kafka Source Connector](https://www.mongodb.com/docs/kafka-connector/current/tutorials/source-connector/#std-label-kafka-tutorial-source-connector)

In this tutorial, you will configure a source connector to read data from a MongoDB collection into an Apache Kafka topic and examine the content of the event messages.

* [Getting Started with the MongoDB Kafka Sink Connector](https://www.mongodb.com/docs/kafka-connector/current/tutorials/sink-connector/#std-label-kafka-tutorial-sink-connector)

In this tutorial, you will configure a sink connector to copy data from a Kafka topic into a MongoDB cluster and then write a Python application to write data into the topic.

* [Replicate Data with a Change Data Capture Handler](https://www.mongodb.com/docs/kafka-connector/current/tutorials/replicate-with-cdc/)

Configure both a MongoDB source and sink connector to replicate data between two collections using the MongoDB CDC handler.

* [Migrate an Existing Collection to a Time Series Collection](https://www.mongodb.com/docs/kafka-connector/current/tutorials/migrate-time-series/)

Time series collections efficiently store sequences of measurements over a period of time, dramatically increasing the performance of time-based data. In this tutorial, you will configure both a source and sink connector to replicate the data from a collection into a time series collection.

These tutorials run locally within a Docker Compose environment that includes Apache Kafka, Kafka Connect, and MongoDB. Before starting them, follow and complete the [Tutorial Setup](https://www.mongodb.com/docs/kafka-connector/current/tutorials/tutorial-setup/).  You will work through the steps using a tutorial shell and containers available on Docker Hub.  The tutorial shell includes tools such as the new Mongo shell, KafkaCat, and helper scripts that make it easy to configure Kafka Connect from the command line.

![](https://mongodb-devhub-cms.s3.us-west-1.amazonaws.com/mongodb_tutorial_shell_docker_f39a104d7b.png)

If you have any questions or feedback on the tutorials, please post them on the [MongoDB Community Forums](https://www.mongodb.com/community/forums/c/data/connectors-integrations/48). 

MongoDB Documentation has released a series of tutorials based on a self-hosted Docker compose environment that includes all components needed.

Learn How to Leverage MongoDB Data Within Kafka with New Tutorials!

Pavel Duchovny

Lead Developer Advocate

Go is a modern language built on typed and native code compiling concepts while feeling and utilizing some benefits of dynamic languages. It is fairly simple to install and use, as it provides readable and robust code for many application use cases.

One of those use cases is building agents that report to a centralized data platform via streaming. A widely accepted approach is to communicate the agent data through subscription of distributed queues like Kafka. The Kafka topics can then propagate the data to many different sources, such as a [MongoDB Atlas](https://mongodb.com/atlas) cluster. 

Having a Go agent allows us to utilize the same code base for various operating systems, and the fact that it has good [integration with JSON data](https://www.mongodb.com/languages/golang) and packages such as a [MongoDB driver](https://docs.mongodb.com/drivers/go/) and [Confluent Go Kafka Client](https://docs.confluent.io/clients-confluent-kafka-go/current/overview.html) makes it a compelling candidate for the presented use case.

This article will demo how file size data on a host is monitored from a cross-platform agent written in Golang via a Kafka cluster using a Confluent hosted sink connector to MongoDB Atlas. MongoDB Atlas stores the data in a [time series collection](https://docs.mongodb.com/manual/core/timeseries-collections/). The [MongoDB Charts](https://www.mongodb.com/products/charts) product is a convenient way to show the gathered data to the user.
![Architecture Overview](https://mongodb-devhub-cms.s3.us-west-1.amazonaws.com/Mongo_DB_Dev_Hub_Article_Go_to_Mongo_DB_using_Kafka_Pavel_D_f06175feb5.jpg)
## Preparing the Golang project, Kafka cluster, and MongoDB Atlas

### Configuring a Go project

Our agent is going to run Go. Therefore, you will need to install the Go language software on your host.

Once this step is done, we will create a Go module to begin our project in our working directory:
``` shell
go mod init example/main
```
Now we will need to add the Confluent Kafka dependency to our Golang project:
``` shell
go get -u gopkg.in/confluentinc/confluent-kafka-go.v1/kafka
```

### Configuring a Kafka cluster 
Creating a Confluent Kafka Cluster is done via the Confluent UI. Start by [creating](https://docs.confluent.io/cloud/current/clusters/create-cluster.html) a basic Kafka cluster in the [Confluent Cloud](https://confluent.cloud/). Once ready, create a topic to be used in the Kafka cluster. I created one named “files.” 

[Generate](https://docs.confluent.io/cloud/current/access-management/authenticate/api-keys/api-keys.html) an api-key and api-secret to interact with this Kafka cluster. For the simplicity of this tutorial, I have selected the “Global Access” api-key. For production, it is recommended to give as minimum permissions as possible for the api-key used. Get a hold of the generated keys for future use.

Obtain the Kafka cluster connection string via Cluster Overview > Cluster Settings > Identification > Bootstrap server for future use. Basic clusters are open to the internet and in production, you will need to amend the access list for your specific hosts to connect to your cluster via advanced cluster ACLs.

> **Important:** The Confluent connector requires that the Kafka cluster and the Atlas cluster are deployed in the same region.
>
### Configuring Atlas project and cluster
[Create a project and cluster](https://docs.atlas.mongodb.com/getting-started/) or use an existing Atlas cluster in your project. 
![Atlas Cluster](https://mongodb-devhub-cms.s3.us-west-1.amazonaws.com/Screenshot_2022_01_31_at_13_58_17_fb16546b42.png)
Since we are using a time series collection, the clusters must use a 5.0+ version. Prepare your Atlas cluster for a Confluent sink Atlas [connection](https://docs.mongodb.com/kafka-connector/current/introduction/connect/). Inside your project’s access list, enable user and relevant IP addresses of your connector IPs. The access list IPs should be associated to the Atlas Sink Connector, which we will configure in a following section. Finally, get a hold of the Atlas [connection string](https://docs.atlas.mongodb.com/connect-to-cluster/#use-the-connect-to-your-cluster-dialog-to-connect-to-your-cluster) and the main cluster DNS. For more information about best securing and getting the relevant IPs from your Confluent connector, please read the following article: [MongoDB Atlas Sink Connector for Confluent Cloud](https://docs.confluent.io/cloud/current/connectors/cc-mongo-db-sink.html).

## Adding agent main logic
Now that we have our Kafka cluster and Atlas clusters created and prepared, we can initialize our agent code by building a small main file that will monitor my `./files` directory and capture the file names and sizes. I’ve added a file called `test.txt` with some data in it to bring it to ~200MB.

Let’s create a file named `main.go` and write a small logic that performs a constant loop with a 1 min sleep to walk through the files in the `files` folder:
``` go
package main

import (
   "fmt"
   "encoding/json"
   "time"
   "os"
   "path/filepath"
)

type Message struct {
  Name string
  Size float64
  Time int64
}

func samplePath (startPath string)  error {
   err := filepath.Walk(startPath,
    func(path string, info os.FileInfo, err error) error {
 
       var bytes int64
       bytes = info.Size()


       var kilobytes int64
       kilobytes = (bytes / 1024)

       var megabytes float64
       megabytes = (float64)(kilobytes / 1024)   // cast to type float64

       var gigabytes float64
       gigabytes = (megabytes / 1024)

      now := time.Now().Unix()*1000




      m := Message{info.Name(), gigabytes, now}
      value, err := json.Marshal(m)

      
      if err != nil {
         panic(fmt.Sprintf("Failed to parse JSON: %s", err))
      }


      fmt.Printf("value: %v\n", string(value))
      return nil;
  })
  if err != nil {
     return err
  }
  return nil;
}

func main() {
   for {
       err := samplePath("./files");
       if err != nil {
            panic(fmt.Sprintf("Failed to run sample : %s", err))
       }
       time.Sleep(time.Minute)
   }

}
```
The above code simply imports helper modules to traverse the directories and for JSON documents out of the files found. 

Since we need the data to be marked with the time of the sample, it is a great fit for time series data and therefore should eventually be stored in a time series collection on Atlas. If you want to learn more about time series collection and data, please read our article, [MongoDB Time Series Data](http://mongodb.com/features/mongodb-time-series-data).

We can test this agent by running the following command:
``` shell
go run main.go
```
The agent will produce JSON documents similar to the following format:
``` shell
value: {"Name":"files","Size":0,"Time":1643881924000}
value: {"Name":"test.txt","Size":0.185546875,"Time":1643881924000}
```
## Creating a Confluent MongoDB connector for Kafka
Now we are going to create a Kafka Sink connector to write the data coming into the “files” topic to our Atlas Cluster’s time series collection.

Confluent Cloud has a very popular integration running MongoDB’s Kafka connector as a hosted solution integrated with their Kafka clusters. Follow these [steps](https://docs.confluent.io/cloud/current/connectors/cc-mongo-db-sink.html#using-the-ccloud-console) to initiate a connector deployment.

The following are the inputs provided to the connector:
![Connector Setup 1](https://mongodb-devhub-cms.s3.us-west-1.amazonaws.com/Screenshot_2022_01_30_at_14_08_39_e1e3eff816.png)
![Connector Setup 2](https://mongodb-devhub-cms.s3.us-west-1.amazonaws.com/Screenshot_2022_01_30_at_14_08_54_6521f4b53e.png)
![Connector Setup 3](https://mongodb-devhub-cms.s3.us-west-1.amazonaws.com/Screenshot_2022_01_30_at_14_09_04_4de0d474ac.png)
Once you set it up,  following the guide, you will eventually have a similar launch summary page:
![Connector Final Summary](https://mongodb-devhub-cms.s3.us-west-1.amazonaws.com/Screenshot_2022_01_24_at_13_02_08_d460ad2de6.png)
After provisioning every populated document into the `files` queue will be pushed to a time series collection `hostMonitor.files` where the date field is `Time` and metadata field is `Name`.
## Pushing data to Kafka
Now let’s edit the `main.go` file to use a Kafka client and push each file measurement into the “files” queue.

Add the client library as an imported module:
``` go
import (
   "fmt"
   "encoding/json"
   "time"
   "os"
   "path/filepath"
   "github.com/confluentinc/confluent-kafka-go/kafka"
)
```
Add the Confluent cloud credentials and cluster DNS information. Replace `<CONFLUENT-SERVER>:<CONFLUENT-PORT>` found on the Kafka Cluster details page and the `<ACCESS-KEY>` , `<SECRET-KEY>` generated in the Kafka Cluster:
``` go
const (
   bootstrapServers = “<CONFLUENT-SERVER>:<CONFLUENT-PORT>"
   ccloudAPIKey     = "<ACCESS-KEY>"
   ccloudAPISecret  = "<SECRET-KEY>"
)
```
The following code will initiate the producer and produce a message out of the marshaled JSON document:    
``` go
               topic := "files"
              // Produce a new record to the topic...
               producer, err := kafka.NewProducer(&kafka.ConfigMap{
                   "bootstrap.servers":       bootstrapServers,
                   "sasl.mechanisms":         "PLAIN",
                   "security.protocol":       "SASL_SSL",
                   "sasl.username":           ccloudAPIKey,
                   "sasl.password":           ccloudAPISecret})

               if err != nil {
                   panic(fmt.Sprintf("Failed to create producer: %s", err))
               }

               producer.Produce(&kafka.Message{
                   TopicPartition: kafka.TopicPartition{Topic: &topic,
                       Partition: kafka.PartitionAny},
                   Value: []byte(value)}, nil)

               // Wait for delivery report
               e := <-producer.Events()

               message := e.(*kafka.Message)
               if message.TopicPartition.Error != nil {
                   fmt.Printf("failed to deliver message: %v\n",
                       message.TopicPartition)
               } else {
                   fmt.Printf("delivered to topic %s [%d] at offset %v\n",
                       *message.TopicPartition.Topic,
                       message.TopicPartition.Partition,
                       message.TopicPartition.Offset)
               }

               producer.Close()
 ```

The entire `main.go` file will look as follows:
``` go 
package main


import (
   "fmt"
   "encoding/json"
   "time"
   "os"
   "path/filepath"
   "github.com/confluentinc/confluent-kafka-go/kafka")

type Message struct {
  Name string
  Size float64
  Time int64
}


const (
   bootstrapServers = "<SERVER-DNS>:<SERVER-PORT>"
   ccloudAPIKey     = "<ACCESS-KEY>"
   ccloudAPISecret  = "<SECRET-KEY>"
)

func samplePath (startPath string)  error {
  
   err := filepath.Walk(startPath,
    func(path string, info os.FileInfo, err error) error {
    if err != nil {
            return err
    }
    fmt.Println(path, info.Size())

       var bytes int64
       bytes = info.Size()


       var kilobytes int64
       kilobytes = (bytes / 1024)

       var megabytes float64
       megabytes = (float64)(kilobytes / 1024)   // cast to type float64

       var gigabytes float64
       gigabytes = (megabytes / 1024)

      now := time.Now().Unix()*1000


      

       m := Message{info.Name(), gigabytes, now}
       value, err := json.Marshal(m)

      
       if err != nil {
         panic(fmt.Sprintf("Failed to parse JSON: %s", err))
       }


      fmt.Printf("value: %v\n", string(value))

               topic := "files"
               // Produce a new record to the topic...
                   producer, err := kafka.NewProducer(&kafka.ConfigMap{
                       "bootstrap.servers":       bootstrapServers,
                       "sasl.mechanisms":         "PLAIN",
                       "security.protocol":       "SASL_SSL",
                       "sasl.username":           ccloudAPIKey,
                       "sasl.password":           ccloudAPISecret})
  
                   if err != nil {
                       panic(fmt.Sprintf("Failed to create producer: %s", err))
                   }
  
                   producer.Produce(&kafka.Message{
                       TopicPartition: kafka.TopicPartition{Topic: &topic,
                           Partition: kafka.PartitionAny},
                       Value: []byte(value)}, nil)
  
                   // Wait for delivery report
                   e := <-producer.Events()
  
                   message := e.(*kafka.Message)
                   if message.TopicPartition.Error != nil {
                       fmt.Printf("failed to deliver message: %v\n",
                           message.TopicPartition)
                   } else {
                       fmt.Printf("delivered to topic %s [%d] at offset %v\n",
                           *message.TopicPartition.Topic,
                           message.TopicPartition.Partition,
                           message.TopicPartition.Offset)
                   }
  
                   producer.Close()

               return nil;
})
if err != nil {
       return err
}
   return nil;
}

func main() {
   for {
       err := samplePath("./files");
       if err != nil {
            panic(fmt.Sprintf("Failed to run sample : %s", err))
       }
     time.Sleep(time.Minute)
  
   }

}
```
Now when we run the agent while the Confluent Atlas sink connector is fully provisioned, we will see messages produced into the `hostMonitor.files` time series collection:
![Atlas Data](https://mongodb-devhub-cms.s3.us-west-1.amazonaws.com/Screenshot_2022_01_24_at_14_58_42_3f16961049.png)
## Analyzing the data using MongoDB Charts
To put our data into use, we can create some beautiful charts on top of the time series data. In a [line graph](https://docs.mongodb.com/charts/chart-type-reference/line-area-chart/#std-label-line-area-chart-ref), we configure the X axis to use the Time field, the Y axis to use the Size field, and the series to use the Name field. The following graph shows the colored lines represented as the evolution of the different file sizes over time.
![Line Graph Chart](https://mongodb-devhub-cms.s3.us-west-1.amazonaws.com/Screenshot_2022_01_24_at_13_00_53_0b3e2b9d25.png)
Now we have an agent and a fully functioning Charts dashboard to analyze growing files trends. This architecture allows big room for extensibility as the Go agent can have further functionalities, more subscribers can consume the monitored data and act upon it, and finally, MongoDB Atlas and Charts can be used by various applications and embedded to different platforms.

## Wrap Up
Building Go applications is simple yet has big benefits in terms of performance, cross platform code, and a large number of supported libraries and clients. Adding MongoDB Atlas via a Confluent Cloud Kafka service makes the implementation a robust and extensible stack, streaming data and efficiently storing and presenting it to the end user via Charts.

In this tutorial, we have covered all the basics you need to know in order to start using Go, Kafka, and MongoDB Atlas in your next streaming projects.

Try [MongoDB Atlas](https://www.mongodb.com/basics/mongodb-atlas-tutorial) and Go today!

When combined with Confluent Kafka streaming to MongoDB Atlas, Go lets you form tools and applications with real-time data streaming and analytics.

Go to MongoDB Using Kafka Connectors - Ultimate Agent Guide

Diego Rodriguez

Senior Technical Services Engineer

MongoDB Connector for Apache Kafka (MongoDB Connector) is an [open-source Java application](https://github.com/mongodb/mongo-kafka) that works with Apache Kafka Connect enabling seamless data integration of MongoDB with the Apache Kafka ecosystem. When working with the MongoDB Connector, the default values cover a great variety of scenarios, but there are some scenarios that require more fine-grained tuning. In this article, we will walk through important configuration properties that affect the MongoDB Kafka Source and Sink Connectors performance, and share general recommendations.

![](https://mongodb-devhub-cms.s3.us-west-1.amazonaws.com/mongodb_connector_ebf714c66b.png)

## Tuning the source connector

Let’s first take a look at the connector when it is configured to read data from MongoDB and write it into a Kafka topic. When you configure the connector this way, it is known as a “source connector.”  

![](https://mongodb-devhub-cms.s3.us-west-1.amazonaws.com/mongodb_connected_to_apache_kafka_7fc20b4f4c.png)

When the connector is configured as a source, a [change stream](https://docs.mongodb.com/manual/changeStreams/) is opened within the MongoDB cluster based upon any configuration you specified, such as pipeline. These [change stream events](https://docs.mongodb.com/manual/reference/change-events/) get read into the connector and then written out to the Kafka topic, and they resemble the following:

```
{
   _id : { <BSON Object> },
   "operationType" : "<operation>",
   "fullDocument" : { <document> },
   "ns" : {
      "db" : "<database>",
      "coll" : "<collection>"
   },
   "to" : {
      "db" : "<database>",
      "coll" : "<collection>"
   },
   "documentKey" : { "_id" : <value> },
   "updateDescription" : {
      "updatedFields" : { <document> },
      "removedFields" : [ "<field>", ... ],
      "truncatedArrays" : [
         { "field" : <field>, "newSize" : <integer> },
         ...
      ]
   },
   "clusterTime" : <Timestamp>,
   "txnNumber" : <NumberLong>,
   "lsid" : {
      "id" : <UUID>,
      "uid" : <BinData>
   }
}
```
The connector configuration properties help define what data is written out to Kafka. For example, consider the scenario where we insert into MongoDB the following:

```
Use Stocks
db.StockData.insertOne({'symbol':'MDB','price':441.67,'tx_time':Date.now()})
```
When **publish.full.document.only** is set to false (the default setting), the connector writes the entire event as shown below:

```
{"_id": 
{"_data": "826205217F000000022B022C0100296E5A1004AA1707081AA1414BB9F647FD49855EE846645F696400646205217FC26C3DE022E9488E0004"},
"operationType": "insert",
"clusterTime":
  {"$timestamp": 
    {"t": 1644503423, "i": 2}},
"fullDocument":
  {"_id":
    {"$oid": "6205217fc26c3de022e9488e"},
    "symbol": "MDB",
    "price": 441.67,
    "tx_time": 1.644503423267E12},
    "ns":
      {"db": "Stocks", "coll": "StockData"},
      "documentKey":
        {"_id": {"$oid": "6205217fc26c3de022e9488e"}}}}
}
```
When **publish.full.document.only** is set to true and we issue a similar statement, it looks like the following:

```
use Stocks
db.StockData.insertOne({'symbol':'TSLA','price':920.00,'tx_time':Date.now()})
```
We can see that the data written to the Kafka topic is just the changed document itself, which in this example, is an inserted document.

```
{"_id": {"$oid": "620524b89d2c7fb2a606aa16"}, "symbol": "TSLA",
  "price": 920,
  "tx_time": 1.644504248732E12}"}
```
### Resume tokens

Another import concept to understand with source connectors is resume tokens. Resume tokens make it possible for the connector to fail, get restarted, and resume where it left off reading the MongoDB change stream. Resume tokens by default are stored in a Kafka topic defined by the **offset.storage.topic** parameter (configurable at the Kafka Connect Worker level for distributed environments) or in the file system in a file defined by the **offset.storage.file.filename** parameter (configurable at the Kafka Connect Worker level for standalone environments). In the event that the connector has been offline and the underlying MongoDB oplog has rolled over, you may get an error when the connector restarts. Read the [Invalid Resume Token](https://docs.mongodb.com/kafka-connector/current/troubleshooting/recover-from-invalid-resume-token/#std-label-kafka-troubleshoot-invalid-resume-token) section of the online documentation to learn more about this condition.

### Configuration properties

The full set of properties for the Kafka Source Connector can be found [in the documentation](https://docs.mongodb.com/kafka-connector/current/source-connector/configuration-properties/). The properties that should be considered with respect to performance tuning are as follows:

* **batch.size**: the cursor batch size that defines how many change stream documents are retrieved on each **getMore** operation. Defaults to 1,000.
* **poll.await.time.ms**: The maximum amount of time in milliseconds that the server waits for new data changes to report to the change stream cursor before returning an empty batch. Defaults to 5,000.
* **poll.max.batch.size**: maximum number of source records to send to Kafka at once. This setting can be used to limit the amount of data buffered internally in the Connector. Defaults to 1,000.
* **pipeline**: an array of aggregation pipeline stages to run in your change stream. Defaults to an empty pipeline that provides no filtering.
* **copy.existing.max.threads**: the number of threads to use when performing the data copy. Defaults to the number of processors.
* **copy.existing.queue.size**: the max size of the queue to use when copying data. This is buffered internally by the Connector. Defaults to 16,000.

### Recommendations
The following are some general recommendations and considerations when configuring the source connector:

#### Scaling the source
One of the most common questions is how to scale the source connector. For scenarios where you have a large amount of data to be copied via **copy.existing**, keep in mind that using the source connector this way may not be the best way to move this large amount of data.  Consider the process for copy.existing:

* Store the latest change stream resume token.
* Spin up a thread (up to **copy.existing.max.threads**) for each namespace that is being copied.
* When all threads finish, the resume tokens are read, written, and caught up to current time.

While technically, the data will eventually be copied, this process is relatively slow. And if your data size is large and your incoming data is faster than the copy process, the connector may never get into a state where new data changes are handled by the connector.

For high throughput datasets trying to be copied with copy.existing, a typical situation is overwriting the resume token stored in (1) due to high write activity. This breaks the copy.existing functionality, and it will need to be restarted, on top of dealing with the messages that were already processed to the Kafka topic. When this happens, the alternatives are:

* Increase the oplog size to make sure the copy.existing phase can finish.
* Throttle write activity in the source cluster until the copy.existing phase finishes. 

Another option for handling high throughput of change data is to configure multiple source connectors. Each source connector should use a **pipeline** and capture changes from a subset of the total data. Keep in mind that each time you create a source connector pointed to the same MongoDB cluster, it creates a separate change stream. Each change stream requires resources from the MongoDB cluster, and continually adding them will decrease server performance. That said, this degradation may not become noticeable until the amount of connectors reaches the 100+ range, so breaking your collections into five to 10 connector pipelines is the best way to increase source performance. In addition, using several different source connectors on the same namespace changes the total ordering of the data on the sink versus the original order of data in the source cluster.  

#### Tune the change stream pipeline
When building your Kafka Source Connector configuration, ensure you appropriately tune the “pipeline” so that only wanted events are flowing from MongoDB to Kafka Connect, which helps reduce network traffic and processing times. For a detailed pipeline example, check out the [Customize a Pipeline to Filter Change Events](https://docs.mongodb.com/kafka-connector/current/source-connector/usage-examples/custom-pipeline/#customize-a-pipeline-to-filter-change-events) section of the online documentation.

#### Adjust to the source cluster throughput
Your Kafka Source Connector can be watching a set of collections with a low volume of events, or the opposite, a set of collections with a very high volume of events.

In addition, you may want to tune your Kafka Source Connector to react faster to changes, reduce round trips to MongoDB or Kafka, and similar changes.

With this in mind, consider adjusting the following properties for the Kafka Source Connector:

* Adjust the value of **batch.size**:
    * Higher values mean longer processing times on the source cluster but fewer round trips to it. It can also increase the chances of finding relevant change events when the volume of events being watched is small.
    * Lower values mean shorter processing times on the source cluster but more round trips to it. It can reduce the chances of finding relevant change events when the volume of events being watched is small.
* Adjust the value of **poll.max.batch.size**:
    * Higher values require more memory to buffer the source records with fewer round trips to Kafka. This comes at the expense of the memory requirements and increased latency from the moment a change takes place in MongoDB to the point the Kafka message associated with that change reaches the destination topic.
    * Lower values require less memory to buffer the source records with more round trips to Kafka. It can also help reduce the latency from the moment a change takes place in MongoDB to the point the Kafka message associated with that change reaches the destination topic.
* Adjust the value of **poll.await.time.ms**:
    * Higher values can allow source clusters with a low volume of events to have any information to be sent to Kafka at the expense of increased latency from the moment a change takes place in MongoDB to the point the Kafka message associated with that change reaches the destination topic.
    * Lower values reduce latency from the moment a change takes place in MongoDB to the point the Kafka message associated with that change reaches the destination topic. But for source clusters with a low volume of events, it can prevent them from having any information to be sent to Kafka.

This information is an overview of what to expect when changing these values, but keep in mind that they are deeply interconnected, with the volume of change events on the source cluster having an important impact too:

1. The Kafka Source Connector issues getMore commands to the source cluster using **batch.size**.
2. The Kafka Source Connector receives the results from step 1 and waits until either **poll.max.batch.size** or **poll.await.time.ms** is reached. While this doesn’t happen, the Kafka Source Connector keeps “feeding” itself with more getMore results.
3. When either **poll.max.batch.size** or **poll.await.time.ms** is reached, the source records are sent to Kafka.

#### “Copy existing” feature 

When running with the **copy.existing** property set to **true**, consider these additional properties:

* **copy.existing.queue.size**: the amount of records the Kafka Source Connector buffers internally. This queue and its size include all the namespaces to be copied by the “Copy Existing” feature. If this queue is full, the Kafka Source Connector blocks until space becomes available.
* **copy.existing.max.threads**: the amount of concurrent threads used for copying the different namespaces. There is a one namespace to one thread mapping, so it is common to increase this up to the maximum number of namespaces being copied. If the number exceeds the number of cores available in the system, then the performance gains can be reduced.
* **copy.existing.allow.disk.use**: allows the copy existing aggregation to use temporary disk storage if required. The default is set to true but should be set to false if the user doesn't have the permissions for disk access.

#### Memory implications 

If you experience JVM “out of memory” issues on the Kafka Connect Worker process, you can try reducing the following two properties that control the amount of data buffered internally:

* **poll.max.batch.size**
* **copy.existing.queue.size**: applicable if the “copy.existing” property is set to true.

It is important to note that lowering these values can result in unwanted impact. Adjusting the JVM Heap Size to your environment needs is recommended as long as you have available resources and the memory needs are not the result of memory leaks.

## Tuning the sink connector
When the MongoDB Connector is configured as a sink, it reads from a Kafka topic and writes to a MongoDB collection.

![](https://mongodb-devhub-cms.s3.us-west-1.amazonaws.com/image1_c8f32992a5.png)

As with the source, there exists a mechanism to ensure offsets are stored in the event of a sink failure. Kafka connect manages this, and the information is stored in the __consumer_offsets topic. The MongoDB Connector has configuration properties that affect performance. They are as follows:

* **max.batch.size**: the maximum number of sink records to batch together for processing. A higher number will result in more documents being sent as part of a single bulk command. Default value is 0.
* **rate.limiting.every.n**: number of processed batches that trigger the rate limit. A value of 0 means no rate limiting. Default value is 0. In practice, this setting is rarely used.
* **rate.limiting.timeout**: how long (in milliseconds) to wait before continuing to process data once the rate limit is reached. Default value is 0. This setting is rarely used.
* **tasks.max**: the maximum number of tasks. Default value is 1.

### Recommendations 
#### Add indexes to your collections for consistent performance 
Writes performed by the sink connector take additional time to complete as the size of the underlying MongoDB collection grows. To prevent performance deterioration, use an index to support these write queries.

#### Achieve as much parallelism as possible 
The Kafka Sink Connector (KSC) can take advantage of parallel execution thanks to the **tasks.max** property. The specified number of tasks will only be created if the source topic has the same number of partitions. Note: A partition should be considered as a logic group of ordered records, and the producer of the data determines what each partition contains.
Here is the breakdown of the different combinations of number of partitions in the source topic and tasks.max values:

**If working with more than one partition but one task:**

* The task processes partitions one by one: Once a batch from a partition is processed, it moves on to another one so the order within each partition is still guaranteed.
* Order among all the partitions is not guaranteed.

**If working with more than one partition and an equal number of tasks:**

* Each task is assigned one partition and the order is guaranteed within each partition.
* Order among all the partitions is not guaranteed.

**If working with more than one partition and a smaller number of tasks:**

* The tasks that are assigned more than one partition process partitions one by one: Once a batch from a partition is processed, it moves on to another one so the order within each partition is still guaranteed.
* Order among all the partitions is not guaranteed.

**If working with more than one partition and a higher number of tasks:**

* Each task is assigned one partition and the order is guaranteed within each partition.
* KSC will not generate an excess number of tasks.
* Order among all the partitions is not guaranteed.

Processing of partitions may not be in order, meaning that Partition B may be processed before Partition A. All messages within the partition conserve strict order.

Note: When using MongoDB to write CDC data, the order of data is important since, for example, you do not want to process a delete before an update on the same data. If you specify more than one partition for CDC data, you run the risk of data being out of order on the sink collection.

#### Tune the bulk operations 
The Kafka Sink Connector (KSC) works by issuing [bulk write operations](https://docs.mongodb.com/manual/core/bulk-write-operations/). All the bulk operations that the KSC executes are, by default, ordered and as such, the order of the messages is guaranteed within a partition. See [Ordered vs Unordered Operations](https://docs.mongodb.com/manual/core/bulk-write-operations/#ordered-vs-unordered-operations) for more information.  Note: As of 1.7, **bulk.write.ordered**, if set to false, will process the bulk out of order, enabling more documents within the batch to be written in the case of a failure of a portion of the batch.

The amount of operations that are sent in a single bulk command can have a direct impact on performance. You can modify this by adjusting **max.batch.size**:

* A higher number will result in more operations being sent as part of a single bulk command. This helps improve throughput at the expense of some added latency. However, a very big number might result in cache pressure on the destination cluster.
* A small number will ease the potential cache pressure issues which might be useful for destination clusters with fewer resources. However, throughput decreases, and you might experience consumer lag on the source topics as the producer might publish messages in the topic faster than the KSC processes them.
* This value affects processing within each of the tasks of the KSC.

#### Throttle the Kafka sink connector 
In the event that the destination MongoDB cluster is not able to handle consistent throughput, you can configure a throttling mechanism. You can do this with two properties:

* **rate.limiting.every.n**: number of processed batches that should trigger the rate limit. A value of 0 means no rate limiting.
* **rate.limiting.timeout**: how long (in milliseconds) to wait before continuing to process data once the rate limit is reached.

The end result is that whenever the KSC writes **rate.limiting.every.n** number of batches, it waits **rate.limiting.timeout milliseconds** before writing the next batch. This allows a destination MongoDB cluster that cannot handle consistent throughput to recover before receiving new load from the KSC.

To build a MongoDB/Apache Kafka solution, the default configuration values need some tweaks. Learn important configuration properties and best practices. 

Tuning the MongoDB Connector for Apache Kafka

Michael Lynn

Principal Developer Advocate

In this podcast episode transcript, we chat with Tim Fogarty, Varsha Subrahmanyam, and Evgeni Dobranov about BI Connector and mongomirror.

MongoDB Podcast Interview With Connectors and Translators Team

MongoDB has released a version 10 of the MongoDB Connector for Apache Spark that leverages the new Spark Data Sources API V2 with support for Spark Structured Streaming.

## Why a new version?

The current version of the [MongoDB Spark Connector](https://www.mongodb.com/docs/spark-connector/current/) was originally written in 2016 and is based upon V1 of the Spark Data Sources API. While this API version is still supported, Databricks has released an updated version of the API, making it easier for data sources like MongoDB to work with Spark. By having the MongoDB Spark Connector use V2 of the API, an immediate benefit is a tighter integration with Spark [Structured Streaming](https://spark.apache.org/docs/latest/structured-streaming-programming-guide.html#overview).

*Note: With respect to the previous version of the MongoDB Spark Connector that supported the V1 API, MongoDB will continue to support this release until such a time as Databricks depreciates V1 of the Data Source API. While no new features will be implemented, upgrades to the connector will include bug fixes and support for the current versions of Spark only.*

## What version should I use?

The new MongoDB Spark Connector release (Version 10.1) is not intended to be a direct replacement for your applications that use the previous version of MongoDB Spark Connector.

The new Connector uses a different namespace with a short name, “mongodb” (full path is  “com.mongodb.spark.sql.connector.MongoTableProvider”), versus “mongo” (full path of  “com.mongodb.spark.DefaultSource”). Having a different namespace makes it possible to use both versions of the connector within the same Spark application! This is helpful in unit testing your application with the new Connector and making the transition on your timeline.  

Also, we are changing how we version the MongoDB Spark Connector. The previous versions of the MongoDB Spark Connector aligned with the version of Spark that was supported—e.g., Version 2.4 of the MongoDB Spark Connector works with Spark 2.4. Keep in mind that going forward, this will not be the case. The MongoDB documentation will make this clear as to which versions of Spark the connector supports.

## Structured Streaming with MongoDB using continuous mode

Apache Spark comes with a stream processing engine called [Structured Streaming](https://spark.apache.org/docs/latest/structured-streaming-programming-guide.html#overview), which is based on Spark's SQL engine and DataFrame APIs. Spark Structured Streaming treats each incoming stream of data as a micro-batch, continually appending each micro-batch to the target dataset. This makes it easy to convert existing Spark batch jobs into a streaming job. Structured Streaming has evolved over Spark releases and in Spark 2.3 introduced Continuous Processing mode, which took the micro-batch latency from over 100ms to about 1ms. Note this feature is still in experimental mode according to the [official Spark Documentation](https://spark.apache.org/docs/latest/structured-streaming-programming-guide.html#continuous-processing). In the following example, we’ll show you how to stream data between MongoDB and Spark using Structured Streams and continuous processing. First, we’ll look at reading data from MongoDB.

### Reading streaming data from MongoDB

You can stream data from MongoDB to Spark using the new Spark Connector. Consider the following example that streams stock data from a MongoDB Atlas cluster. A sample document in MongoDB is as follows:

```
{
  _id: ObjectId("624767546df0f7dd8783f300"),
  company_symbol: 'HSL',
  company_name: 'HUNGRY SYNDROME LLC',
  price: 45.74,
  tx_time: '2022-04-01T16:57:56Z'
}
```
In this code example, we will use the new MongoDB Spark Connector and read from the StockData collection. When the Spark Connector opens a streaming read connection to MongoDB, it opens the connection and creates a [MongoDB Change Stream](https://www.mongodb.com/docs/manual/changeStreams/) for the given database and collection. A change stream is used to subscribe to changes in MongoDB. As data is inserted, updated, and deleted, change stream events are created. It’s these [change events](https://www.mongodb.com/docs/manual/reference/change-events/) that are passed back to the client in this case the Spark application. There are configuration options that can change the structure of this event message. For example, if you want to return just the document itself and not include the change stream event metadata, set  “spark.mongodb.change.stream.publish.full.document.only” to true.

```
from pyspark import SparkContext
from pyspark.streaming import StreamingContext
from pyspark.sql import SparkSession
from pyspark.sql.functions import *

spark = SparkSession.\
        builder.\
        appName("streamingExampleRead").\
        config('spark.jars.packages', 'org.mongodb.spark:mongo-spark-connector_2.12::10.1.1').\
        getOrCreate()

query=(spark.readStream.format("mongodb")
.option('spark.mongodb.connection.uri', '<CONNECTION STRING>')
    	.option('spark.mongodb.database', 'Stocks') \
    	.option('spark.mongodb.collection', 'StockData') \
.option('spark.mongodb.change.stream.publish.full.document.only','true') \
    	.option("forceDeleteTempCheckpointLocation", "true") \
    	.load())

query.printSchema()
```

The schema is inferred from the MongoDB collection. You can see from the printSchema command that our document structure is as follows:


| root: |  |  |  |
| --- | --- | --- | --- |
|  |_id  |string  |(nullable=true)  |
|  |company_name| string  | (nullable=true) |
|  | company_symbol | string  | (nullable=true) |
|  | price | double | (nullable=true) |
|  | tx_time | string  | (nullable=true) |

We can verify that the dataset is streaming with the isStreaming command.

```
query.isStreaming
```

Next, let’s read the data on the console as it gets inserted into MongoDB.

```
query2=(query.writeStream \
  .outputMode("append") \
  .option("forceDeleteTempCheckpointLocation", "true") \
  .format("console") \
  .trigger(continuous="1 second")
  .start().awaitTermination());
```

When the above code was run through spark-submit, the output resembled the following:

… removed for brevity …

-------------------------------------------
Batch: 2
-------------------------------------------
+--------------------+--------------------+--------------+-----+-------------------+
|                 _id|        company_name|company_symbol|price|            tx_time|
+--------------------+--------------------+--------------+-----+-------------------+
|62476caa6df0f7dd8...| HUNGRY SYNDROME LLC|           HSL|45.99|2022-04-01 17:20:42|
|62476caa6df0f7dd8...|APPETIZING MARGIN...|           AMP|12.81|2022-04-01 17:20:42|
|62476caa6df0f7dd8...|EMBARRASSED COCKT...|           ECC|38.18|2022-04-01 17:20:42|
|62476caa6df0f7dd8...|PERFECT INJURY CO...|           PIC|86.85|2022-04-01 17:20:42|
|62476caa6df0f7dd8...|GIDDY INNOVATIONS...|           GMI|84.46|2022-04-01 17:20:42|
+--------------------+--------------------+--------------+-----+-------------------+

… removed for brevity …

-------------------------------------------

Batch: 3
-------------------------------------------
+--------------------+--------------------+--------------+-----+-------------------+
|                 _id|        company_name|company_symbol|price|            tx_time|
+--------------------+--------------------+--------------+-----+-------------------+
|62476cab6df0f7dd8...| HUNGRY SYNDROME LLC|           HSL|46.04|2022-04-01 17:20:43|
|62476cab6df0f7dd8...|APPETIZING MARGIN...|           AMP| 12.8|2022-04-01 17:20:43|
|62476cab6df0f7dd8...|EMBARRASSED COCKT...|           ECC| 38.2|2022-04-01 17:20:43|
|62476cab6df0f7dd8...|PERFECT INJURY CO...|           PIC|86.85|2022-04-01 17:20:43|
|62476cab6df0f7dd8...|GIDDY INNOVATIONS...|           GMI|84.46|2022-04-01 17:20:43|
+--------------------+--------------------+--------------+-----+-------------------+
 
### Writing streaming data to MongoDB

Next, let’s consider an example where we stream data from Apache Kafka to MongoDB. Here the source is a kafka topic “stockdata.Stocks.StockData.” As data arrives in this topic, it’s run through Spark with the message contents being parsed, transformed, and written into MongoDB. Here is the code listing with comments in-line:

```
from pyspark import SparkContext
from pyspark.streaming import StreamingContext
from pyspark.sql import SparkSession
from pyspark.sql import functions as F
from pyspark.sql.functions import *
from pyspark.sql.types import StructType,TimestampType, DoubleType, StringType, StructField

spark = SparkSession.\
        builder.\
        appName("streamingExampleWrite").\
        config('spark.jars.packages', 'org.mongodb.spark:mongo-spark-connector:10.1.1').\
        config('spark.jars.packages', 'org.apache.spark:spark-sql-kafka-0-10_2.12:3.0.0').\
        getOrCreate()

df = spark \
  .readStream \
  .format("kafka") \
  .option("startingOffsets", "earliest") \
  .option("kafka.bootstrap.servers", "KAFKA BROKER HOST HERE") \
  .option("subscribe", "stockdata.Stocks.StockData") \
  .load()

schemaStock = StructType([ \
    StructField("_id",StringType(),True), \
    StructField("company_name",StringType(), True), \
    StructField("company_symbol",StringType(), True), \
    StructField("price",StringType(), True), \
    StructField("tx_time",StringType(), True)])

schemaKafka = StructType([ \
    StructField("payload",StringType(),True)])
```

Note that Kafka topic message arrives in this format -> key (binary), value (binary), topic (string), partition (int), offset (long), timestamp (long), timestamptype (int). See [Structured Streaming + Kafka Integration Guide (Kafka broker version 0.10.0 or higher)](https://spark.apache.org/docs/2.3.4/structured-streaming-kafka-integration.html) for more information on the Kafka and Spark integration.

To process the message for consumption into MongoDB, we want to pick out the value which is in binary format and convert it to JSON.

```
stockDF=df.selectExpr("CAST(value AS STRING)")
```

For reference, here is an example of an event (the value converted into a string) that is on the Kafka topic:

```
{
  "schema": {
    "type": "string",
    "optional": false
  },
  "payload": "{\"_id\": {\"$oid\": \"6249f8096df0f7dd8785d70a\"}, \"company_symbol\": \"GMI\", \"company_name\": \"GIDDY INNOVATIONS\", \"price\": 87.57, \"tx_time\": \"2022-04-03T15:39:53Z\"}"
}
```

We want to isolate the payload field and convert it to a JSON representation leveraging the shcemaStock defined above. For clarity, we have broken up the operation into multiple steps to explain the process. First, we want to convert the value into JSON.

```
stockDF=stockDF.select(from_json(col('value'),schemaKafka).alias("json_data")).selectExpr('json_data.*')
```

The dataset now contains data that resembles

```
…
 {
    _id: ObjectId("624c6206e152b632f88a8ee2"),
    payload: '{"_id": {"$oid": "6249f8046df0f7dd8785d6f1"}, "company_symbol": "GMI", "company_name": "GIDDY MONASTICISM INNOVATIONS", "price": 87.62, "tx_time": "2022-04-03T15:39:48Z"}'
  }, …
```

Next, we want to capture just the value of the payload field and convert that into JSON since it’s stored as a string.

```
stockDF=stockDF.select(from_json(col('payload'),schemaStock).alias("json_data2")).selectExpr('json_data2.*')
```

Now we can do whatever transforms we would like to do on the data. In this case, let’s convert the tx_time into a timestamp.

```
stockDF=stockDF.withColumn("tx_time",col("tx_time").cast("timestamp"))
```

The Dataset is in a format that’s ready for consumption into MongoDB, so let’s stream it out to MongoDB. To do this, use the writeStream method. Keep in mind there are various options to set. For example, when present, the “trigger” option processes the results in batches. In this example, it’s every 10 seconds. Removing the trigger field will result in continuous writing. For more information on options and parameters, check out the Structured Streaming Guide.

```
dsw = (
  stockDF.writeStream
    .format("mongodb")
    .queryName("ToMDB")
    .option("checkpointLocation", "/tmp/pyspark7/")
    .option("forceDeleteTempCheckpointLocation", "true")
    .option('spark.mongodb.connection.uri', ‘<CONNECTION STRING>')
    .option('spark.mongodb.database', 'Stocks')
    .option('spark.mongodb.collection', 'Sink')
    .trigger(continuous="10 seconds")
    .outputMode("append")
    .start().awaitTermination());
```

## Structured Streaming with MongoDB using Microbatch mode
While continuous mode offers a lot of promise in terms of the latency and performance characteristics, the support for various popular connectors like AWS S3 for example is non-existent.  Thus, you might end up using microbatch mode within your solution.  The key difference between the two is how spark handles obtaining the data from the stream.  As mentioned previously, the data is batched and processed versus using a continuous append to a table.  The noticeable difference is the advertised latency of microbatch around 100ms which for most workloads might not be an issue.
### Reading streaming data from MongoDB using microbatch

Unlike when we specify a write, when we read from MongoDB, there is no special configuration to tell Spark to use microbatch or continuous. This behavior is determined only when you write.  Thus, in our code example, to read from MongoDB is the same in both cases, e.g.:

```
query=(spark.readStream.format("mongodb").\
	option('spark.mongodb.connection.uri', '<<MONGODB CONNECTION STRING>>').\
	option('spark.mongodb.database', 'Stocks').\
	option('spark.mongodb.collection', 'StockData').\
	option('spark.mongodb.change.stream.publish.full.document.only','true').\
	option("forceDeleteTempCheckpointLocation", "true").\
	load())
```

Recall from the previous discussion on reading MongoDB data, when using `spark.readStream.format("mongodb")`, MongoDB opens a [change stream](https://www.mongodb.com/docs/manual/changeStreams/) and subscribes to changes as they occur in the database.  With microbatch each microbatch event opens a new change stream cursor making this form of microbatch streaming less efficient than continuous streams.  That said, some consumers of streaming data such as AWS S3 only support data from microbatch streams.

### Writing streaming data to MongoDB using microbatch
Consider the previous writeStream example code:

```
dsw = (
  stockDF.writeStream
    .format("mongodb")
    .queryName("ToMDB")
    .option("checkpointLocation", "/tmp/pyspark7/")
    .option("forceDeleteTempCheckpointLocation", "true")
    .option('spark.mongodb.connection.uri', '<<MONGODB CONNECTION STRING>>')
    .option('spark.mongodb.database', 'Stocks')
    .option('spark.mongodb.collection', 'Sink')
    .trigger(continuous="10 seconds")
    .outputMode("append")
    .start().awaitTermination());
```

Here the .trigger parameter was used to tell Spark to use Continuous mode streaming, to use microbatch simply remove the .trigger parameter.

## Go forth and stream!

Streaming data is a critical component of many types of applications. MongoDB has evolved over the years, continually adding features and functionality to support these types of workloads. With the MongoDB Spark Connector version 10.1, you can quickly stream data to and from MongoDB with a few lines of code.

For more information and examples on the new MongoDB Spark Connector version 10.1, check out the [online documentation](https://docs.mongodb.com/spark-connector). Have questions about the connector or MongoDB? Post a question in the [MongoDB Developer Community Connectors & Integrations](https://www.mongodb.com/community/forums/c/data/connectors-integrations/48) forum.

MongoDB has released MongoDB Spark Connector V10. Learn how to read from and write to MongoDB through Spark Structured Streaming.

Streaming Data With Apache Spark and MongoDB

Venkatesh Shanbhag

Partner Solution Architect

Partner

Zi Wang

Partner Engineer at Google Cloud

Maruti C

Global Partner Engineering Lead, Google

To empower enterprises that strive to transform their data into insights, BigQuery has emerged as a powerful, scalable, cloud-based data warehouse solution offered by Google Cloud Platform (GCP). Its cloud-based approach allows efficient data management and manipulation, making BigQuery a game-changer for businesses seeking advanced data insights. Notably, one of BigQuery’s standout features is its seamless integration with Spark-based data processing that enables users to further enhance their queries. Now, leveraging BigQuery APIs, users can create and execute Spark stored procedures, which are reusable code modules that can encapsulate complex business logic and data transformations. This feature helps data engineers, data scientists, and data analysts take advantage of BigQuery’s advanced capabilities and Spark’s robust data processing capabilities.

MongoDB, a developer data platform, is a popular choice for storing and managing operational data for its scalability, performance, flexible schema, and real-time capabilities (change streams and aggregation). By combining the capabilities of BigQuery with the versatility of Apache Spark and the flexibility of MongoDB, you can unlock a powerful data processing pipeline.

Apache Spark is a powerful open-source distributed computing framework that excels at processing large amounts of data quickly and efficiently. It supports a wide range of data formats, including structured, semi-structured, and unstructured data, making it an ideal choice for integrating data from various sources, such as MongoDB.

BigQuery Spark stored procedures are routines that are executed within the BigQuery environment. These procedures can perform various tasks, such as data manipulation, complex calculations, and even external data integration. They provide a way to modularize and reuse code, making it easier to maintain and optimize data processing workflows. Spark stored procedures use the [serverless Spark](https://cloud.google.com/solutions/spark) engine that enables serverless, autoscaling Spark. However, you don’t need to enable Dataproc APIs or be charged for Dataproc when you leverage this new capability. 

Let's explore how to extend BigQuery’s data processing to Apache Spark, and integrate MongoDB with BigQuery to effectively facilitate data movement between the two platforms.

## Connecting them together

![Data movement between MongoDB and BigQuery using Spark stored procedure running on serverless Spark][1]

This tutorial guides you through creating a PySpark procedure using the BigQuery editor.

Before we start with the Spark stored procedure setup, you need to [upload](https://cloud.google.com/storage/docs/uploading-objects#uploading-an-object) the [MongoDB Spark connector JAR](https://repo1.maven.org/maven2/org/mongodb/spark/) file to Google Cloud Storage to connect and read from MongoDB Atlas. Copy and save the gsutil URI for the JAR file that will be used in upcoming steps.

![Copy the path for gsutil URI][2]

1. As a prerequisite for completing the tutorial, you need to [set up](https://www.mongodb.com/docs/atlas/getting-started/) a MongoDB Atlas cluster with [sample data](https://www.mongodb.com/docs/guides/atlas/sample-data/) loaded to it. 
2. Navigate to the [BigQuery page](https://console.cloud.google.com/bigquery) on the Google Cloud console.
3. [Create](https://cloud.google.com/bigquery/docs/datasets#create-dataset) a **BigQuery dataset** with the name **spark_run**.
4. You will type the PySpark code directly into the query editor. To create a PySpark stored procedure, click on **Create Pyspark Procedure**, and then select **Create PySpark Procedure**.

![Creating PySPark procedure from BigQuery explorer][3]

5. To set options, click **More** > **PySpark Options**, and then do the following:

    1. Specify the location where you want to run the PySpark code.

    2. In the **Connection** field, click on **CREATE NEW CONNECTION** and enter the below values for each field.

        1. Connection Type > Apache Spark

        2. Connection id > Name it as **mongodb-to-bigquery** for the sake of this tutorial.
        3. Leave the other option empty and click on **Create**.

    3. In the Stored procedure invocation section select **Set a dataset for stored procedure invocation** to **spark_run.**
    4. Click on the advanced options. Copy the gsutil URI name copied at the beginning of the setup and paste in the **JAR files in the pre-requisite**. Leave the other options empty. Press enter and click on **Save**.

6. Open a new tab and go to BigQuery. Navigate to the **External connections** > Find the **mongodb-to-bigquery Connection** > Copy the **Service account id**. [Grant](https://cloud.google.com/iam/docs/grant-role-console#grant_an_iam_role) BigQuery Storage Admin, Secret Manager Secret Accessor, and Storage Object Admin access to this service account from IAM. 

![External connections created using Spark stored procedures][4]

7. (Optional) [Add your username and password](https://cloud.google.com/secret-manager/docs/creating-and-accessing-secrets#create) into Google Cloud Secret Manager, or you can hardcode it in the MongoDB URI string itself. 
8. Copy the below Python script in the PySpark procedure editor and click on **RUN**. The snippet takes around two to three minutes to complete. The below script will create a new table under dataset **spark_run** with the name **sample_mflix_comments**.

```python
from pyspark.sql import SparkSession
from google.cloud import secretmanager

def access_secret_version(secret_id, project_id):
    client = secretmanager.SecretManagerServiceClient()
    name = f"projects/{project_id}/secrets/{secret_id}/versions/1"
    response = client.access_secret_version(request={"name": name})
    payload = response.payload.data.decode("UTF-8")
    return payload
# Update project_number, username_secret_id and password_secret_id, comment them out if you did not create the secrets earlier 

project_id = "<Your project number, 12 digit number>"
username_secret_id = "<Your username secret id>"
password_secret_id = "<Your password secret id>"

username = access_secret_version(username_secret_id, project_id)
password = access_secret_version(password_secret_id, project_id)

  # Update the mongodb_uri directly if with your username and password if you did not create a secret from Step 7, update the hostname with your hostname
mongodb_uri = "mongodb+srv://"+username+":"+password+"@<hostname>/sample_mflix.comments"

my_spark = SparkSession \
  .builder \
  .appName("myApp") \
  .config("spark.mongodb.read.connection.uri", mongodb_uri) \
  .config("spark.mongodb.write.connection.uri", mongodb_uri) \
  .getOrCreate()


dataFrame = my_spark.read.format("mongodb").option("database", "sample_mflix").option("collection", "comments").load()

dataFrame.show()

# Saving the data to BigQuery
dataFrame.write.format("bigquery") \
  .option("writeMethod", "direct") \
    .save("spark_run.sample_mflix_comments")
```

![Stored procedure on Google BigQuery Explorer][5]

9. Navigate to the spark_run dataset to validate the data is loaded from MongoDB Atlas to BigQuery under the table with the name **sample_mflix_comments**.
10. Now that the data is in BigQuery, leverage BQML to run some Generative AI on the new MongoDB data in BigQuery. 
    1. Create a connection with the name **gentext-conn,** using either the [console](https://cloud.google.com/bigquery/docs/create-cloud-resource-connection#create-cloud-resource-connection) or bq command line with connection type as CLOUD_RESOURCE.

```
!bq mk \
  --connection \
  --location=US \
  --project_id=<GCP Project id> \
  --connection_type=CLOUD_RESOURCE gentext-conn 
```

11. To grant IAM permissions to access Vertex AI from BigQuery, navigate to **External connections** > Find the **gettext-conn** connection > Copy the **Service account id**. Grant the **Vertex AI User** access to this service account from **IAM**.
12. Create a model using the CREATE MODEL command.

```
CREATE OR REPLACE MODEL `gcp-pov.spark_run.llm_model`
REMOTE WITH CONNECTION `us.gentext-conn`
OPTIONS (ENDPOINT = 'gemini-pro');
```

13. Run the SQL command against the BigQuery table. This query allows the user to extract the host name from the email leveraging the Gemini Pro model. The resulting [output](https://cloud.google.com/bigquery/docs/reference/standard-sql/bigqueryml-syntax-generate-text#output) includes the response and safety attributes.

```
SELECT prompt,ml_generate_text_result
FROM
ML.GENERATE_TEXT( MODEL `gcp-pov.spark_run.llm_model`,
  (
    SELECT CONCAT('Extract the host name from the email: ', email) AS prompt,
      *  FROM  `gcp-pov.spark_run.sample_mflix_comments`
    LIMIT 5),
  STRUCT(
    0.9 AS temperature,
    100 AS max_output_tokens
    )
  );
```

14. Here is the [sample output](https://cloud.google.com/bigquery/docs/reference/standard-sql/bigqueryml-syntax-generate-text#output) showing the prompt as well as the response. The prompt parameter provides the text for the model to analyze. Prompt design can strongly affect the responses returned by the LLM.

![Output after running the Spark stored procedure][6]

## Benefits of the integration

1. This integration enables you to use MongoDB for OLTP and BigQuery for OLAP, providing a complete data management solution.
2. Once the data is transformed and copied to BigQuery, BigQuery ML lets you [create and run machine learning (ML) models](https://cloud.google.com/bigquery/docs/e2e-journey) by using GoogleSQL queries. 
3. BigQuery ML also lets you [access LLMs and Cloud AI APIs](https://cloud.google.com/bigquery/docs/ai-application-overview) to perform artificial intelligence (AI) tasks like text generation and machine translation. 

## Conclusion

By combining the power of BigQuery, Spark stored procedures, and MongoDB, you can create a robust and scalable data processing pipeline that leverages the strengths of each technology. BigQuery provides a reliable and scalable data warehouse for storing and analyzing structured data, while Spark allows you to process and transform data from various sources, including semi-structured and unstructured data from MongoDB. Spark stored procedures enable you to encapsulate and reuse this logic, making it easier to maintain and optimize your data processing workflows.

### Further reading

1. [Get started](https://www.mongodb.com/docs/atlas/getting-started/) with MongoDB Atlas on Google Cloud.
2. Work with [stored procedures](https://cloud.google.com/bigquery/docs/spark-procedures) for Apache Spark.
3. [Create](https://cloud.google.com/bigquery/docs/create-machine-learning-model) machine learning models in BigQuery ML.


  [1]: https://images.contentstack.io/v3/assets/blt39790b633ee0d5a7/blt3cc99ff9b6ad9cec/66155da90c478454e8a349f1/1.png
  [2]: https://images.contentstack.io/v3/assets/blt39790b633ee0d5a7/bltdd635dea2d750e73/66155dc254d7c1521e8eea3a/2.png
  [3]: https://images.contentstack.io/v3/assets/blt39790b633ee0d5a7/blta48182269e87f9e7/66155dd2cbc2fbae6d8175ea/3.png
  [4]: https://images.contentstack.io/v3/assets/blt39790b633ee0d5a7/bltb399b53e83efffb9/66155de5be36f52825d96ea5/4.png
  [5]: https://images.contentstack.io/v3/assets/blt39790b633ee0d5a7/blt6efef490b3d34cf0/66155dfd2b98e91579101401/5.png
  [6]: https://images.contentstack.io/v3/assets/blt39790b633ee0d5a7/blt56b91d83e11dea5f/66155e0f7cacdc153bd4a78b/6.png

Explore how to extend BigQuery’s data processing to Apache Spark, and integrate MongoDB with BigQuery to facilitate data movement between the two.

Spark Up Your MongoDB and BigQuery Using BigQuery Spark Stored Procedures

MongoDB

Jagadish Nallapaneni

[AWS Identity and Access Management (IAM)](https://aws.amazon.com/iam/) is a widely used service that enables developers to control secure access to AWS services and resources. A key benefit of AWS IAM is its ability to make secure authenticated connections without the client having to pass sensitive passwords.

As of version 1.13, the MongoDB Connector for Apache Kafka (Kafka Connector) now supports authentication via AWS IAM. AWS IAM support has been one of the most commonly requested features for the Kafka Connector, and in this tutorial, we will walk through how to use it.

To achieve support for AWS IAM, we added support for overriding the MongoClient interface allowing a developer to provide a *CustomCredentialProvider*. While this provider can be used for any authentication method, this tutorial will focus on building an AWS IAM credential provider.

To illustrate this, we will walk through a simple configuration of an AWS EC2 instance installed with Apache Kafka. This instance will use the [MongoDB Connector for Apache Kafka](https://www.mongodb.com/docs/kafka-connector/current/) to [sink](https://en.wikipedia.org/wiki/Sink_(computing)) data from a Kafka topic to an Atlas cluster, and authenticate using AWS IAM.

## Prerequisites

If you wish to follow along in this article, make sure you have the following:

- An AWS IAM role, “KafkaBlogAtlasRole”. This role will later be mapped as an Atlas database user.
- MongoDB Atlas cluster. If you do not have an Atlas cluster, learn [how to get started with Atlas](https://www.mongodb.com/docs/atlas/getting-started/)..
- AWS EC2 instance
    - Install [Apache Kafka](https://kafka.apache.org/downloads).
    - Configure the EC2 instance with an AWS IAM role, “KafkaConnectBlogRole.”
    - Add a policy to KafkaConnectBlogRole that allows “AssumeRole” on the KafkaBlogAtlasRole.

Example policy:

```json
{
    "Version": "2012-10-17",
    "Statement": [
        {
            "Sid": "KafkaAtlasBlogPolicy",
            "Effect": "Allow",
            "Action": "sts:AssumeRole",
            "Resource": "arn:aws:iam::979559056307:role/KafkaBlogAtlasRole"
        }
    ]
}
```


- Java Development Kit version 8 or later
- A database user in the desired MongoDB Atlas cluster

    - Configure the database user with AWS IAM and use the AWS Role ARN for KafkaBlogAtlasRole. The Atlas database user should have access to read/write to the desired collection.

![MongoDB Atlas edit user dialog showing AWS IAM authentication method][1]

At this point, we are ready to start building the custom authentication provider.

## Getting started

Starting with version 1.13 of the MongoDB Connector for Apache Kafka, we now have the ability to plug in a custom authentication provider. In this article, we will create a custom authentication provider in Java, compile it, and add it to the lib directory of your Kafka deployment. The provider we will create will support AWS IAM authentication. 

### Building the custom authentication component

To start, in your favorite Java editor, create a project and build the following JAR that contains this source:

SampleAssumeRoleCredential.java

```java
package com.mongodb;

import java.util.Map;
import java.util.function.Supplier;

import com.mongodb.kafka.connect.util.custom.credentials.CustomCredentialProvider;

import com.amazonaws.auth.AWSCredentialsProvider;
import com.amazonaws.auth.DefaultAWSCredentialsProviderChain;
import com.amazonaws.services.securitytoken.AWSSecurityTokenService;
import com.amazonaws.services.securitytoken.AWSSecurityTokenServiceAsyncClientBuilder;
import com.amazonaws.services.securitytoken.model.AssumeRoleRequest;
import com.amazonaws.services.securitytoken.model.AssumeRoleResult;
import com.amazonaws.services.securitytoken.model.Credentials;
import com.amazonaws.util.StringUtils;

public class SampleAssumeRoleCredential implements CustomCredentialProvider {

  public SampleAssumeRoleCredential() {}
  @Override
  public MongoCredential getCustomCredential(Map<?, ?> map) {
    AWSCredentialsProvider provider = new DefaultAWSCredentialsProviderChain();
    Supplier<AwsCredential> awsFreshCredentialSupplier = () -> {
      AWSSecurityTokenService stsClient = AWSSecurityTokenServiceAsyncClientBuilder.standard()
          .withCredentials(provider)
          .withRegion("us-east-1")
          .build();
      AssumeRoleRequest assumeRoleRequest = new AssumeRoleRequest().withDurationSeconds(3600)
          .withRoleArn((String)map.get("mongodbaws.auth.mechanism.roleArn"))
          .withRoleSessionName("Test_Session");
      AssumeRoleResult assumeRoleResult = stsClient.assumeRole(assumeRoleRequest);
      Credentials creds = assumeRoleResult.getCredentials();
      // Add your code to fetch new credentials
      return new AwsCredential(creds.getAccessKeyId(), creds.getSecretAccessKey(), creds.getSessionToken());
    };
    return MongoCredential.createAwsCredential(null, null)
        .withMechanismProperty(MongoCredential.AWS_CREDENTIAL_PROVIDER_KEY, awsFreshCredentialSupplier);
  }

  @Override
  public void validate(Map<?, ?> map) {
    String roleArn = (String) map.get("mongodbaws.auth.mechanism.roleArn");
    if (StringUtils.isNullOrEmpty(roleArn)) {
      throw new RuntimeException("Invalid value set for customProperty");
    }
  }

  @Override
  public void init(Map<?, ?> map) {

  }
}
```

Example pom.xml file

```xml
<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"
         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
    <modelVersion>4.0.0</modelVersion>
    <groupId>com.mongodb</groupId>
    <artifactId>SampleAssumeRoleCredential</artifactId>
    <version>1.0-SNAPSHOT</version>
    <build>
        <plugins>
            <plugin>
                <groupId>org.apache.maven.plugins</groupId>
                <artifactId>maven-shade-plugin</artifactId>
                <version>3.5.3</version>
                <configuration>
                    
                </configuration>
                <executions>
                    <execution>
                        <phase>package</phase>
                        <goals>
                            <goal>shade</goal>
                        </goals>
                    </execution>
                </executions>
            </plugin>
        </plugins>
    </build>

    <dependencies>
        
        
        <dependency>
            <groupId>org.mongodb</groupId>
            <artifactId>mongodb-driver-sync</artifactId>
            <version>5.1.0</version>
        </dependency>
        
        <dependency>
            <groupId>com.amazonaws</groupId>
            <artifactId>aws-java-sdk</artifactId>
            <version>1.12.723</version>
        </dependency>

        
        <dependency>
            <groupId>org.slf4j</groupId>
            <artifactId>slf4j-jdk14</artifactId>
            <version>1.7.28</version>
        </dependency>

        <dependency>
            <groupId>kafka-connect</groupId>
            <artifactId>kafka-connect</artifactId>
            <scope>system</scope>
            <version>1.13.0</version>
        </dependency>
    </dependencies>

    <properties>
        <maven.compiler.source>17</maven.compiler.source>
        <maven.compiler.target>17</maven.compiler.target>
        <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
    </properties>

</project>
```

Note: We need to reference MongoDB Connector for Apache Kafka version 1.13 or later. In this example, the connector was downloaded from the Kafka Connector GitHub repository and stored in the local Maven repository. 

```bash
mvn install:install-file -Dfile=/Users/robert.walters/Downloads/mongodb-kafka-connect-mongodb-1.13.0/lib/mongo-kafka-connect-1.13.0-confluent.jar -DgroupId=kafka-connect -DartifactId=kafka-connector -Dversion=1.13.0  -Dpackaging=jar\n
```

Compile and place this SampleAssumeRole.jar file in the Kafka libs directory in your Kafka Connect deployment. You are now ready to configure your MongoDB Connector for Apache Kafka to use this custom authentication provider.

### Configure the MongoDB Connector for Apache Kafka

To support the custom provider in version 1.13, the following configuration options were added to the connector:

| Configuration Parameter                   | Description                                                  |
| ----------------------------------------- | ------------------------------------------------------------ |
| mongo.custom.auth.mechanism.enable        | Determines if the connector should use the provider class; default is false |
| mongo.custom.auth.mechanism.providerClass | Java class entry point for the custom provider               |
| mongodbaws.auth.mechanism.roleArn         | Role that has IAM access to the database                     |

In this example, the custom authentication component will use AWS IAM AssumeRole to authenticate with MongoDB Atlas. Here is an example of this configuration connecting to a MongoDB Atlas cluster.

```json
{
    "name": "mongo-tutorial-sink",
    "config": {
        "connector.class": "com.mongodb.kafka.connect.MongoSinkConnector",
        "topics": "Stocks2",
        "connection.uri": "<MongoDB Connection String>",
        "key.converter": "org.apache.kafka.connect.storage.StringConverter",
        "value.converter": "org.apache.kafka.connect.json.JsonConverter",
        "value.converter.schemas.enable": false,
        "database": "Stocks",
        "collection": "StockData",
        "mongo.custom.auth.mechanism.enable":"true",
        "mongo.custom.auth.mechanism.providerClass":"com.mongodb.SampleAssumeRoleCredential",
        "mongodbaws.auth.mechanism.roleArn":"<AWS IAM ARN>"      
    }
}
```

Notice that the roleArn is the ARN (Amazon Resource Name) of the AWS IAM role that is used for Atlas users, KafkaBlogAtlasRole. At runtime, since Kafka Connect is running under the AWS IAM role of the EC2 instance, KafkaConnectBlogRole, and this role has a policy that allows AssumeRole on the KafkaBlogAtlasRole, the service can connect to Atlas without storing any credentials!

You are now ready to send data to Kafka and watch it arrive in a MongoDB collection using the MongoDB Connector for Apache Kafka configured with AWS IAM authentication to MongoDB Atlas.

## Summary

In this article, we discussed the requirements needed to support a custom authentication provider such as AWS IAM authentication. Version 1.13 of the Kafka Connector introduced new parameters that define which class to use within the custom provider JAR file. This JAR file includes the logic for the custom authentication. 

Note that while adding AWS IAM authentication capability natively to the Kafka Connector is possible (it is just software, after all), doing so would not only add a version dependency on the AWS IAM SDK, but it would also significantly increase the size of the connector, making it impractical for all other users who do not wish to use this authentication method. For these reasons, writing the custom credential provider is the solution for integrating the Kafka Connector with authentication methods such as AWS IAM.

Read the [MongoDB Connector for Apache Kafka documentation](https://www.mongodb.com/docs/kafka-connector/current/) to learn more. Questions? Join us in the [MongoDB Developer Community](https://www.mongodb.com/community/forums/).


  [1]: https://images.contentstack.io/v3/assets/blt39790b633ee0d5a7/bltb44f9cef8df8f2b2/6682f292b1132488c03a422b/image1.png

To connect to an Atlas cluster, users authenticate with AWS IAM authentication. Learn how to set up the MongoDB Connector for Apache Kafka for authentication.

Using AWS IAM Authentication with the MongoDB Connector for Apache Kafka

Rutuja Rajwade

Product Marketing Manager

Zuhair Ahmed

Sr. Product Manager

An event-driven architecture is a highly effective framework for creating applications that are responsive, scalable, and maintainable with loosely coupled components. This approach has become essential in developing applications that can rapidly respond to customer demands and dynamic market conditions.

For example, in healthcare, event-driven systems process data emitted by medical devices (like ECGs and blood glucose monitors) to provide real-time insights and alerts to healthcare providers, enabling timely interventions. Financial institutions utilize event-driven architectures to process and analyze transactions in real time. This allows for immediate fraud detection, risk assessment, and dynamic pricing based on current market conditions.

In this tutorial, we cover how to get started with MongoDB Atlas for near real-time data processing with HashiCorp Terraform. By integrating [MongoDB Atlas Stream Processing](https://www.mongodb.com/products/platform/atlas-stream-processing) with the [HashiCorp Terraform MongoDB Atlas Provider](https://registry.terraform.io/providers/mongodb/mongodbatlas/latest), developers can efficiently manage real-time data workflows and automate resource provisioning.

Let’s get started!

## Prerequisites

- Create a [MongoDB Atlas](https://www.mongodb.com/cloud/atlas/register) account. (More configuration details will be shown below.)
- Install [HashiCorp Terraform](https://developer.hashicorp.com/terraform/tutorials/aws-get-started/install-cli) on your terminal or sign up for a free [Terraform Cloud](https://app.terraform.io/public/signup/account) account.
- Create [MongoDB Atlas programmatic API keys](https://www.mongodb.com/docs/atlas/configure-api-access/#grant-programmatic-access-to-service) and [associate them with Terraform](https://www.mongodb.com/developer/products/atlas/deploy-mongodb-atlas-terraform-aws/) (further details below).
- Choose your IDE. For this tutorial, we will be using [VS Code](https://code.visualstudio.com/).

## Deploy foundational Atlas Stream Processing resources using the Terraform provider 

There are three main components to consider with Atlas Stream Processing: 

- [Stream Processing Instance](https://www.mongodb.com/docs/atlas/atlas-sp/overview/#mongodb-expression-exp.Stream-Processing-Instance) (SPI) — This acts as the primary environment where your streaming data processes are executed. It's responsible for managing and executing the stream processing tasks, maintaining performance, and ensuring data integrity throughout the streaming lifecycle.
- [Connection registry](https://www.mongodb.com/docs/atlas/atlas-sp/overview/#mongodb-expression-exp.Connection-Registry) — This serves as the conduit between your data sources/sinks and the processing instance. It's tasked with the secure and efficient transport of streaming data, facilitating continuous data ingestion from various sources into the SPI.
- [Stream processor](https://www.mongodb.com/docs/atlas/atlas-sp/overview/#mongodb-expression-exp.Stream-Processor) — This is where the business logic for data transformation and analysis is defined and executed. Using MongoDB's expressive querying and aggregation capabilities, this component processes incoming data streams continuously, enabling immediate data manipulation and enrichment before forwarding it for storage or further processing.

At this time, the Terraform Atlas Provider supports management of the SPI ([mongodbatlas_stream_instance](https://registry.terraform.io/providers/mongodb/mongodbatlas/latest/docs/resources/stream_instance)) and connections in the Connection Registry ([mongodbatlas_stream_connection](https://registry.terraform.io/providers/mongodb/mongodbatlas/latest/docs/resources/stream_connection)), with programmatic support for stream processors coming in the future. 

First, we need to deploy basic Atlas resources to get started. As part of this tutorial, this includes an Atlas project, M10 dedicated Atlas cluster, database user, and IP access list entry. 

In addition, we will be deploying one Stream Processing Instance and two connections in the connection registry. See our Atlas Admin [Open API Specification documentation](https://www.mongodb.com/docs/atlas/reference/api-resources-spec/v2/#tag/Streams/operation/createStreamInstance) for a full list of supported cloud providers, regions, and all other parameter inputs for these resources:

- In this example, our Stream Processing Instance will be hosted on AWS us-east-1 using an SP30 tier. More information on Stream Processing tiers is available in the [documentation](https://www.mongodb.com/docs/atlas/billing/stream-processing-costs/).
- For the connection registry, we will point our data source to type [Sample](https://www.mongodb.com/docs/atlas/atlas-stream-processing/tutorial/#add-the-sample-stream-connection-to-the-connection-registry.) (solar data that we provide as a sample source to use when getting started) which is great for testing purposes. (Type Kafka is also supported.)
- For a connection to the built-in sample_stream_solar, provide a configuration file with the following syntax:

```
{
  "name": "sample_stream_solar",
  "type": "Sample"
}
```

- Our data sink will point to the M10 dedicated Atlas cluster we mentioned earlier. 
- In the sample Terraform script provided below, replace the “org_id” field, and enter your own organization’s ID and your unique username and password in the “username” and “password” fields.
- Ensure that the IP address added to the script matches the one in Terraform. If you are unsure of the IP address that you are running Terraform on (and you are performing this step from that machine), simply click “Use Current IP Address and Save”. Another option is to open up your IP access list to all, but this comes with significant potential risks. To do this, you can add the following two CIDRs: 0.0.0.0/1 and 128.0.0.0/1. These entries will open your IP access list to, at most, 4,294,967,296 (or 2^32) IPv4 addresses and should be used with caution.

See the below Terraform script as part of our `main.tf` file.

```
terraform {
  required_providers {
    mongodbatlas = {
      source = "mongodb/mongodbatlas"
    }
  }
  required_version = ">= 0.13"
}

resource "mongodbatlas_project" "testProject" {
  name   = "testProject16"
  org_id = "[YOUR ORG_ID]"
}

resource "mongodbatlas_project_ip_access_list" "testIPAccessList" {
  project_id = mongodbatlas_project.testProject.id
  ip_address = "[YOUR IP ADDRESS]"
}

resource "mongodbatlas_database_user" "testUser" {
  project_id         = mongodbatlas_project.testProject.id
  username           = "[YOUR USERNAME]"
  password           = "[YOUR PASSWORD]"
  auth_database_name = "admin"
  roles {
    role_name     = "readWrite"
    database_name = "dbforApp"
  }
}

resource "mongodbatlas_advanced_cluster" "testCluster" {
  project_id   = mongodbatlas_project.testProject.id
  name         = "testCluster"
  cluster_type = "REPLICASET"
  replication_specs {
    region_configs {
      electable_specs {
        instance_size = "M10"
        node_count    = 3
      }
      provider_name = "AWS"
      priority      = 7
      region_name   = "US_EAST_1"
    }
  }
}

resource "mongodbatlas_stream_instance" "myFirstStreamInstance" {
  project_id    = mongodbatlas_project.testProject.id
  instance_name = "myFirstStreamInstance"
  data_process_region = {
    region         = "VIRGINIA_USA"
    cloud_provider = "AWS"
  }
  stream_config = {
    tier = "SP30"
  }
}

resource "mongodbatlas_stream_connection" "dataSource" {
    project_id = mongodbatlas_project.testProject.id
    instance_name = mongodbatlas_stream_instance.myFirstStreamInstance.instance_name
    connection_name = "sample_stream_solar"
    type = "Sample"
}

resource "mongodbatlas_stream_connection" "dataSink" {
    project_id = mongodbatlas_project.testProject.id
    instance_name = mongodbatlas_stream_instance.myFirstStreamInstance.instance_name
    connection_name = "dataSink"
    type = "Cluster"
    cluster_name = mongodbatlas_advanced_cluster.testCluster.name
    db_role_to_execute = {
      role = "atlasAdmin"
      type = "BUILT_IN"
    }
}
```

Note: Before deploying, be sure to store your MongoDB Atlas programmatic API keys created as part of prerequisites as [Environment Variables](https://developer.hashicorp.com/terraform/cli/config/environment-variables). To deploy, you can use the below commands from the terminal: 

```
terraform init 
terraform plan
terraform apply 
```

If your deployment was successful, you should be greeted with “Apply complete!” 

![Screenshot of completed deployment][1]

To confirm, you should be able to see your newly created SPI in the Atlas UI. (To navigate from the home screen, click on “Stream Processing” on the left-hand menu.) 

![Screenshot of Atlas UI after creation of stream processing instance][2]

By going into the Connection Registry, we can also verify that our connections were successfully created. 

![Screenshot of Atlas UI showcasing connection registry][3]

## Get connection string and create stream processor 

Next, you’ll need to retrieve the Atlas Stream Processing connection string and create your first stream processor.

While still in the Atlas UI, click “Connect” and select the tool to access your data through MongoDB Shell or MongoDB for VS Code. 

![Screenshot of Atlas UI showcasing connection to MongoDB Shell and MongoDB VSCode][4]

This will generate your connection string which you can use to connect to your Stream Processing Instance. 

At this point, your first stream processors can be created to define how your data will be processed in your SPI. As mentioned earlier, there are currently no Terraform resources as of yet to provide this configuration. To learn more about how stream processors can be configured, refer to [Manage Stream Processors](https://www.mongodb.com/docs/atlas/atlas-sp/manage-stream-processor/#manage-stream-processors).

For more detailed information, also see our [Atlas Stream Processing example](https://github.com/mongodb/terraform-provider-mongodbatlas/blob/master/examples/mongodbatlas_stream_instance/atlas-streams-user-journey.md) in our Terraform repo. 

## All done 

Congratulations! You have everything that you need now to run your first [Atlas Stream Processing](https://www.mongodb.com/docs/atlas/atlas-sp/overview/)-based workflow. 

With the above steps, teams can utilize Atlas Stream Processing capabilities within the Terraform MongoDB Atlas Provider. Whether you are developing complex event processing systems, real-time analytics, or interactive user engagement features, Atlas Stream Processing integrated with Terraform provides a robust infrastructure to support scalable and resilient applications. 

The HashiCorp Terraform Atlas Provider is open-sourced under the [Mozilla Public License v2.0](https://github.com/mongodb/terraform-provider-mongodbatlas?tab=MPL-2.0-1-ov-file) and we welcome community contributions. To learn more, see our [contributing guidelines](https://github.com/mongodb/terraform-provider-mongodbatlas/blob/master/CONTRIBUTING.md). 

The fastest way to get started is to create a MongoDB Atlas account from the [AWS Marketplace](https://aws.amazon.com/marketplace/seller-profile?id=c9032c7b-70dd-459f-834f-c1e23cf3d092). To learn more about the Terraform Provider, check out our [documentation](https://registry.terraform.io/providers/mongodb/mongodbatlas/latest/docs), [tutorials](https://www.mongodb.com/developer/search/?s=Terraform&sortMode=0), [solution brief](https://www.mongodb.com/collateral/build-resilient-applications-with-mongodb-atlas-and-hashicorp-terraform), or [get started today](https://cloud.mongodb.com).

Go build with MongoDB Atlas and the HashiCorp Terraform Atlas Provider today! 


  [1]: https://images.contentstack.io/v3/assets/blt39790b633ee0d5a7/bltd8b6fd1c026e0dfa/664bc07e1983ea56f5ac2899/1.png
  [2]: https://images.contentstack.io/v3/assets/blt39790b633ee0d5a7/bltc57e0777ae3b69f3/664bc0907853391c269e3c1f/2.png
  [3]: https://images.contentstack.io/v3/assets/blt39790b633ee0d5a7/bltb73dbc3b43a2e8a0/664bc09f941bc85c3f9df7b1/3.png
  [4]: https://images.contentstack.io/v3/assets/blt39790b633ee0d5a7/blt6e9c931aa8a32787/664bc0ae55b39716c16049d8/4.png

How to Get Started with MongoDB Atlas Stream Processing and the HashiCorp Terraform MongoDB Atlas Provider

Champion

Arek Borucki

Principal SRE Database Engineer

Learn how to deploy MongoDB across multiple Kubernetes clusters using the operator and the MongoDBMulti CRD.

Deploying the MongoDB Enterprise Kubernetes Operator on Google Cloud

Mastering MongoDB Ops Manager on Kubernetes

Deploying MongoDB Across Multiple Kubernetes Clusters With MongoDBMulti

Deploying MongoDB Across Multiple Kubernetes Clusters

Juan Soto

Senior Consulting Engineer

With today’s need of flexible event-driven architectures, companies across the globe choose best of breed technologies like MongoDB and Apache Kafka to help solve these challenges.  While these two complementary technologies provide the power and flexibility to solve these large scale challenges, performance has always been at the forefront of concerns. In this blog, we will cover how to measure performance of the MongoDB Connector for Apache Kafka in both a source and sink configuration. 

## Measuring Sink Performance

Recall that the MongoDB sink connector writes data from a Kafka topic into MongoDB. Writes by default use the [ReplaceOneModel](https://mongodb.github.io/mongo-java-driver/4.2/apidocs/mongodb-driver-core/com/mongodb/client/model/ReplaceOneModel.html) where the data is either updated if it's present on the destination cluster or created as a new document if it is not present. You are not limited to this upsert behavior. In fact, you can change the sink to perform deletes or inserts only. These write behaviors are defined by the [Write Model Strategy](https://docs.mongodb.com/kafka-connector/current/sink-connector/fundamentals/write-strategies/) setting in the sink configuration.

To determine the performance of the sink connector, we need a timestamp of when the document was written to MongoDB. Currently, the only write model strategy that writes a timestamp field on behalf of the user is [UpdateOneTimestampsStrategy](http://UpdateOneTimestampsStrategy) and [UpdateOneBusinessKeyTimestampStrategy](http://UpdateOneTimestampsStrategy). These two write models insert a new field named **_insertedTS**, which can be used to query the lag between Kafka and MongoDB.

In this example, we’ll use [MongoDB Atlas](https://www.mongodb.com/cloud/atlas). MongoDB Atlas is a public cloud MongoDB data platform providing out-of-the-box capabilities such as [MongoDB Charts](https://docs.mongodb.com/charts/saas/), a tool to create visual representations of your MongoDB data. If you wish to follow along, you can create a [free forever tier](https://www.mongodb.com/cloud/atlas/register).

### Generate Sample Data

We will generate sample data using the [datagen](https://www.confluent.io/blog/easy-ways-generate-test-data-kafka/) Kafka Connector provided by Confluent.  [Datagen](https://www.confluent.io/hub/confluentinc/kafka-connect-datagen) is a convenient way of creating test data in the Kafka ecosystem. There are a few quickstart schema specifications bundled with this connector. We will use a quickstart called **users**.

```
curl -X POST -H "Content-Type: application/json" --data '
  {"name": "datagen-users",
   "config": { "connector.class":    "io.confluent.kafka.connect.datagen.DatagenConnector",
   "kafka.topic": "topic333",
   "quickstart": "users",
   "key.converter": "org.apache.kafka.connect.storage.StringConverter",
   "value.converter": "org.apache.kafka.connect.json.JsonConverter",
   "value.converter.schemas.enable": "false",
   "max.interval": 50,
   "iterations": 5000,
   "tasks.max": "2"
}}' http://localhost:8083/connectors -w "\n"

```

### Configure Sink Connector

Now that the data is generated and written to the Kafka topic, “topic333,” let’s create our MongoDB sink connector to write this topic data into MongoDB Atlas. As stated earlier, we will add a field **_insertedTS** for use in calculating the lag between the message timestamp and this value. To perform the insert, let’s use the **UpdateOneTimestampsStrategy** write mode strategy.

```
curl -X POST -H "Content-Type: application/json" --data '
{"name": "kafkametadata3",
  "config": {
    "connector.class": "com.mongodb.kafka.connect.MongoSinkConnector",
    "topics": "topic333",
    "connection.uri": "MONGODB CONNECTION STRING GOES HERE",
    "writemodel.strategy": "com.mongodb.kafka.connect.sink.writemodel.strategy.UpdateOneTimestampsStrategy",
    "database": "kafka",
    "collection": "datagen",
    "errors.log.include.messages": true,
    "errors.deadletterqueue.context.headers.enable": true,
    "key.converter": "org.apache.kafka.connect.storage.StringConverter",
    "value.converter": "org.apache.kafka.connect.json.JsonConverter",
    "document.id.strategy": "com.mongodb.kafka.connect.sink.processor.id.strategy.KafkaMetaDataStrategy",
    "tasks.max": 2,
    "value.converter.schemas.enable":false,
    "transforms": "InsertField",
    "transforms.InsertField.type": "org.apache.kafka.connect.transforms.InsertField$Value",
    "transforms.InsertField.offset.field": "offsetColumn",
    "transforms": "InsertField",
    "transforms.InsertField.type": "org.apache.kafka.connect.transforms.InsertField$Value",
    "transforms.InsertField.timestamp.field": "timestampColumn"
}}' http://localhost:8083/connectors -w "\n"

```

Note: The field **_insertedTS** is populated with the time value of the Kafka connect server.

### Viewing Results with MongoDB Charts

Take a look at the MongoDB Atlas collection “datagen” and familiarize yourself with the added fields.

![Figure 1: Datagen collection as seen in MongoDB Atlas Collections page](https://mongodb-devhub-cms.s3.us-west-1.amazonaws.com/mongodb_atlas_web_page_00e3cd3149.png)

In this blog, we will use [MongoDB Charts](https://www.mongodb.com/products/charts) to display a performance graph. To make it easy to build the chart, we will create a view.

```
use kafka
db.createView("SinkView","datagen",
[
	{
		"$sort" : {
			"_insertedTS" : 1,
			"timestampColumn" : 1
		}
	},
	{
		"$project" : {
			"_insertedTS" : 1,
			"timestampColumn" : 1,
			"_id" : 0
		}
	},
	{
		"$addFields" : {
			"diff" : {
				"$subtract" : [
					"$_insertedTS",
					{
						"$convert" : {
							"input" : "$timestampColumn",
							"to" : "date"
						}
					}
				]
			}
		}
	}
])

```

To create a chart, click on the Charts tab in MongoDB Atlas:

![](https://mongodb-devhub-cms.s3.us-west-1.amazonaws.com/charts_tab_mongodb_atlas_2ad1d61e2c.png)

Click on Datasources and “Add Data Source.”  The dialog will show the view that was created.

![](https://mongodb-devhub-cms.s3.us-west-1.amazonaws.com/add_data_source_page_1075b95bba.png)

Select the SinkView and click Finish.

Download the [MongoDB Sink performance Chart](https://gist.github.com/RWaltersMA/555b5f17791ecb58e6e683c54bafd381) from Gist. ​​

```
curl https://gist.githubusercontent.com/RWaltersMA/555b5f17791ecb58e6e683c54bafd381/raw/748301bcb7ae725af4051d40b2e17a8882ef2631/sink-chart-performance.charts -o sink-performance.charts

```

Choose **Import Dashbaord** from the Add Dashboard dropdown and select the downloaded file.

![](https://mongodb-devhub-cms.s3.us-west-1.amazonaws.com/import_dashboard_button_468f3e569d.png)

Load the sink-perfromance.chart file.

![](https://mongodb-devhub-cms.s3.us-west-1.amazonaws.com/import_dashboard_webpage_534e6fe5f8.png)

Select the kafka.SinkView as the data source at the destination then click Save.

Now the KafkaPerformance chart is ready to view. When you click on the chart, you will see something like the following: 

![](https://mongodb-devhub-cms.s3.us-west-1.amazonaws.com/mongodb_chart_showing_metrics_55d64aba1a.png)

This chart shows statistics on the differences between the timestamp in the Kafka topic and Kafka connector. In the above example, the maximum time delta is approximately one second (997ms) from inserting 40,000 documents.

## Measuring Source Performance

To measure the source, we will take a different approach using KSQL to create a stream of the clusterTime timestamp from the MongoDB change stream and the time the row was written in the Kafka topic. From here, we can push this data into a MongoDB sink and display the results in a MongoDB Chart. 

### Configure Source Connector

The first step will be to create the MongoDB Source connector that will be used to push data onto the Kafka topic.

```
curl -X POST -H "Content-Type: application/json" --data '
{"name": "mongo-source-perf",
  "config": {
  "connector.class": "com.mongodb.kafka.connect.MongoSourceConnector",
  "errors.log.enable": "true",
  "errors.log.include.messages": "true",
  "connection.uri": "mongodb+srv://MONGODB CONNECTION STRING HERE",
  "database": "kafka",
  "collection": "source-perf-test",
  "mongo.errors.log.enable": "true",
  "topic.prefix":"mdb",
  "output.json.formatter" : "com.mongodb.kafka.connect.source.json.formatter.SimplifiedJson",
    "output.format.value":"schema",
    "output.schema.infer.value":true,
    "output.format.key":"json",
    "publish.full.document.only": "false",
  "change.stream.full.document": "updateLookup"
}}' http://localhost:8083/connectors -w "\n"
```

### Generate Sample Data

There are many ways to generate sample data on MongoDB. In this blog post, we will use the [doc-gen](https://hub.docker.com/repository/docker/robwma/doc-gen/general) tool (Github repo) to quickly create sample documents based upon the user’s schema, which is defined as follows:

```
{
   "_id" : ObjectId("59b99db4cfa9a34dcd7885b6"),
   "name" : "Ned Stark",
   "email" : "sean_bean@gameofthron.es",
   "password" : "$2b$12$UREFwsRUoyF0CRqGNK0LzO0HM/jLhgUCNNIJ9RJAqMUQ74crlJ1Vu"
}

```

To generate data in your MongoDB cluster, issue the following:

```
docker run robwma/doc-gen:1.0 python doc-gen.py -s '{"name":"string","email":"string","password":"string"}' -c "MONGODB CONNECTION STRING GOES HERE" -t 1000 -db "kafka" -col "source-perf-test"
```

### Create KSQL Queries

Launch KSQL and create a stream of the clusterTime within the message. 

Note: If you do not have KSQL, you can run it as part of the Confluent Platform all in Docker using the following instructions.

If using Control Center, click ksQLDB, click Editor, and then paste in the following KSQL:

```
CREATE STREAM stats (
   clusterTime BIGINT
 ) WITH (
   KAFKA_TOPIC='kafka.source-perf-test',
       VALUE_FORMAT='AVRO'
 );

```

The only information that we need from the message is the clusterTime. This value is provided within the [change stream](https://docs.mongodb.com/kafka-connector/current/kafka-source/#change-stream-event-document-format) event. For reference, this is a sample event from change streams.

```
{
  _id: { <BSON Object> },
  "operationType": "<operation>",
  "fullDocument": { <document> },
  "ns": {
    "db": <database>,
    "coll": <collection>
  },
  "to": {
    "db": <database>,
    "coll": <collection>
  },
  "documentKey": {
    _id: <value>
  },
  "updateDescription": {
    "updatedFields": { <document> },
    "removedFields": [ <field>, ... ]
  },
  "clusterTime": <Timestamp>,
  "txnNumber": <NumberLong>,
  "lsid": {
    "id": <UUID>,
    "uid": <BinData>
  }
}

```

**Step 3**

Next, we will create a ksql stream that calculates the difference between the cluster time (time when it was created on MongoDB) and the time where it was inserted on the broker. 

```
CREATE STREAM STATS2 AS
  select ROWTIME - CLUSTERTIME as diff, 1 AS ROW from STATS EMIT CHANGES;

```

As stated previously, this diff value may not be completely accurate if the clocks on Kafka and MongoDB are different. 

**Step 4**

To see how the values change over time, we can use a window function and write the results to a table which can then be written into MongoDB via a sink connector.

```
SET 'ksql.suppress.enabled' = 'true';

CREATE TABLE STATSWINDOW2 AS
  SELECT  AVG( DIFF ) AS AVG, MAX(DIFF) AS MAX,  count(*) AS COUNT,  ROW  FROM STATS2
 WINDOW TUMBLING (SIZE 10 SECONDS)
 GROUP BY ROW
 EMIT FINAL;

```

Windowing lets you control how to group records that have the same key for stateful operations, such as aggregations or joins into so-called windows. There are three ways to define time windows in ksqlDB: hopping windows, tumbling windows, and session windows. In this example, we will use tumbling as it is a fixed-duration, non-overlapping, and gap-less window.

![](https://mongodb-devhub-cms.s3.us-west-1.amazonaws.com/tumbling_window_sequence_03d52d2c68.png)

### Configure Sink Connector

The final step is to create a sink connector to insert all this aggregate data on MongoDB.

```
curl -X POST -H "Content-Type: application/json" --data '
{
 "name": "MongoSource-SinkPerf",
 "config": {
   "connector.class": "com.mongodb.kafka.connect.MongoSinkConnector",
   "tasks.max": "1",
   "errors.log.enable": true,
   "errors.log.include.messages": true,
   "topics": "STATSWINDOW2",
   "errors.deadletterqueue.context.headers.enable": true,
   "connection.uri": "MONGODB CONNECTION STRING GOES HERE",
   "database": "kafka",
   "collection": "sourceStats",
   "mongo.errors.log.enable": true,
   "transforms": "InsertField",
   "transforms.InsertField.type": "org.apache.kafka.connect.transforms.InsertField$Value",
   "transforms.InsertField.timestamp.field": "timestampColumn"
}}' http://localhost:8083/connectors -w "\n"

```

### Viewing Results with MongoDB Charts

Download the [MongoDB Source performance Chart](https://gist.githubusercontent.com/RWaltersMA/011f1473cf937badc61b752a6ab769d4/raw/bc180b9c2db533536e6c65f34c30b2d2145872f9/mongodb-source-performance.chart) from Gist. ​​

```
curl https://gist.githubusercontent.com/RWaltersMA/011f1473cf937badc61b752a6ab769d4/raw/bc180b9c2db533536e6c65f34c30b2d2145872f9/mongodb-source-performance.chart -o source-performance.charts

```

Choose **Import Dashboard** from the Add Dashboard dropdown and select the downloaded file.

![](https://mongodb-devhub-cms.s3.us-west-1.amazonaws.com/import_dashboard_button_468f3e569d.png)

You will need to create a Datasource to the new sink collection, “kafka.sourceStats.”

![](https://mongodb-devhub-cms.s3.us-west-1.amazonaws.com/sourcestats_collection_7b8554f5a0.png)

Click on the Kafka Performance Source chart to view the statistics.

![](https://mongodb-devhub-cms.s3.us-west-1.amazonaws.com/mongodb_chart_showing_various_metrics_5bc97176db.png)

In the above example, you can see the 10-second sliding window performance statistics for 1.5M documents. The average difference was 252s, with the maximum difference being 480s.  Note that some of this delta could be differences in clocks between MongoDB and Kafka. While not taking these numbers as absolute, simply using this technique is good enough to determine trends and if the performance is getting worse or better.

If you have any opinions on features or functionality enhancements that you would like to see with respect to monitoring performance or monitoring the MongoDB Connector for Apache Kafka in general, please add a comment to [KAFKA-64](https://jira.mongodb.org/browse/KAFKA-64). 

Have any questions? Check out our [Connectors and Integrations](https://www.mongodb.com/community/forums/c/data/connectors-integrations/48) MongoDB community forum.

Learn about measuring the performance of the MongoDB Connector for Apache Kafka in both a source and sink configuration.

Measuring MongoDB Kafka Connector Performance

This article is part of a three-parts series on deploying MongoDB across multiple Kubernetes clusters using the operators.

-   Deploying the MongoDB Enterprise Kubernetes Operator on Google Cloud

-   [Mastering MongoDB Ops Manager](https://www.mongodb.com/developer/products/connectors/mastering-ops-manager)

-   [Deploying MongoDB Across Multiple Kubernetes Clusters With MongoDBMulti](https://www.mongodb.com/developer/products/connectors/deploying-across-multiple-kubernetes-clusters)

Deploying and managing MongoDB on Kubernetes can be a daunting task. It requires creating and configuring various Kubernetes resources, such as persistent volumes, services, and deployments, which can be time-consuming and require a deep understanding of both Kubernetes and MongoDB products. Furthermore, tasks such as scaling, backups, and upgrades must be handled manually, which can be complex and error-prone. This can impact the reliability and availability of your MongoDB deployment and may require frequent manual intervention to keep it running smoothly. Additionally, it can be hard to ensure that your MongoDB deployment is running in the desired state and is able to recover automatically from failures.

Fortunately, MongoDB offers operators, which are software extensions to the Kubernetes API that use custom resources to manage applications and their components. The MongoDB Operator translates human knowledge of creating a MongoDB instance into a scalable, repeatable, and standardized method, and leverages Kubernetes features to operate MongoDB for you. This makes it easier to deploy and manage MongoDB on Kubernetes, providing advanced features and functionality for running MongoDB in cloud-native environments.

There are three main Kubernetes operators available for deploying and managing MongoDB smoothly and efficiently in Kubernetes environments:

-   [The MongoDB Community Kubernetes Operator](https://github.com/mongodb/mongodb-kubernetes-operator) is an open-source operator that is available for free and can be used to deploy and manage MongoDB Replica Set on any Kubernetes cluster. It provides basic functionality for deploying and managing MongoDB but does not include some of the more advanced features available in the Enterprise and Atlas operators.

-   [The MongoDB Enterprise Kubernetes Operator ](https://github.com/mongodb/mongodb-enterprise-kubernetes)is a commercial Kubernetes operator included with the MongoDB Enterprise subscription. It allows you to easily deploy and manage any type of MongoDB deployment (standalone, replica set, sharded cluster) on Kubernetes, providing advanced features and functionality for deploying and managing MongoDB in cloud-native environments.

-   [The MongoDB Atlas Kubernetes Operator](https://github.com/mongodb/mongodb-atlas-kubernetes) is an operator that is available as part of the Atlas service. It allows you to quickly deploy and manage MongoDB on the Atlas cloud platform, providing features such as automatic provisioning and scaling of MongoDB clusters, integration with Atlas features and services, and automatic backups and restores. You can learn more about this operator in our blog post on [application deployment in Kubernetes](https://www.mongodb.com/developer/products/atlas/kubernetes-operator-application-deployment/).

This article will focus on the Enterprise Operator. [The MongoDB Enterprise Kubernetes Operator](https://www.mongodb.com/docs/kubernetes-operator/master/) seamlessly integrates with other MongoDB Enterprise features and services, such as MongoDB Ops Manager (which can also run on Kubernetes) and MongoDB Cloud Manager. This allows you to easily monitor, back up, upgrade, and manage your MongoDB deployments from a single, centralized location, and provides access to a range of tools and services for managing, securing, and optimizing your deployment.

## MongoDB Enterprise Kubernetes Operator

The MongoDB Enterprise [Kubernetes Operator](https://kubernetes.io/docs/concepts/extend-kubernetes/operator/) automates the process of creating and managing MongoDB instances in a scalable, repeatable, and standardized manner. It uses the Kubernetes API and tools to handle the lifecycle events of a MongoDB cluster, including provisioning storage and computing resources, configuring network connections, setting up users, and making changes to these settings as needed. This helps to ease the burden of manually configuring and managing stateful applications, such as databases, within the Kubernetes environment.

![The Kubernetes Operator manages the typical lifecycle events for a MongoDB cluster: provisioning storage and computing power, configuring network connections, setting up users, and changing these settings as needed](https://mongodb-devhub-cms.s3.us-west-1.amazonaws.com/kubernetes_operator_bakedsvg_1c1a6d427b.svg)

## Kubernetes Custom Resource Definitions

[Kubernetes CRDs](https://kubernetes.io/docs/tasks/extend-kubernetes/custom-resources/custom-resource-definitions/) (Custom Resource Definitions) is a feature in Kubernetes that allows users to create and manage custom resources in their Kubernetes clusters. [Custom resources](https://kubernetes.io/docs/concepts/extend-kubernetes/api-extension/custom-resources/) are extensions of the Kubernetes API that allow users to define their own object types and associated behaviors. With CRDs, you can create custom resources that behave like built-in Kubernetes resources, such as StatefulSets, Deployments, Pods, and Services, and manage them using the same tools and interfaces. This allows you to extend the functionality of Kubernetes and tailor it to their specific needs and requirements.

The MongoDB Enterprise Operator currently provides the following custom resources for deploying MongoDB on Kubernetes:

-   [MongoDBOpsManager Custom Resource](https://github.com/mongodb/mongodb-enterprise-kubernetes/tree/master/samples/ops-manager)

-   [MongoDB Custom Resource](https://www.mongodb.com/docs/kubernetes-operator/master/tutorial/mdb-resources-arch/#the-mongodb-custom-resource-definition)

    -   Standalone

    -   ReplicaSet

    -   ShardedCluster

-   [MongoDBUser Custom Resource](https://www.mongodb.com/docs/kubernetes-operator/master/tutorial/mdb-resources-arch/#reconciling-the-mongodbuser-custom-resource)

-   [MongoDBMulti](https://www.mongodb.com/docs/kubernetes-operator/master/multi-cluster-arch/#deployment-architecture-and-diagram)

![The following diagram describes how the Kubernetes Operator behaves if you make changes to a sharded cluster's](https://mongodb-devhub-cms.s3.us-west-1.amazonaws.com/Mongo_DB_2d08c18419.png)

[Example of Ops Manager and MongoDB Custom Resources on Kubernetes](https://drive.google.com/file/d/1q7dvZ9PT9YU2EtJNyeYyl00DjKvQajl9/view)

## Installing and configuring Enterprise Kubernetes Operator

For this tutorial, we will need the following tools: 

-   [gcloud](https://cloud.google.com/sdk/docs/install)[ ](https://go.dev/dl/)

-   [gke-cloud-auth-plugin](https://cloud.google.com/blog/products/containers-kubernetes/kubectl-auth-changes-in-gke)

-   [Helm](https://helm.sh/docs/intro/install/)

-   [kubectl](https://kubernetes.io/docs/tasks/tools/)

-   [kubectx](https://github.com/ahmetb/kubectx)

-   [Git](https://git-scm.com/downloads)

## GKE Kubernetes cluster creation 

To start, let's create a Kubernetes cluster in a new [project](https://cloud.google.com/resource-manager/docs/creating-managing-projects). We will be using [GKE Kubernetes](https://cloud.google.com/kubernetes-engine). I use this script to create the cluster. The cluster will have four worker nodes and act as Ops Manager and MongoDB Enterprise Operators Kubernetes Cluster.

```bash
CLUSTER_NAME=master-operator
ZONE=us-south1-a
K8S_VERSION=1.23
MACHINE=n2-standard-2
gcloud container clusters create "${CLUSTER_NAME}" \
  --zone "${ZONE}" \
  --machine-type "${MACHINE}" --cluster-version="${K8S_VERSION}" \
  --disk-type=pd-standard --num-nodes 4
```

Now that the cluster has been created, we need to obtain the credentials.

```bash
gcloud container clusters get-credentials "${CLUSTER_NAME}" \
  --zone "${ZONE}"
```


Display the newly created cluster.

```bash
gcloud container clusters list

NAME                LOCATION       MASTER_VERSION    NUM_NODES  STATUS
master-operator     us-south1-a    1.23.14-gke.1800      4      RUNNING
```

We can also display Kubernetes full cluster name using [`kubectx`](https://github.com/ahmetb/kubectx).

```bash
kubectx
```

You should see your cluster listed here. Make sure your context is set to master cluster.

```bash
kubectx $(kubectx | grep "master-operator" | awk '{print $1}')
```


We are able to start MongoDB Kubernetes Operator installation on our newly created Kubernetes cluster! 

## Enterprise Kubernetes Operator 

We can install the MongoDB Enterprise Operator with a single line Helm command. The first step is to  add the [MongoDB Helm Charts for Kubernetes](https://mongodb.github.io/helm-charts/) repository to Helm.

```bash
helm repo add mongodb https://mongodb.github.io/helm-charts
```


I want to create the operator in a separate, dedicated Kubernetes namespace (the operator uses `default` namespace by default). This will allow me to isolate the operator and any resources it creates from other resources in my cluster. The following command will install the CRDs and the Enterprise Operator in the `mongodb-operator`namespace. The operator will be watching only the `mongodb-operator` namespace. You can read more about [setting up the operator to watch more namespaces](https://www.mongodb.com/docs/kubernetes-operator/master/tutorial/set-scope-k8s-operator/#operator-uses-a-subset-of-namespaces) in the official MongoDB documentation.

Start by creating the `mongodb-operator`namespace.

```bash
NAMESPACE=mongodb-operator
kubectl create ns "${NAMESPACE}"
```


Install the MongoDB Kubernetes Operator and set it to watch only the `mongodb-operator` namespace.

```bash
HELM_CHART_VERSION=1.16.3
helm install enterprise-operator mongodb/enterprise-operator \
  --namespace "${NAMESPACE}" \
  --version="${HELM_CHART_VERSION}" \
  --set operator.watchNamespace="${NAMESPACE}"
```


The namespace has been created and the operator is running! You can see this by listing the pods in the newly created namespace.

```bash
kubectl get ns

NAME               STATUS   AGE
default            Active   4m9s
kube-node-lease    Active   4m11s
kube-public        Active   4m12s
kube-system        Active   4m12s
mongodb-operator   Active   75s
```

```bash
kubectl get po -n "${NAMESPACE}"

NAME                                    READY   STATUS   RESTARTS   AGE
mongodb-enterprise-operator-649bbdddf5   1/1    Running   0         7m9s
```


You can see that the helm chart is running with this command.

```bash
helm list --namespace "${NAMESPACE}"

NAME                NAMESPACE     REVISION       VERSION
enterprise-operator mongodb-operator 1 deployed enterprise-operator-1.17.2
```

### Verify the installation

You can verify that the installation was successful and is currently running with the following command.

```bash
helm get manifest enterprise-operator --namespace "${NAMESPACE}"
```

Let's display Custom Resource Definitions installed in the step above in the watched namespace.

```bash
kubectl -n "${NAMESPACE}" get crd | grep -E '^(mongo|ops)'

mongodb.mongodb.com                              2022-12-30T16:17:07Z
mongodbmulti.mongodb.com                         2022-12-30T16:17:08Z
mongodbusers.mongodb.com                         2022-12-30T16:17:09Z
opsmanagers.mongodb.com                          2022-12-30T16:17:09Z
```


All required service accounts has been created in watched namespace.

```bash
kubectl -n "${NAMESPACE}" get sa | grep -E '^(mongo)'

mongodb-enterprise-appdb           1         36s
mongodb-enterprise-database-pods   1         36s
mongodb-enterprise-operator        1         36s
mongodb-enterprise-ops-manager     1         36s
```

Validate if the Kubernetes Operator was installed correctly by running the following command and verify the output.

```bash
kubectl describe deployments mongodb-enterprise-operator -n \
  "${NAMESPACE}"
```


Finally, double-check watched namespaces.

```bash
kubectl describe deploy mongodb-enterprise-operator -n "${NAMESPACE}" | grep WATCH

      WATCH_NAMESPACE:               mongodb-operator
```


The MongoDB Enterprise Operator is now running in your GKE cluster.

## MongoDB Atlas Kubernetes Operator

It's worth mentioning another operator here --- a new service that integrates Atlas resources with your Kubernetes cluster. Atlas can be deployed in multi-cloud environments including Google Cloud. The Atlas Kubernetes Operator allows you to deploy and manage cloud-native applications that require data services in a single control plane with secure enterprise platform integration.

This operator is responsible for managing resources in Atlas using Kubernetes custom resources, ensuring that the configurations of projects, database deployments, and database users in Atlas are consistent with each other. The Atlas Kubernetes Operator uses the `AtlasProject`, `AtlasDeployment`, and `AtlasDatabaseUser` Custom Resources that you create in your Kubernetes cluster to manage resources in Atlas.

These custom resources allow you to define and configure the desired state of your projects, database deployments, and database users in Atlas. To learn more, head over to our blog post on [application deployment in Kubernetes](https://www.mongodb.com/developer/products/atlas/kubernetes-operator-application-deployment/) with the MongoDB Atlas Operator.

## Conclusion

Upon the successful installation of the Kubernetes Operator, we are able to use the capabilities of the MongoDB Enterprise Kubernetes Operator to run MongoDB objects on our Kubernetes cluster. The Operator enables easy deploy of the following applications into Kubernetes clusters:

-   MongoDB --- replica sets, sharded clusters, and standalones --- with authentication, TLS, and many more options.

-   Ops Manager --- enterprise management, monitoring, and backup platform for MongoDB. The Operator can install and manage Ops Manager in Kubernetes for you. Ops Manager can manage MongoDB instances both inside and outside Kubernetes. Installing Ops Manager is covered in the [second article](https://www.mongodb.com/developer/products/connectors/mastering-ops-manager) of the series.

-   MongoMulti --- Multi-Kubernetes-cluster deployments allow you to add MongoDB instances in global clusters that span multiple geographic regions for increased availability and global distribution of data. This is covered in the [final part](https://www.mongodb.com/developer/products/connectors/deploying-across-multiple-kubernetes-clusters) of this series.

Want to see the MongoDB Enterprise Kubernetes Operator in action and discover all the benefits it can bring to your Kubernetes deployment? Continue reading the [next blog](https://www.mongodb.com/developer/products/connectors/mastering-ops-manager) of this series and we'll show you how to best utilize the Operator for your needs

Learn how to deploy the MongoDB Enterprise Kubernetes Operator in this tutorial.

This article is part of a three-parts series on deploying MongoDB across multiple Kubernetes clusters using the operators.

-   [Deploying the MongoDB Enterprise Kubernetes Operator on Google Cloud](https://www.mongodb.com/developer/products/connectors/deploying-kubernetes-operator)

-   Mastering MongoDB Ops Manager

-   [Deploying MongoDB Across Multiple Kubernetes Clusters With MongoDBMulti](https://www.mongodb.com/developer/products/connectors/deploying-across-multiple-kubernetes-clusters)

Managing MongoDB deployments can be a rigorous task, particularly when working with large numbers of databases and servers. Without the right tools and processes in place, it can be time-consuming to ensure that these deployments are running smoothly and efficiently. One significant issue in managing MongoDB clusters at scale is the lack of automation, which can lead to time-consuming and error-prone tasks such as backups, recovery, and upgrades. These tasks are crucial for maintaining the availability and performance of your clusters.

Additionally, monitoring and alerting can be a challenge, as it may be difficult to identify and resolve issues with your deployments. To address these problems, it's essential to use software that offers monitoring and alerting capabilities. Optimizing the performance of your deployments also requires guidance and support from the right sources.

Finally, it's critical for your deployments to be secure and compliant with industry standards. To achieve this, you need features that can help you determine if your deployments meet these standards.

[MongoDB Ops Manager](https://www.mongodb.com/products/ops-manager) is a web-based application designed to assist with the management and monitoring of MongoDB deployments. It offers a range of features that make it easier to deploy, manage, and monitor MongoDB databases, such as:

-   [Automated backups and recovery](https://www.mongodb.com/docs/ops-manager/current/application/#backup): Ops Manager can take automated backups of your MongoDB deployments and provide options for recovery in case of failure.

-   [Monitoring and alerting](https://www.mongodb.com/docs/ops-manager/current/application/#monitoring): Ops Manager provides monitoring and alerting capabilities to help identify and resolve issues with your MongoDB deployments.

-   [Performance optimization](https://www.mongodb.com/docs/ops-manager/current/tutorial/performance-advisor/): Ops Manager offers tools and recommendations to optimize the performance of your MongoDB deployments.

-   [Upgrade management](https://www.mongodb.com/docs/ops-manager/current/tutorial/change-mongodb-version/): Ops Manager can help you manage and plan upgrades to your MongoDB deployments, including rolling upgrades and backups to ensure data availability during the upgrade process.

-   [Security and compliance](https://www.mongodb.com/docs/ops-manager/current/tutorial/nav/security/): Ops Manager provides features to help you secure your MongoDB deployments and meet compliance requirements.

However, managing Ops Manager can be a challenging task that requires a thorough understanding of its inner workings and how it interacts with the internal MongoDB databases. It is necessary to have the knowledge and expertise to perform upgrades, monitor it, audit it, and ensure its security. As Ops Manager is a crucial part of managing the operation of your MongoDB databases, its proper management is essential.

Fortunately, the MongoDB Enterprise Kubernetes Operator enables us to run Ops Manager on Kubernetes clusters, using native Kubernetes capabilities to manage Ops Manager for us, which makes it more convenient and efficient.

## Kubernetes: MongoDBOpsManager custom resource

The [MongoDB Enterprise Kubernetes Operator](https://www.mongodb.com/docs/kubernetes-operator/master/) is software that can be used to deploy Ops Manager and MongoDB resources to a Kubernetes cluster, and it's responsible for managing the lifecycle of each of these deployments. It has been developed based on years of experience and expertise, and it's equipped with the necessary knowledge to properly install, upgrade, monitor, manage, and secure MongoDB objects on Kubernetes.

The Kubernetes Operator uses the [MongoDBOpsManager custom resource](https://github.com/mongodb/mongodb-enterprise-kubernetes/tree/master/samples/ops-manager) to manage Ops Manager objects. It constantly monitors the specification of the custom resource for any changes and, when changes are detected, the operator validates them and makes the necessary updates to the resources in the Kubernetes cluster.

MongoDBOpsManager custom resources specification defines the following Ops Manager components:

-   The [Application Database](https://www.mongodb.com/docs/kubernetes-operator/master/tutorial/om-arch/#application-database)

-   The [Ops Manager application](https://www.mongodb.com/docs/kubernetes-operator/master/tutorial/om-arch/#application)

-   The [Backup Daemon](https://www.mongodb.com/docs/kubernetes-operator/master/tutorial/om-arch/#backup-daemon)

![MongoDBOpsManager custom resources specification defines the following Ops Manager components:
the Application Database,
the Ops Manager application, and
the Backup Daemon.](https://mongodb-devhub-cms.s3.us-west-1.amazonaws.com/meko_arch_a94d843eaa.svg)

When you use the Kubernetes Operator to create an instance of Ops Manager, the Ops Manager MongoDB Application Database will be deployed as a replica set. It's not possible to configure the Application Database as a standalone database or a sharded cluster.

The Kubernetes Operator automatically sets up Ops Manager to monitor the Application Database that powers the Ops Manager Application. It creates a project named  `<ops-manager-deployment-name>-db` to allow you to monitor the Application Database deployment. While Ops Manager monitors the Application Database deployment, it does not manage it.

When you deploy Ops Manager, you need to configure it. This typically involves using the [configuration wizard](https://www.mongodb.com/docs/ops-manager/current/reference/config/ui-settings/). However, you can bypass the configuration wizard if you set certain essential settings in your object specification before deployment. I will demonstrate that in this post.

The Operator automatically enables backup. It deploys a StatefulSet, which consists of a single pod, to host the Backup Daemon Service and creates a Persistent Volume Claim and Persistent Volume for the Backup Daemon's head database. The operator uses the Ops Manager API to enable the Backup Daemon and configure the head database.

## Getting started

Alright, let's get started using the operator and build something! For this tutorial, we will need the following tools: 

-   [gcloud](https://cloud.google.com/sdk/docs/install)[ ](https://go.dev/dl/)

-   [gke-cloud-auth-plugin](https://cloud.google.com/blog/products/containers-kubernetes/kubectl-auth-changes-in-gke)

-   [Helm](https://helm.sh/docs/intro/install/)

-   [kubectl](https://kubernetes.io/docs/tasks/tools/)

-   [kubectx](https://github.com/ahmetb/kubectx)

-   [git](https://git-scm.com/downloads)

To get started, we should first create a Kubernetes cluster and then install the MongoDB Kubernetes Operator on the cluster. [Part 1](https://www.mongodb.com/developer/products/connectors/deploying-kubernetes-operator) of this series provides instructions on how to do so.

> **Note**
> For the sake of simplicity, we are deploying Ops Manager in the same namespace as our MongoDB Operator. In a production environment, you should deploy Ops Manager in its own namespace.

### Environment pre-checks 

Upon successful creation of a cluster and installation of the operator (described in [Part 1](https://www.mongodb.com/developer/products/connectors/deploying-kubernetes-operator)), it's essential to validate their readiness for use.

```bash
gcloud container clusters list

NAME                LOCATION       MASTER_VERSION    NUM_NODES  STATUS\
master-operator     us-south1-a    1.23.14-gke.1800      4      RUNNING
```

Display our new Kubernetes full cluster name using `[kubectx](https://github.com/ahmetb/kubectx)`.

```bash
kubectx
```

You should see your cluster listed here. Make sure your context is set to master cluster.

```bash
kubectx $(kubectx | grep "master-operator" | awk '{print $1}')
```

In order to continue this tutorial, make sure that the operator is in the `running`state.

```bash
kubectl get po -n "${NAMESPACE}"

NAME                                    READY   STATUS   RESTARTS   AGE\
mongodb-enterprise-operator-649bbdddf5   1/1    Running   0         7m9s
```

## Using the MongoDBOpsManager CRD

Create a secret containing the username and password on the master Kubernetes cluster for accessing the Ops Manager user interface after installation.

```bash
kubectl -n "${NAMESPACE}" create secret generic om-admin-secret \
  --from-literal=Username="opsmanager@example.com" \
  --from-literal=Password="p@ssword123" \
  --from-literal=FirstName="Ops" \
  --from-literal=LastName="Manager"
```
​​
### Deploying Ops Manager 

Then, we can [deploy Ops Manger on the master Kubernetes](https://github.com/mongodb/mongodb-enterprise-kubernetes/tree/master/samples/ops-manager) cluster with the help of `opsmanagers`  Custom Resource, creating `MongoDBOpsManager` object, using the following manifest:

```bash
OM_VERSION=6.0.5
APPDB_VERSION=5.0.5-ent
kubectl apply -f - <<EOF
apiVersion: mongodb.com/v1
kind: MongoDBOpsManager
metadata:
  name: ops-manager
  namespace: "${NAMESPACE}"
spec:
  version: "${OM_VERSION}"
  # the name of the secret containing admin user credentials.
  adminCredentials: om-admin-secret
  externalConnectivity:
    type: LoadBalancer
  configuration:
    mms.ignoreInitialUiSetup: "true"
    automation.versions.source: mongodb
    mms.adminEmailAddr: support@example.com
    mms.fromEmailAddr: support@example.com
    mms.replyToEmailAddr: support@example.com
    mms.mail.hostname: example.com
    mms.mail.port: "465"
    mms.mail.ssl: "false"
    mms.mail.transport: smtp
  # the Replica Set backing Ops Manager.
  applicationDatabase:
    members: 3
    version: "${APPDB_VERSION}"
EOF
```
 
 After a few minutes, we should see our Ops Manager and Ops Manager MongoDB application database pods running.

```​​bash
kubectl -n "${NAMESPACE}" get pods

NAME                                       READY      STATUS     RESTARTS
ops-manager-0                           1/1      Running      0
ops-manager-db-0                        3/3      Running      0
ops-manager-db-1                        3/3      Running      0
ops-manager-db-2                        3/3      Running      0
```

Storage part creation has been orchestrated by the operator. [Persistent Volumes Claims ](https://kubernetes.io/docs/concepts/storage/persistent-volumes/) are created and can be displayed via

```bash
kubectl -n "${NAMESPACE}" get pvc

NAME       STATUS   VOLUME     CAPACITY  ACCESS MODES   STORAGE CLASS   AGE
data-ops-manager-db-0 Bound pvc-d5a1b385-6d1b  15Gi  RWO  standard  59m
data-ops-manager-db-1 Bound pvc-db0e89dc-d73a  15Gi  RWO  standard  58m
data-ops-manager-db-2 Bound pvc-c7e124a2-917   15Gi  RWO  standard  57m
```

[Kubernetes StatefulSets](https://kubernetes.io/docs/concepts/workloads/controllers/statefulset/) has also been created by the Operator.

```bash
kubectl -n "${NAMESPACE}" get sts

NAME             READY   AGE
ops-manager      1/1     29m31s
ops-manager-db   3/3     30m29s
```

The operator has created all required network services, including a Load Balancer service type for the Ops manager access, so we now have an external IP address and can log into the Ops Manager from outside the Kubernetes cluster. You can verify IPs by viewing the services.

```bash
kubectl -n "${NAMESPACE}" get svc

NAME                  TYPE    CLUSTER-IP    EXTERNAL-IP  PORT(S)
ops-manager-db-svc  ClusterIP    None        <none>      27017/TCP
ops-manager-svc     ClusterIP    None        <none       8080/TCP,25999/TCP
ops-manager-svc-ext LoadBalancer 10.76.10.231 34.174.54.103 8080:32078/TCP ,25999:31961/TCP    
```

The following diagram describes how the Kubernetes Operator reconciles changes to the  MongoDBOpsManager CustomResourceDefinition.

![The following diagram describes how the Kubernetes Operator reconciles changes to the MongoDBOpsManager CustomResourceDefinition](https://mongodb-devhub-cms.s3.us-west-1.amazonaws.com/Ops_Manager_objects_5e8862d468.png)

To generate Ops Manager URL address, execute:

```bash
URL=http://$(kubectl -n "${NAMESPACE}" get svc ops-manager-svc-ext -o jsonpath='{.status.loadBalancer.ingress[0].ip}:{.spec.ports[0].port}')
echo $URL
```

The URL address will look similar to the example provided, but IP address may vary depending on the Kubernetes cluster.

```bash
http://34.174.54.105:8080
```

The final step is to update the Ops Manager Kubernetes manifest to include an external IP address created by Load Balancer in `spec.configuration.mms.centralUrl`  via `kubectl patch`.

```bash
kubectl -n "${NAMESPACE}" patch om ops-manager --type=merge -p "{\"spec\":{\"configuration\":{\"mms.centralUrl\":\"${URL}\"}}}"
```

We should wait a few minutes. The Ops Manager pod must be restarted, so wait until the `ops-manager-0` pod is in the `running` state again.

Using the username and password stored in the `om-admin-secret` (opsmanager@example.com : p@ssword123), we can log in to the Ops Manager User Interface using the address in the `$URL` variable.

![Enter your username and password and click Login](https://mongodb-devhub-cms.s3.us-west-1.amazonaws.com/Ops_Manager_c44549fd1e.png)

The Kubernetes Operator was in the Ops Manager `ops-manager-db` organization and the `ops-manager-db` project.

![Select Ops Manager organization](https://mongodb-devhub-cms.s3.us-west-1.amazonaws.com/Ops_Manager_UI_834b2820aa.png)

If we click on the project `ops-manager-db`, we will be redirected to the panel where we can see the database pods of the Ops Manager application. Ops Manager monitors this database.

![Manage your MongoDB cluster from Ops Manager](https://mongodb-devhub-cms.s3.us-west-1.amazonaws.com/Ops_Manager_App_DB_5f3ae3ae3e.png)

## Basic troubleshooting

If you run through issues during the installation, here are some clues to help you investigate what went wrong.

If you want to display an Ops Manager cluster, use this command. It will show you a general overview of `om` objects, including internal database and Backup Daemon state.

```bash
kubectl -n "${NAMESPACE}" get om

NAME REPLICAS VERSION STATE(OPSMANAGER) STATE(APPDB) STATE (BACKUP)   AGE
ops-manager     6.0.5     Running        Running     Pending          18m  
```
 

You can use `describe` to get a detailed overview.

```bash
kubectl -n "${NAMESPACE}" describe om ops-manager
```

It's always a good idea to check the Ops Manager logs

```bash
kubectl -n "${NAMESPACE}" logs -f po/ops-manager-0
```

or current `events` from the namespace.

```bash
kubectl -n "${NAMESPACE}" get events
```

## Summary

We have just installed Ops Manager on our Kubernetes cluster. This gives us many benefits. The operator has properly installed and configured the Ops Manager instance with the internal database, took care of the storage portion, and created network services, including a Load Balancer. We can now use Ops Manager and easily create any kind of MongoDB database on the Kubernetes cluster following the best practices, monitor instances of our databases, introduce query optimizations with the help of the Ops Manager performance advisor, and provide backup, restore, or rolling upgrades through automation.

In the [next part](https://www.mongodb.com/developer/products/connectors/deploying-across-multiple-kubernetes-clusters), I will demonstrate how to run the latest type of MongoDB Kubernetes Custom Resource, a Multi Cluster Replica Set. This replica set will be deployed across three separate Kubernetes clusters located in different regions, providing the ideal solution for critical applications that require continuous availability, even in the event of a Kubernetes cluster failure.
Want to get started with the MongoDB Ops Manager on your own Kubernetes cluster? [Install  it now](https://github.com/mongodb/helm-charts/tree/main/charts/enterprise-operator) and see for yourself how it can simplify your operations. Make sure to visit the [MongoDB community forum](https://www.mongodb.com/community/forums/) for the latest discussions and resources

Learn how to deploy the MongoDB Ops Manager in a Kubernetes cluster with the MongoDB Kubernetes Operators.

Connectors

Featured

Spark Up Your MongoDB and BigQuery Using BigQuery Spark Stored Procedures

Using AWS IAM Authentication with the MongoDB Connector for Apache Kafka

How to Get Started with MongoDB Atlas Stream Processing and the HashiCorp Terraform MongoDB Atlas Provider

All Connectors Content