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 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 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 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 overall sync progress. stats().stats.initialSync reports the combined sync progress and document count across all target collections. To learn more, see Verify Initial Sync Progress.

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 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 with Atlas connections to both the source and destination clusters. To add a connection, see Manage Connections.
A processor tier that supports the number of collections you plan to list. To learn more, see Resource Allocation.

Run the commands in this section against your cluster. To connect, see Connect to a Cluster via mongosh.

Complete the following steps to prepare the source collections:

Load the sample dataset.

This procedure uses the customers and accounts collections from the sample_analytics dataset. To learn how to load sample data, see Import Sample Data Into Your Atlas Deployment.

Enable change stream pre- and post-images.

Enable pre- and post-images on the customers and accounts collections, because the procedure's $source stage sets fullDocument: "required". Run the following commands against the cluster:

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

{ ok: 1, ... }

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

{ ok: 1, ... }

To learn more about fullDocument: "required", see MongoDB Collection Change Stream.

Procedure

Select the Atlas UI or mongosh to configure the stream processor.

Interface

Verify Initial Sync Progress

After you start the stream processor, run sp.<processor-name>.stats().stats.initialSync to check the overall sync progress across all target collections:

sp.replicate_analytics_sp.stats().stats.initialSync

{ progress: 1, numInputDocuments: Long('2246') }

progress reports whether initial sync has completed. numInputDocuments reports the total number of documents that Atlas Stream Processing has ingested from the target collections so far.

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 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:

The $source stage synchronizes orders and customers on <source-connection-name>, then opens a change stream scoped to those two collections.
The $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.
The $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.

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.
To learn how processor tier affects the number of collections you can synchronize, see Resource Allocation.
To learn how Atlas Stream Processing checkpoints an in-progress sync, see Checkpoints.
To learn how to create, start, stop, and monitor stream processors, see Develop and Manage Stream Processors.

Back

Build Streaming Materialized Views

Architecture