How to Migrate PostgreSQL to MongoDB With Confluent Kafka

1
-- Create the database mytestdb
2
CREATE DATABASE mytestdb;
3
4
-- Connect to the mytestdb database
5
\c org;
6
7
-- Create the users table
8
CREATE TABLE users (
9
    id SERIAL PRIMARY KEY,
10
    firstname VARCHAR(50),
11
    lastname VARCHAR(50),
12
    age INT
13
);
14
15
-- Insert sample data into the 'users' table
16
INSERT INTO users (firstname, lastname, age)
17
VALUES
18
    ('John', 'Doe', 25),
19
    ('Jane', 'Smith', 30),
20
    ('Bob', 'Johnson', 22);

1
{
2
  "name": "mongodb-atlas-sink",
3
  "config": {
4
    "connector.class": "com.mongodb.kafka.connect.MongoSinkConnector",
5
    "tasks.max": "1",
6
    "topics": "your-topic-name",
7
    "connection.uri": "mongodb+srv://<username>:<password>@cluster.mongodb.net/test",
8
    "database": "your-database",
9
    "collection": "your-collection",
10
    "key.converter": "org.apache.kafka.connect.storage.StringConverter",
11
    "value.converter": "org.apache.kafka.connect.json.JsonConverter",
12
    "value.converter.schemas.enable": "false",
13
    "transforms": "addFields,unwrap",
14
    "transforms.addFields.type": "org.apache.kafka.connect.transforms.InsertField$Value",
15
    "transforms.addFields.static.field": "name",
16
    "transforms.addFields.static.value": "${r:firstname}-${r:lastname}",
17
    "transforms.unwrap.type": "io.debezium.transforms.UnwrapFromEnvelope",
18
    "transforms.unwrap.drop.tombstones": "false",
19
    "transforms.unwrap.delete.handling.mode": "none"
20
  }
21
}