> For the complete MongoDB documentation index, see www.mongodb.com/docs/llms.txt

# Configure Multi-Collection Initial Sync

`initialSync` lets an Atlas Stream Processing stream processor ingest preexisting documents in an Atlas collection as though they were insert `changeEvent` documents before it starts tailing the change stream. A single [`$source`](https://www.mongodb.com/docs/atlas/atlas-stream-processing/sp-agg-source.md#mongodb-pipeline-pipe.-source) stage can now run `initialSync` against more than one collection at once.

This page compares single-collection and multi-collection sync characteristics, walks through configuring a multi-collection sync, and demonstrates a cross-cluster replication pipeline built on it.

## Comparison With Single-Collection Initial Sync

The `coll` field of the [`$source`](https://www.mongodb.com/docs/atlas/atlas-stream-processing/sp-agg-source.md#mongodb-pipeline-pipe.-source) stage accepts either a single collection name or an array of collection names. Every collection that you name in the array must belong to the database that you specify in the `db` field. You can't synchronize collections from more than one database in a single [`$source`](https://www.mongodb.com/docs/atlas/atlas-stream-processing/sp-agg-source.md#mongodb-pipeline-pipe.-source) stage. The value you provide determines the sync scope:

| Scope | `$source` Configuration | Behavior |
| --- | --- | --- |
| Single collection | `db: "app", coll: "orders"` | Atlas Stream Processing synchronizes the one named collection. |
| Explicit list | `db: "app", coll: ["a", "b", "c"]` | Atlas Stream Processing synchronizes each collection in the list in the order in which you list them. |

### Traits of Multi-Collection Initial Sync

A multi-collection `initialSync` operation has the following traits:

- **Syncs collections in list order.** With `coll: ["a", "b", "c"]`, Atlas Stream Processing drains `a`'s partitions before `b`'s, and `b`'s before `c`'s, except where spillover fills idle capacity with partitions from a later collection.

- **Resumes without resyncing completed collections.** Atlas Stream Processing tracks which collections have finished, so a restarted stream processor skips completed collections and resumes any in-progress collection from its last synchronized document. It might redeliver documents it already emitted, so design your sink to handle repeated documents idempotently.

- **Exposes granular collection level sync progress.** `stats().stats.operatorStats[0].targetStats` reports sync progress and document counts for each target collection. To learn more, see [Verify Initial Sync Progress.](https://www.mongodb.com/docs/atlas/atlas-stream-processing/multi-collection-initial-sync.md#std-label-atlas-sp-multi-collection-initial-sync-verify)

### Behavior Differences to Consider

Multi-collection `initialSync` differs from single-collection sync in ways that affect how you configure and operate your stream processor:

- **Parallelism is global, not per collection.** Setting `initialSync.parallelism` caps the number of concurrent partition reads across the entire set of target collections, not per collection.

- **Catchup scope follows source scope.** After the sync phase, Atlas Stream Processing opens a change stream matching the source's scope: a database-level change stream for a whole-database source, a cluster-level change stream for a whole-cluster source, or a filtered change stream for an explicit list of collections. You don't need to configure this separately.

- **A collection sync failure fails the whole processor.** As with single-collection sync, if Atlas Stream Processing fails to synchronize any one collection in the list, the entire stream processor fails.

- **Some source-side configuration doesn't replicate to the
  destination.** `initialSync` doesn't replicate indexes, collection options such as capped collections, time series collections, validators, or default collation, or view definitions. Create these on the destination before you start the stream processor. Atlas Stream Processing also doesn't discover collections that you create after the stream processor starts; the set of collections to synchronize is fixed at that point.

## Configure Multi-Collection Initial Sync

The following procedure uses two collections from the [sample\_analytics](https://www.mongodb.com/docs/manual/sample-data/sample-analytics.md#std-label-sample-analytics) dataset: `customers`, which holds customer profiles, and `accounts`, which holds each customer's financial accounts. The procedure configures a stream processor that copies both collections from one Atlas cluster to another, then keeps them current as documents change.

### Prerequisites

Before you complete this procedure, you must have:

- A [stream processing workspace](https://www.mongodb.com/docs/atlas/atlas-stream-processing/manage-processing-instance.md#std-label-atlas-sp-manage-spi) with Atlas connections to both the source and destination clusters. To add a connection, see [Manage Connections.](https://www.mongodb.com/docs/atlas/atlas-stream-processing/manage-connection-registry.md#std-label-atlas-sp-manage-connections)

- A processor tier that supports the number of collections you plan to list. To learn more, see [Resource Allocation.](https://www.mongodb.com/docs/atlas/atlas-stream-processing/tier-guide.md#std-label-atlas-sp-tier-guide-resources)

Run the commands in this section against your cluster. To connect, see [Connect to a Cluster via mongosh.](https://www.mongodb.com/docs/atlas/mongo-shell-connection.md#std-label-connect-mongo-shell)

Complete the following steps to prepare the source collections:

1. Load the sample dataset.

   This procedure uses the `customers` and `accounts` collections from the [sample\_analytics](https://www.mongodb.com/docs/manual/sample-data/sample-analytics.md#std-label-sample-analytics) dataset. To learn how to load sample data, see [Import Sample Data Into Your Atlas Deployment.](https://www.mongodb.com/docs/atlas/import/load-sample-data.md#std-label-load-sample-data)

2. Enable change stream pre- and post-images.

   Enable pre- and post-images on the `customers` and `accounts` collections, because the procedure's [`$source`](https://www.mongodb.com/docs/atlas/atlas-stream-processing/sp-agg-source.md#mongodb-pipeline-pipe.-source) stage sets `fullDocument: "required"`. Run the following commands against the cluster:

   ```javascript
   db.getSiblingDB("sample_analytics").runCommand({
      collMod: "customers",
      changeStreamPreAndPostImages: { enabled: true }
   })
   ```

   **Output:**

   ```javascript
   { ok: 1, ... }
   ```

   ```javascript
   db.getSiblingDB("sample_analytics").runCommand({
      collMod: "accounts",
      changeStreamPreAndPostImages: { enabled: true }
   })
   ```

   **Output:**

   ```javascript
   { ok: 1, ... }
   ```

   To learn more about `fullDocument: "required"`, see [MongoDB Collection Change Stream.](https://www.mongodb.com/docs/atlas/atlas-stream-processing/sp-agg-source.md#std-label-atlas-sp-agg-source-syntax-coll)

### Procedure

Select the Atlas UI or [`mongosh`](https://www.mongodb.com/docs/mongodb-shell.md#mongodb-binary-bin.mongosh) to configure the stream processor.

1. Go to the Stream Processing page for your project.

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

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

2. Open the Visual Builder.

   Click Create stream processor.

   Select the Visual Builder.

3. Configure the source with an explicit collection list.

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

   In the JSON (Javascript Object Notation) text box, configure the [`$source`](https://www.mongodb.com/docs/atlas/atlas-stream-processing/sp-agg-source.md#mongodb-pipeline-pipe.-source) stage:

   ```json
   {
     "$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.

4. Add a stage that prepares each event for the sink.

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

   In the JSON (Javascript Object Notation) text box, add a [`$replaceRoot`](https://www.mongodb.com/docs/manual/reference/operator/aggregation/replaceRoot.md#mongodb-pipeline-pipe.-replaceRoot) stage:

   ```json
   {
     "$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.

5. Configure the sink with dynamic namespace routing.

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

   In the JSON (Javascript Object Notation) text box, configure the [`$merge`](https://www.mongodb.com/docs/manual/reference/operator/aggregation/merge.md#mongodb-pipeline-pipe.-merge) stage:

   ```json
   {
     "$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.

6. Enter processor details.

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

   Select the tier for the stream processor. To choose a tier for your workload, see [Atlas Stream Processing Tier Selection Guide.](https://www.mongodb.com/docs/atlas/atlas-stream-processing/tier-guide.md#std-label-atlas-sp-tier-guide)

7. Create the stream processor.

   Click Create stream processor.

8. 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.](https://www.mongodb.com/docs/atlas/atlas-stream-processing/manage-stream-processor.md#std-label-atlas-sp-manage-processor)

1) Connect to your stream processing workspace.

   Use the connection string associated with your stream processing workspace to connect using [`mongosh`.](https://www.mongodb.com/docs/mongodb-shell.md#mongodb-binary-bin.mongosh)

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

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

   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`](https://www.mongodb.com/docs/manual/reference/built-in-roles.md#mongodb-atlasrole-atlasAdmin) role:

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

   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.

2) Define the source stage.

   Declare a variable and assign it a [`$source`](https://www.mongodb.com/docs/atlas/atlas-stream-processing/sp-agg-source.md#mongodb-pipeline-pipe.-source) stage specification that lists both collections and enables `initialSync`:

   ```javascript
   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.

3) Define a stage that prepares each event for the sink.

   Declare a variable and assign it a [`$replaceRoot`](https://www.mongodb.com/docs/manual/reference/operator/aggregation/replaceRoot.md#mongodb-pipeline-pipe.-replaceRoot) stage specification:

   ```javascript
   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.

4) Define the sink stage.

   Declare a variable and assign it a [`$merge`](https://www.mongodb.com/docs/manual/reference/operator/aggregation/merge.md#mongodb-pipeline-pipe.-merge) stage specification that routes each event by its source namespace:

   ```javascript
   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.

5) Create the stream processor.

   Use the [`sp.createStreamProcessor()`](https://www.mongodb.com/docs/manual/reference/method/sp.createStreamProcessor.md#mongodb-method-sp.createStreamProcessor) method to create a stream processor named `replicate_analytics_sp` from the stages you defined:

   ```javascript
   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.](https://www.mongodb.com/docs/atlas/atlas-stream-processing/tier-guide.md#std-label-atlas-sp-tier-guide)

6) Start the stream processor.

   Use the [`sp.processor.start()`](https://www.mongodb.com/docs/manual/reference/method/sp.processor.start.md#mongodb-method-sp.processor.start) method to start `replicate_analytics_sp`:

   ```javascript
   sp.replicate_analytics_sp.start()
   ```

   **Output:**

   ```javascript
   { 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.](https://www.mongodb.com/docs/atlas/atlas-stream-processing/manage-stream-processor.md#std-label-atlas-sp-manage-processor)

### Verify Initial Sync Progress

After you start the stream processor, call `sp.<processor-name>.stats()` with the `verbose` option to view sync progress for each target collection. The source operator's entry in `stats.operatorStats` reports a `targetStats` array with one element per collection:

```javascript
sp.replicate_analytics_sp.stats(
  { options: { verbose: true } }
).stats.operatorStats[0].targetStats
```

**Output:**

```javascript
[
  {
    db: 'sample_analytics',
    coll: 'customers',
    inputMessageCount: Long('500'),
    inputMessageSize: 672326,
    initialSync: {
      estimatedDocs: Long('500'),
      copiedDocs: Long('251'),
      status: 'in_progress'
    }
  },
  {
    db: 'sample_analytics',
    coll: 'accounts',
    inputMessageCount: Long('1746'),
    inputMessageSize: 1883681,
    initialSync: {
      estimatedDocs: Long('1746'),
      copiedDocs: Long('1746'),
      status: 'completed'
    }
  }
]
```

Each target's `initialSync.status` field cycles through `pending`, `in_progress`, and `completed` as Atlas Stream Processing syncs that collection, or reports `failed` if the sync fails. `estimatedDocs` and `copiedDocs` report the estimated and completed document counts for that one collection.

To view total progress across every target collection, use the source operator's own `initialSync` field, which sums `estimatedDocs` and `copiedDocs` across all collections:

```javascript
sp.replicate_analytics_sp.stats(
  { options: { verbose: true } }
).stats.operatorStats[0].initialSync
```

**Output:**

```javascript
{ estimatedDocs: Long('2246'), copiedDocs: Long('1997') }
```

## Examples

The preceding procedure replicates two collections between Atlas clusters. The following example applies the same pipeline to a different set of collections. Only the `db` and `coll` values in the [`$source`](https://www.mongodb.com/docs/atlas/atlas-stream-processing/sp-agg-source.md#mongodb-pipeline-pipe.-source) stage change.

### Replicate a database to another cluster

#### Combine multi-collection initial sync with dynamic namespace routing to replicate a set of collections.

The pipeline synchronizes existing documents in `orders` and `customers`, then continues to replicate changes to both collections as they occur. The aggregation has three stages:

1. The [`$source`](https://www.mongodb.com/docs/atlas/atlas-stream-processing/sp-agg-source.md#mongodb-pipeline-pipe.-source) stage synchronizes `orders` and `customers` on `<source-connection-name>`, then opens a change stream scoped to those two collections.

2. The [`$replaceRoot`](https://www.mongodb.com/docs/manual/reference/operator/aggregation/replaceRoot.md#mongodb-pipeline-pipe.-replaceRoot) stage selects the document key for delete events and the full document for every other event. Because metadata isn't part of the document body, `stream.source.*` metadata survives this stage.

3. The [`$merge`](https://www.mongodb.com/docs/manual/reference/operator/aggregation/merge.md#mongodb-pipeline-pipe.-merge) stage writes each event to the matching collection on `<destination-connection-name>`, using the source event's namespace metadata to route the write and the operation type to choose between inserting, replacing, or deleting the destination document.

```javascript
sp = db.createStreamProcessor("replicate-app-db", [
  { $source: {
      connectionName: "<source-connection-name>",
      db: "app",
      coll: ["orders", "customers"],
      config: { fullDocument: "required" },
      initialSync: { enable: true }
  } },
  { $replaceRoot: {
      newRoot: {
        $cond: {
          if:   { $eq: [{ $meta: "stream.source.operationType" }, "delete"] },
          then: { $meta: "stream.source.documentKey" },
          else: "$fullDocument"
        }
      }
  } },
  { $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"
      } }
  } }
]);

sp.start();
```

Before you start this stream processor, create any indexes, collection options, or views that `orders` and `customers` need on `<destination-connection-name>`. `initialSync` doesn't replicate these.

## Additional Information

To learn more about the stages and concepts in this guide, see the following resources:

- To learn about the `$source` stage and its `initialSync` fields, see [MongoDB Collection Change Stream.](https://www.mongodb.com/docs/atlas/atlas-stream-processing/sp-agg-source.md#std-label-atlas-sp-agg-source-syntax-coll)

- To learn how processor tier affects the number of collections you can synchronize, see [Resource Allocation.](https://www.mongodb.com/docs/atlas/atlas-stream-processing/tier-guide.md#std-label-atlas-sp-tier-guide-resources)

- To learn how Atlas Stream Processing checkpoints an in-progress sync, see [Checkpoints.](https://www.mongodb.com/docs/atlas/atlas-stream-processing/architecture.md#std-label-atlas-sp-checkpoints)

- To learn how to create, start, stop, and monitor stream processors, see [Develop and Manage Stream Processors.](https://www.mongodb.com/docs/atlas/atlas-stream-processing/manage-stream-processor.md#std-label-atlas-sp-manage-processor)