Configure Multi-Collection Initial Sync

Go to the Stream Processing page for your project.

1. In the Atlas UI, go to the Stream Processing page for your Atlas project.

2. Click Manage in the pane of your stream processing workspace that holds your Atlas connections to the source and destination clusters.

Open the Visual Builder.

1. Click Create stream processor.

2. Select the Visual Builder.

Configure the source with an explicit collection list.

1. In the Source field, select your Atlas connection to the source cluster from the Connection drop-down list.

2. In the JSON text box, configure the $source stage:

{
  "$source": {
    "connectionName": "<source-connection-name>",
    "db": "sample_analytics",
    "coll": ["customers", "accounts"],
    "config": { "fullDocument": "required" },
    "initialSync": { "enable": true }
  }
}

The $source stage synchronizes existing documents in customers and accounts, in that order, then opens a change stream scoped to both collections:

• coll: ["customers", "accounts"] sets the sync scope to both collections instead of one.

• initialSync: { enable: true } turns on the sync phase before Atlas Stream Processing starts tailing the change stream.

• fullDocument: "required" includes the current version of the document in every change event.

Add a stage that prepares each event for the sink.

1. In the Start building your pipeline pane, click + Custom stage.

2. In the JSON text box, add a $replaceRoot stage:

{
  "$replaceRoot": {
    "newRoot": {
      "$cond": {
        "if": { "$eq": [{ "$meta": "stream.source.operationType" }, "delete"] },
        "then": { "$meta": "stream.source.documentKey" },
        "else": "$fullDocument"
      }
    }
  }
}

The $replaceRoot stage prepares each event for the sink stage. It selects the document key for a delete event and the full document for every other event, so the $merge stage always receives a document that has an _id field.

Configure the sink with dynamic namespace routing.

1. In the Sink field, select your Atlas connection to the destination cluster from the Connection drop-down list.

2. In the JSON text box, configure the $merge stage:

{
  "$merge": {
    "into": {
      "connectionName": "<destination-connection-name>",
      "db": { "$meta": "stream.source.ns.db" },
      "coll": { "$meta": "stream.source.ns.coll" }
    },
    "on": "_id",
    "whenMatched": {
      "$cond": {
        "if": { "$eq": [{ "$meta": "stream.source.operationType" }, "delete"] },
        "then": "delete",
        "else": "replace"
      }
    },
    "whenNotMatched": {
      "$cond": {
        "if": { "$eq": [{ "$meta": "stream.source.operationType" }, "delete"] },
        "then": "discard",
        "else": "insert"
      }
    }
  }
}

The $merge stage writes each event to the collection on <destination-connection-name> that matches its source namespace:

• db and coll read the source event's namespace metadata, so the same pipeline routes customers events to customers and accounts events to accounts.

• whenMatched and whenNotMatched each branch on operation type, so the stage deletes, replaces, or inserts the destination document depending on the source event.

Enter processor details.

1. In the Stream processor name field, enter replicate_analytics_sp.

2. Select the tier for the stream processor. To choose a tier for your workload, see Atlas Stream Processing Tier Selection Guide.

Create the stream processor.

Click Create stream processor.

Start the stream processor.

On the Stream Processors tab, select replicate_analytics_sp and click Start.

Atlas Stream Processing synchronizes customers and accounts to <destination-connection-name>, then keeps both collections current. To learn more about starting, stopping, and monitoring stream processors, see Develop and Manage Stream Processors.

Connect to your stream processing workspace.

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

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

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

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

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

4. Paste the connection string into your terminal and replace the <password> placeholder with the credentials for the user. Press Enter to connect to your stream processing workspace.

Define the source stage.

Declare a variable and assign it a $source stage specification that lists both collections and enables initialSync:

let s = {
  $source: {
    connectionName: "<source-connection-name>",
    db: "sample_analytics",
    coll: ["customers", "accounts"],
    config: { fullDocument: "required" },
    initialSync: { enable: true }
  }
}

The $source stage synchronizes existing documents in customers and accounts, in that order, then opens a change stream scoped to both collections:

• coll: ["customers", "accounts"] sets the sync scope to both collections instead of one.

• initialSync: { enable: true } turns on the sync phase before Atlas Stream Processing starts tailing the change stream.

• fullDocument: "required" includes the current version of the document in every change event.

Define a stage that prepares each event for the sink.

Declare a variable and assign it a $replaceRoot stage specification:

let r = {
  $replaceRoot: {
    newRoot: {
      $cond: {
        if:   { $eq: [{ $meta: "stream.source.operationType" }, "delete"] },
        then: { $meta: "stream.source.documentKey" },
        else: "$fullDocument"
      }
    }
  }
}

The $replaceRoot stage prepares each event for the sink stage. It selects the document key for a delete event and the full document for every other event, so the $merge stage always receives a document that has an _id field.

Define the sink stage.

Declare a variable and assign it a $merge stage specification that routes each event by its source namespace:

let m = {
  $merge: {
    into: {
      connectionName: "<destination-connection-name>",
      db:   { $meta: "stream.source.ns.db" },
      coll: { $meta: "stream.source.ns.coll" }
    },
    on: "_id",
    whenMatched: { $cond: {
        if:   { $eq: [{ $meta: "stream.source.operationType" }, "delete"] },
        then: "delete",
        else: "replace"
    } },
    whenNotMatched: { $cond: {
        if:   { $eq: [{ $meta: "stream.source.operationType" }, "delete"] },
        then: "discard",
        else: "insert"
    } }
  }
}

The $merge stage writes each event to the collection on <destination-connection-name> that matches its source namespace:

• db and coll read the source event's namespace metadata, so the same pipeline routes customers events to customers and accounts events to accounts.

• whenMatched and whenNotMatched each branch on operation type, so the stage deletes, replaces, or inserts the destination document depending on the source event.

Create the stream processor.

Use the sp.createStreamProcessor() method to create a stream processor named replicate_analytics_sp from the stages you defined:

sp.createStreamProcessor("replicate_analytics_sp", [s, r, m])

To assign the processor to a tier other than the tier of your stream processing workspace, pass a tier option in the options object. To choose a tier for your workload, see Atlas Stream Processing Tier Selection Guide.

Start the stream processor.

Use the sp.processor.start() method to start replicate_analytics_sp:

sp.replicate_analytics_sp.start()

{ ok: 1 }

Atlas Stream Processing synchronizes customers and accounts to <destination-connection-name>, then keeps both collections current. To learn more about starting, stopping, and monitoring stream processors, see Develop and Manage Stream Processors.