Community

Rajesh S Nair

Senior Software Engineer

In today's data-driven world, businesses are continuously seeking innovative ways to harness the full potential of their data. One critical aspect of this journey involves data migration – the process of transferring data from one database system to another, often driven by evolving business needs, scalability requirements, or the desire to tap into new technologies.

In this era of digital transformation, where agility and scalability are paramount, organizations are increasingly turning to NoSQL databases like MongoDB for their ability to handle unstructured or semi-structured data at scale. On the other hand, relational databases like PostgreSQL have long been trusted for their robustness and support for structured data.

As businesses strive to strike the right balance between the structured and unstructured worlds of data, the question arises: How can you seamlessly migrate from a relational database like PostgreSQL to the flexible [documented-oriented model](https://www.mongodb.com/document-databases) of MongoDB while ensuring data integrity, minimal downtime, and efficient synchronization?

The answer lies in an approach that combines the power of Confluent Kafka, a distributed streaming platform, with the agility of MongoDB. In this article, we'll explore the art and science of migrating from PostgreSQL to MongoDB Atlas, leveraging Confluent Kafka as our data streaming bridge. We'll delve into the step-by-step tutorial that can make this transformation journey a success, unlocking new possibilities for your data-driven initiatives.

## Kafka: a brief introduction

### What is Apache Kafka?
Apache Kafka is an open-source distributed streaming platform developed by the Apache Software Foundation that is designed to handle real-time data streams.

To understand Kafka, imagine a busy postal system in a bustling city. In this city, there are countless businesses and individuals sending packages and letters to one another, and it's essential that these messages are delivered promptly and reliably.

Apache Kafka is like the central hub of this postal system, but it's not an ordinary hub; it's a super-efficient, high-speed hub with a memory that never forgets. When someone sends a message (data) to Kafka, it doesn't get delivered immediately. Instead, it's temporarily stored within Kafka's memory. Messages within Kafka are not just one-time deliveries. They can be read and processed by multiple parties. Imagine if every package or letter sent through the postal system had a copy available for anyone who wanted it. This is the core concept of Kafka: It's a distributed, highly scalable, and fault-tolerant message streaming platform.

From maintaining real-time inventory information for e-commerce to supporting real-time patient monitoring, Kafka has varied business use cases in different industries and can be used for log aggregation and analysis, event sourcing, real-time analytics, data integration, etc.

## Kafka Topics
In the same analogy of the postal system, the system collects and arranges its letters and packages into different sections and organizes them into compartments for each type of item. Kafka does the same. The messages it receives from the producer of data are arranged and organized into Kafka topics. Kafka topics are like different mailboxes where messages with a similar theme are placed, and various programs can send and receive these messages to exchange information. This helps keep data well-organized and ensures that the right people or systems can access the information they need from the relevant topic.

## Kafka connectors
Kafka connectors are like special mailboxes that format and prepare letters (data) in a way that Kafka can understand, making it easier for data to flow between different systems. Say the sender (system) wants to send a letter (data) to the receiver (another system) using our postal system (Kafka). Instead of just dropping the letter in the regular mailbox, the sender places it in a special connector mailbox outside their house. This connector mailbox knows how to format the letter properly. So connectors basically act as a bridge that allows data to flow between Kafka and various other data systems.

## Confluent Kafka
Confluent is a company that builds tools and services. It has built tools and services for Apache Kafka to make it more robust and feature-rich. It is like working with a more advanced post office that not only receives and delivers letters but also offers additional services like certified mail, tracking, and package handling. The migration in this article is done using Confluent Kafka through its browser user interface.

## Setting up a Confluent Kafka account
To begin with, you can set up an account on Confluent Kafka by registering on the [Confluent Cloud website](https://confluent.cloud/login). You can sign up with your email account or using GitHub.

Once you log in, this is how the home page looks:
![Confluent Cloud Homepage](https://images.contentstack.io/v3/assets/blt39790b633ee0d5a7/blt18f94d7a969ea461/6576ded4c5fb31557b9f9106/Confluent_Cloud_Homepage.png)

This free account comes with free credits worth $400 which you can use to utilize the resources in the Confluent Cloud. If your database size is small, your migration could also be completed within this free credit limit. If you go to the billing section, you can see the details regarding the credits.

To create a new cluster, topics, and connectors for your migration, click on the Environments tab from the side menu and create a new environment and cluster.

![Confluent environments overview page](https://images.contentstack.io/v3/assets/blt39790b633ee0d5a7/bltc729be1ddf14c580/6576ded47d37e21c40a4a0f7/Environment.png)

You can select the type of cluster. Select the type “basic” which is the free tier with basic configuration. If you want to have a higher configuration for the cluster, you can select the “standard”, “enterprise,” or “dedicated” cluster types which have higher storage, partition, and uptime SLA respectively with hourly rates.

![Create cluster tier selection page](https://images.contentstack.io/v3/assets/blt39790b633ee0d5a7/blt08795d40d8637ffd/6576ded480f4ba868a8d11b4/Tier_Selection.png)

Next, you can select the region/zone where your cluster has to be deployed along with the cloud provider you want for your cluster (AWS, GCP, or Azure ). The prerequisite for your data migration to work through Kafka connectors is that the Kafka cluster where you create your connectors should be in the same region as your MongoDB Atlas cluster to where you will migrate your PostgreSQL data.

Then, you can provide your payment information and launch your cluster.

Once your cluster is launched, this is how the cluster menu looks with options to have a cluster overview and create topics and connectors, among other features.

![Cluster overview page](https://images.contentstack.io/v3/assets/blt39790b633ee0d5a7/blteb27d3cd5daf96e2/6576ded338d46b5d667cafd8/Cluster_Overview.png)

With this, we are ready with the basic Kafka setup to migrate your data from PostgreSQL to MongoDB Atlas.

## Setting up PostgreSQL test data
For this example walkthrough, if you do not have an existing PostgreSQL database that you would like to migrate to a MongoDB Atlas instance using Confluent Kafka, you can create a sample database in PostgreSQL by following the below steps and then continue with this tutorial.

 1. [Download PostgreSQL Database Server](https://www.postgresql.org/download/) from the official website and start your instance locally.
 2. Download the [pgadmin tool](https://www.pgadmin.org/download/) and connect to your local instance.
 3. Create a database ```mytestdb``` and table ```users``` and put some sample data into the employee table.
```sql
-- Create the database mytestdb
CREATE DATABASE mytestdb;

-- Connect to the mytestdb database
\c org;

-- Create the users table
CREATE TABLE users (
    id SERIAL PRIMARY KEY,
    firstname VARCHAR(50),
    lastname VARCHAR(50),
    age INT
);

-- Insert sample data into the 'users' table
INSERT INTO users (firstname, lastname, age)
VALUES
    ('John', 'Doe', 25),
    ('Jane', 'Smith', 30),
    ('Bob', 'Johnson', 22);
```
Keep in mind that the host where your PostgreSQL is running — in this case, your local machine — should have Confluent Kafka whitelisted in a firewall. Otherwise, the source connector will not be able to reach the PostgreSQL instance.

## Steps for data migration using Confluent Kafka
To migrate the data from PostgreSQL to MongoDB Atlas, we have to configure a source connector to connect to PostgreSQL that will stream the data into the Confluent Cloud topic. Then, we will configure a sink connector for MongoDB Atlas to read the data from the created topic and write to the respective database in the MongoDB Atlas cluster.

### Configuring the PostgreSQL source connector
To configure the PostgreSQL source connector, follow the below steps:

 1. Click on the Connectors tab in your newly created cluster in Confluent. It will list popular plugins available in the Confluent Cloud. You can search for the “postgres source” connector plugin and use that to create your custom connector to connect to your PostgreSQL database.
![Selecting Postgres source as connector](https://images.contentstack.io/v3/assets/blt39790b633ee0d5a7/blt2cc46c66e5f28e8e/6576ded431d3b5bb7f227e9f/Postgres_Source.png)

 2. Next, you will be prompted for the topic prefix.  Provide the name of the topic into which you want to stream your PostgreSQL data. If you leave it empty, the topic will be created with the table name for you.

 3. You can then specify the access levels for the new connector you are creating. You can keep it global and also download the API credentials that you can use in your applications, if needed to connect to your cluster. For this migration activity, you will not need it — but you will need to create it to move to the next step.

 4. Next, you will be prompted for connection details of PostgreSQL.You can provide the connection params, schema context, transaction isolation levels, poll intervals, etc. for the connection. 

 5. Select the output record type as JSON. MongoDB natively uses the JSON format. You will also have to provide the name of the table that you are trying to migrate.

 6. In the next screen, you will be redirected to an overview page with all the configurations you provided in JSON format along with the cost for running this source connector per hour.
![Json as output configuration](https://images.contentstack.io/v3/assets/blt39790b633ee0d5a7/blt3e095a9915109a7d/6576ded33a0a12c0c6960c2f/JSON_Source_Settings.png)

 7. Once you create your source connector, you can see its status in the
    Connectors tab  and if it is running or has failed. The source
    connector will start syncing the data to the Confluent Cloud topic
    immediately after starting up. You can check the number of messages
    processed by the connector by clicking on the new connector. If the
    connector has failed to start, you can check connector logs and
    rectify any issues by reconfiguring the connector settings.
![Connectors overview status page](https://images.contentstack.io/v3/assets/blt39790b633ee0d5a7/blt3620c4b3113b518b/6576ded4116ab85ba86eb1c6/Postgres_Connector.png)

### Validating data in the new topic
Once your Postgres source connector is running, you can switch to the Topics tab to list all the topics in your cluster, and you will be able to view the new topic created by the source connector.

If you click on the newly created topic and navigate to the “Messages” tab, you will be able to view the processed messages. If you are not able to see any recent messages, you can check them by selecting the “Jump to time” option, selecting the default partition 0, and providing a recent past time from the date picker. Here, my topic name is “users.”

![Creating users topic](https://images.contentstack.io/v3/assets/blt39790b633ee0d5a7/blt76230cc5ac7c1dd7/6576ded3cc0790c11b39f495/Topics.png)

Below, you can see the messages processed into my “users” topic from the users table in PostgreSQL.
![Processed messages](https://images.contentstack.io/v3/assets/blt39790b633ee0d5a7/blte1c33e44d1027443/6576ded3ec26f800f0a88b15/Users_Topic.png)

### Configuring the MongoDB Atlas sink connector
Now that we have the data that you wanted to migrate (one table, in our example) in our Confluent Cloud topic, we can create a sink connector to stream that data into your MongoDB Atlas cluster. Follow the below steps to configure the data inflow:

 1. Go to the Connectors tab and search for “MongoDB Atlas Sink” to find the MongoDB Atlas connector plugin that you will use to create your custom sink connector.
![Atlas Sink Connector](https://images.contentstack.io/v3/assets/blt39790b633ee0d5a7/blt259f6dbde195b4c2/6576ded319da2638891e88b3/Atlas_Sink_Connector.png)
 2. You will then be asked to select the topic for which you are creating this sink connector. Select the respective topic and click on “Continue.”
![Adding MongoDB Atlas Sink connector](https://images.contentstack.io/v3/assets/blt39790b633ee0d5a7/blt90e77eea6efd5122/6576ded3cd283050e0e464aa/Add_Atlas_Sink_Connector.png)
 3. You can provide the access levels for the sink connector and also download the API credentials if needed, as in the case of the source connector.
 4. In the next section, you will have to provide the connection details for your MongoDB Atlas cluster — including the hostname, username/password, database name, and collection name — into which you want to push the data. The connection string for Atlas will be in the format ```mongodb+srv://<username>:<password>@<hostname>```, so you can get the details from this format. Remember that the Atlas cluster should be in the same region and hosted on the same cloud provider for the Kafka connector to be able to communicate with it. You have to add your Confluent cluster static IP address into the firewall’s allowlist of MongoDB Atlas to allow the connections to your Altas cluster from Confluent Cloud. For non-prod environments, you can also add 0.0.0.0/0 to allow access from anywhere, but it is not recommended for a production environment as it is a security concern allowing any IP access.
![Configuring the MongoDB Atlas Sink connector](https://images.contentstack.io/v3/assets/blt39790b633ee0d5a7/bltdb39db51ec23fe38/6576ded38e05b95a5689ab6c/Configure_Add_Atlas_Sink_Connector.png)
 5. You can select the Kafka input message type as JSON as in the case of the source connector and move to the final review page to view the configuration and cost for your new sink connector.
![Reviewing configurating of MongoDB Atlas Sink connector](https://images.contentstack.io/v3/assets/blt39790b633ee0d5a7/blt0adf57836c9d8d1d/6576ded338d46b66977cafd4/Atlas_Sink_Connector_Review_Page.png)
 6. Once the connector has started, you can query the collection mentioned in your sink connector configuration and you would be able to see the data from your PostgreSQL table in the new collection of your MongoDB Atlas cluster.
### Validating PostgreSQL to Atlas data migration
This data is synced in real-time from PostgreSQL to MongoDB Atlas using the source and sink connectors, so if you try adding a new record or updating/deleting existing records in PostgreSQL, you can see it reflect real-time in your MongoDB Atlas cluster collection, as well.

If your data set is huge, the connectors will catch up and process all the data in due time according to the data size. After completion of the data transfer, you can validate your MongoDB Atlas DB and stop the data flow by stopping the source and sink connectors directly from the Confluent Cloud Interface.

Using Kafka, not only can you sync the data using its event-driven architecture, but you can also transform the data in transfer in real-time while migrating it from PostgreSQL to MongoDB. For example, if you would like to rename a field or concat two fields into one for the new collection in Atlas, you can do that while configuring your MongoDB Atlas sink connector.

Let’s say PostgreSQL had the fields “firstname” and “lastname” for your “users” table, and in MongoDB Atlas post-migration, you only want the “name” field which would be a concatenation of the two fields. This can be done using the “transform” attribute in the sink connector configuration. This provides a list of transformations to apply to your data before writing it to the database. Below is an example configuration.
```json
{
  "name": "mongodb-atlas-sink",
  "config": {
    "connector.class": "com.mongodb.kafka.connect.MongoSinkConnector",
    "tasks.max": "1",
    "topics": "your-topic-name",
    "connection.uri": "mongodb+srv://<username>:<password>@cluster.mongodb.net/test",
    "database": "your-database",
    "collection": "your-collection",
    "key.converter": "org.apache.kafka.connect.storage.StringConverter",
    "value.converter": "org.apache.kafka.connect.json.JsonConverter",
    "value.converter.schemas.enable": "false",
    "transforms": "addFields,unwrap",
    "transforms.addFields.type": "org.apache.kafka.connect.transforms.InsertField$Value",
    "transforms.addFields.static.field": "name",
    "transforms.addFields.static.value": "${r:firstname}-${r:lastname}",
    "transforms.unwrap.type": "io.debezium.transforms.UnwrapFromEnvelope",
    "transforms.unwrap.drop.tombstones": "false",
    "transforms.unwrap.delete.handling.mode": "none"
  }
}
```

## Relational Migrator: an intro
As we are discussing data migration from relational to MongoDB, it’s worth mentioning the MongoDB Relational Migrator. This is a tool designed natively by MongoDB to simplify the process of moving data from relational databases into MongoDB. Relational Migrator analyzes your relational schema and gives recommendations for mapping to a new MongoDB schema.

Its features — including schema analysis, data extraction, indexing, and validation — make it a valuable asset for organizations seeking to harness the benefits of MongoDB's NoSQL platform while preserving their existing relational data assets. Whether for application modernization, data warehousing, microservices, or big data analytics, this tool is a valuable asset for those looking to make the shift from relational to NoSQL databases. It helps to migrate from major relational database technologies including Oracle, SQL Server, MySQL, and PostgreSQL.

Get more information and download and use [relational migrator]( https://www.mongodb.com/products/relational-migrator).

## Conclusion
In the ever-evolving landscape of data management, MongoDB has emerged as a leading NoSQL database, known for its flexibility, scalability, and document-oriented structure. However, many organizations still rely on traditional relational databases to store their critical data. The challenge often lies in migrating data between these disparate systems efficiently and accurately.

Confluent Kafka acts as a great leverage in this context with its event driven architecture and native support for major database engines including MongoDB Atlas.The source and sink connectors would have inbound and outbound data through Topics and acts as a platform for a transparent and hassle free data migration from relational to MongoDB Atlas cluster.


Migrating PostgreSQL to MongoDB Using Confluent Kafka


MongoDB has come a long way from being a database engine developed at the internet company DoubleClick to now becoming this leading NoSQL data store that caters to huge clients from many domains.

With the growth of the database engine, MongoDB kept adding new features and improvements in its database product which makes it the go-to NoSQL database for new requirements and product developments.

One such feature added to the MongoDB tool kit is change streams, which was added with the MongoDB 3.6 release. Before version 3.6, keeping a tailable cursor open was used to perform similar functionality. Change streams are a feature that enables real-time streaming of data event changes on the database.

The event-driven streaming of data is a critical requirement in many use cases of product/feature developments implemented these days. Many applications developed today require that changes in data from one data source need to propagate to another source in real-time. They might also require the application to perform certain actions when a change happens in the data in the data source. Logging is one such use case where the application might need to collect, process, and transmit logs in real-time and thus would require a streaming tool or platform like change streams to implement it.

## What are change streams in MongoDB?

As the word indicates, change streams are the MongoDB feature that captures "change" and "streams" it to the desired target data source.

It is an API that allows the user to subscribe their application to any change in collection, database, or even on the entire deployment. There is no middleware or data polling action to be initiated by the user to leverage this feature of event-driven, real-time data capture.

MongoDB uses replication as the underlying technology for change streams by using the operation logs generated for the data replication between replica members.

The oplog is a special capped collection that records all operations that modify the data stored in the databases. The larger the oplog, the more operations can be recorded on it. Using the oplog for change stream guarantees that the change stream will be triggered in the same order as they were applied to the database.

![MongoDB Change Stream Process](https://mongodb-devhub-cms.s3.us-west-1.amazonaws.com/Change_Stream_Process_1df6ce4806.png)

As seen in the above flow, when there is a CRUD operation on the MongoDB database, the oplog captures it, and those oplog files are used by MongoDB to stream those changes into real-time applications/data receivers.

## Kafka vs change streams

If we compare MongoDB and Kafka technologies, both would fall under completely separate buckets. MongoDB is classified as a NoSQL database,  which can store [JSON-like](https://www.mongodb.com/json-and-bson) document structures.  Kafka is an event streaming platform for real-time data feeds. It is primarily used as a publisher-subscriber model messaging service that provides a replicated message log system to stream data from one source to another.

Kafka helps to ingest huge data sets from desired data sources, filter/aggregate this data, and send it to the intended data source reliably and efficiently. Although MongoDB is a database system and its use case is miles apart from a messaging system like Kafka, the change streams feature does provide it with functionalities similar to those of Kafka.

Basically, change streams act as a messaging service to stream real-time data of any collection from your MongoDB database. It helps you to aggregate/filter that data and store it back to your same MongoDB database data source. In short, if you have a narrow use case that does not require a generalized solution but is curtailed to your data source (MongoDB), then you could go ahead with change streams as your streaming solution. Still, if you want to involve different data sources outside of MongoDB and would like a generalized solution for messaging data sets, then Kafka would make more sense.

By using change streams, you do not need a separate license or server to host your messaging service. Unlike Kafka, you would get the best of both worlds, which is a great database and an efficient messaging system.

MongoDB does provide [Kafka connectors](https://www.mongodb.com/kafka-connector) which could be used to read data in and out of Kafka topics in real-time, but if your use case is not big enough to invest in Kafka, change streams could be the perfect substitute for streaming your data.

Moreover, the Kafka connectors use change streams under the hood, so you would have to build your Kafka setup by setting up connector services and start source and sink connectors for MongoDB. In the case of change streams, you would simply watch for changes in the collection you would want without any prerequisite setup.

## How Change Streams works

Change streams, once open for a collection, act as an event monitoring mechanism on your database/collection or, in some cases, documents within your database.

The core functionality lies in helping you "watch" for changes in an entity. The background work required for this mechanism of streaming changes is implemented by an already available functionality in MongoDB, which is the oplog.

Although it comes with its overheads of blocking system resources, this event monitoring for your source collection has use cases in many business-critical scenarios, like capturing log inputs of application data or monitoring inventory changes for an e-commerce webshop, and so on. So, it's important to fit the change stream with the correct use case.

As the oplog is the driver of the entire change stream mechanism, a replicated environment of at least a single node is the first prerequisite to using change streams. You will also need the following:

-   Start change stream for the collection/database intended.

-   Have the necessary CPU resources for the cluster.

Instead of setting up a self-hosted cluster for fulfilling the above checklist, there is always an option to use the cloud-based hosted solution, [MongoDB Atlas](https://www.mongodb.com/cloud/atlas/register). Using Atlas, you can get a ready-to-use setup with a few clicks. Since change streams are resource-intensive, the cost factor has to be kept in mind while firing an instance in Atlas for your data streaming.

## Implementing change streams in your Java Spring application

In the current backend development world, streams are a burning topic as they help the developers to have a systematic pipeline in place to process the persisted data used in their application. The streaming of data helps to generate reports, have a notification mechanism for certain criteria, or, in some cases, alter some schema based on the events received through streams.

Here, I will demonstrate how to implement a change stream for a Java Spring application.

Once the prerequisite to enable change streams is completed, the steps at the database level are almost done. You will now need to choose the collection on which you want to enable change streams.

Let's consider that you have a Java Spring application for an e-commerce website, and you have a collection called `e_products`, which holds product information of the product being sold on the website.

To keep it simple, the fields of the collection can be:

```json
{"_id"  , "productName", "productDescription" , "price" , "colors" , "sizes"}
```

Now, these fields are populated from your collection through your Java API to show the product information on your website when a product is searched for or clicked on.

Now, say there exists another collection, `vendor_products`, which holds data from another source (e.g., another product vendor). In this case, it holds some of the products in your `e_products` but with more sizes and color options.

You want your application to be synced with the latest available size and color for each product. Change streams can help you do just that. They can be enabled on your `vendor_products` collection to watch for any new product inserted, and then for each of the insertion events, you could have some logic to add the colors/sizes to your `e_products` collection used by your application.

You could create a microservice application specifically for this use case. By using a dedicated microservice, you could allocate sufficient CPU/memory for the application to have a thread to watch on your `vendor_products` collection. The configuration class in your Spring application would have the following code to start the watch:

```java
@Async
    public void runChangeStreamConfig() throws InterruptedException {
        CodecRegistry pojoCodecRegistry = fromRegistries(MongoClientSettings.getDefaultCodecRegistry(),
                fromProviders(PojoCodecProvider.builder().automatic(true).build()));
        MongoCollection<VendorProducts> vendorCollection = mongoTemplate.getDb().withCodecRegistry(pojoCodecRegistry).getCollection("vendor_products", VendorProducts.class);
        List<Bson> pipeline = singletonList(match(eq("operationType", "insert")));
        oldEcomFieldsCollection.watch(pipeline).forEach(s ->
                mergeFieldsVendorToProducts(s.getDocumentKey().get("_id").asString().getValue())
        );
    }
```

In the above code snippet, you can see how the collection is selected to be watched and that the monitored operation type is "insert." This will only check for new products added to this collection. If needed, we could also do the monitoring for "update" or "delete."

Once this is in place, whenever a new product is added to `vendor_products`, this method would be invoked and the `_id` of that product would then be passed to `mergeFieldsVendorToProducts()` method where you can write your logic to merge the various properties from `vendor_products` to the `e_products` collection.

```java
forEach(s ->
        {
            Query query = new Query();
            query.addCriteria(Criteria.where("_id").is(s.get("_id")));
            Update update = new Update();
            update.set(field, s.get(field));
            mongoTemplate.updateFirst(query, update, EProducts.class);
        })
```

This is a small use case for change streams; there are many such examples where change streams can come in handy. It's about using this tool for the right use case.

## Conclusion

In conclusion, change streams in MongoDB provide a powerful and flexible way to monitor changes to your database in real time. Whether you need to react to changes as they happen, synchronize data across multiple systems, or build custom event-driven workflows, change streams can help you achieve these goals with ease.

By leveraging the power of change streams, you can improve the responsiveness and efficiency of your applications, reduce the risk of data inconsistencies, and gain deeper insights into the behavior of your database.

While there is a bit of a learning curve when working with change streams, MongoDB provides comprehensive [documentation](https://www.mongodb.com/docs/manual/changeStreams/) and a range of examples to help you get started. With a little practice, you can take advantage of the full potential of change streams and build more robust, scalable, and resilient applications.

Change streams are an API that allows the user to subscribe to their application to any change in collection, database, or even on the entire deployment. There is no middleware or data polling action to be initiated by the user to leverage this event-driven, real-time data capture feature. Learn how to use change streams with Java in this article.


Rajesh S Nair

Content by Rajesh S Nair

Migrating PostgreSQL to MongoDB Using Confluent Kafka

Using MongoDB Change Streams in Java