Replicate Data from MongoDB®

Prerequisites

Source Database Requirements

The replication feature uses MongoDB® change streams, and it works only with replica sets. A MongoDB® user must have the following roles (privileges) to replicate the collections in SingleStore:

Read the admin database for the operation log (oplog)
Read the config database in the configuration server
listDatabases privilege
Cluster-wide find and changeStream privileges

You can provide the privileges or assign roles to the MongoDB® user using the MongoDB® Atlas UI or MongoDB® commands. For example:

Using the UI for MongoDB® Atlas:
1. On the MongoDB® Cloud dashboard, select Projects > <your_project>.
2. In the left navigation pane, under Security, select Database Access.
3. On the Database Users page, select Edit for the MongoDB® user used to connect to the MongoDB® instance.
4. On the Edit User dialog, under Database User Privileges, select Specific Privileges > Add Specific Privilege.
5. Add the following privileges:
  - readAnyDatabase
  - read for the config database
  - readWrite for the singlestore database
6. Select Update User.

Using MongoDB® commands for self-managed deployments: The following commands create a role and then assign the role to a user.

MongoDB

db.adminCommand({ createRole: 'cdcRole', 
  privileges: [
    {'resource': {'cluster': true}, 'actions': ['find', 'changeStream']}
  ], 
  roles: [
    {'role': 'read', 'db': 'admin'},
    {'role': 'read', 'db': 'local'},
    {'role': 'read', 'db': 'config'},
    {'role': 'readWrite', 'db': 'singlestore'}
]})

db.adminCommand({ createUser: 'cdcUser', 
  pwd: 's2mongoCDC12', 
  roles: ['cdcRole']
})

Refer to Database Commands for more information.

Replicate MongoDB® Collections using SQL

To replicate your existing MongoDB® collections to your SingleStore database using SQL commands via Change Data Capture (CDC) pipelines, perform the following tasks:

Ensure that the Prerequisites are met.
(Optional) Create a link to the MongoDB® instance. Refer to CREATE LINK for more information. You can also specify the link configuration and credentials in the CONFIG/CREDENTIALS clause of the CREATE {TABLES | TABLE} AS INFER PIPELINE SQL statement instead of creating a link.
Create the required table(s), stored procedure(s), and pipeline(s) using the CREATE {TABLES | TABLE} AS INFER PIPELINE SQL statement. Refer to Syntax for more information. You can either replicate the MongoDB® collections as is or apply custom transformations.
Note
Before restarting the INFER PIPELINE operation, delete all the related artifacts.
SQL
DROP TABLE <target_table_name>;
DROP PIPELINE <pipeline_name>;
DROP PROCEDURE <procedure_name>;
Once all the components are configured, start the pipelines.
- To start all the pipelines, run the START ALL PIPELINES SQL statement.
- To start a specific pipeline, run the START PIPELINE <pipeline_name> SQL statement. By default, the pipeline is named <source_db_name>.<table_name>.

For more information, refer to the relevant section on this page:

Replication Strategies
CDC Snapshot Strategies
Configure Ingestion Speed Limit using Engine Variables

Replicate MongoDB® Collections Example

The following example shows how to replicate MongoDB® collections without any custom transformations. This example uses the LINK clause to specify the MongoDB® Atlas endpoint connection configuration.

Create a link to the MongoDB® endpoint, for example, the primary node of a MongoDB® Atlas cluster.

SQL

CREATE LINK <linkname> AS MONGODB
  CONFIG '{"mongodb.hosts":"<Hostname>",
    "collection.include.list": "<Collection list>",
    "mongodb.ssl.enabled":"true",
    "mongodb.authsource":"admin"}'
  CREDENTIALS '{"mongodb.user":"<username>",
    "mongodb.password":"<password>"}';

Create tables, pipelines, and stored procedures in SingleStore based on the inference from the source collections.
SQL
```
CREATE TABLES AS INFER PIPELINE AS LOAD DATA 
  LINK <linkname> '*' FORMAT AVRO;
```
Once the link and the tables are created, run the following command to start all the pipelines and begin the data replication process:
SQL
```
## Start pipelines
START ALL PIPELINES;
```

Syntax

SQL

CREATE TABLE <table_name> 
  AS INFER PIPELINE AS LOAD DATA 
  { LINK <link_name> "<source_db>.<source_collection>" |
    MONGODB "<source_db>.<source_collection>" CONFIG <config_json> CREDENTIALS <credentials_json> }
  FORMAT AVRO;

SQL

CREATE TABLES 
  AS INFER PIPELINE AS LOAD DATA 
  { LINK <link_name> "*" |
    MONGODB '*' CONFIG <config_json> CREDENTIALS <credentials_json> }
  FORMAT AVRO;

`CREATE TABLE ... AS INFER PIPELINE` Behavior

The CREATE TABLE <table_name> AS INFER PIPELINE statement,

Connects to the MongoDB® servers using the specified LINK <link_name> or MONGODB <collection> CONFIG <conf_json> CREDENTIALS <cred_json> clause.
Discovers the available databases and collections filtered by collection.include.list.
Infers the schema of the collection and then creates a table (named <table_name>) in SingleStore using the inferred schema. You can also specify a table name that differs from the name of the source MongoDB® collection.
Creates a pipeline (named <table_name>) and stored procedure (named <table_name>_apply_changes) that maps the AVRO data structure to the SingleStore data structure.

`CREATE TABLES AS INFER PIPELINE` Behavior

The CREATE TABLES AS INFER PIPELINE statement creates a table for each collection in the source database using the same set of operations as the CREATE TABLE <table_name> AS INFER PIPELINE statement (specified above).

Arguments

<table_name>: Name of the table to create in the SingleStore database. You can also specify a table name that differs from the name of the source MongoDB® collection.
<link_name>: Name of the link to the MongoDB® endpoint. Refer to CREATE LINK for more information.
<collection>: Name of the source MongoDB® collection.
<config_json>: Configuration parameters, including the source MongoDB® configuration, in the JSON format. Refer to Parameters for supported parameters.
<credentials_json>: Credentials to use to access the MongoDB® database, in JSON format. For example:
```
CREDENTIALS '{"mongodb.password": "<password>", "mongodb.user": "<user>"}'
```
- mongodb.user: The name of the database user to use when connecting to MongoDB® servers.
- mongodb.password: The password to use when connecting to MongoDB® servers.

Parameters

The CREATE {TABLE | TABLES}, CREATE LINK. and CREATE AGGREGATOR PIPELINE statement supports the following parameters in the CONFIG clause:

mongodb.hosts: A comma-separated list of MongoDB® servers (nodes) in the replica set, in 'hostname:[port]' format.
- If mongodb.members.auto.discover is set to FALSE, you must prefix the 'hostname:[port]' with the name of the replica set in mongodb.hosts, e.g., rset0/svchost-xxx:27017. The first node specified in mongodb.hosts is always selected as the primary node.
- If mongodb.members.auto.discover is set to TRUE, you must specify both the primary and secondary nodes in the replica set in mongodb.hosts.
mongodb.members.auto.discover: Specifies whether the MongoDB® servers defined in mongodb.hosts should be used to discover all the members of the replica set. If disabled, the servers are used as is.
mongodb.ssl.enabled: Enables the connector to use SSL when connecting to MongoDB® servers.
mongodb.authsource: Specifies the database containing MongoDB® credentials to use as an authentication source. This parameter is only required when the MongoDB® instance is configured to use authentication with an authentication database other than admin.
collection.include.list: A comma-separated list of regular expressions that match fully-qualified namespaces (in databaseName.collectionName format) for MongoDB® collections to monitor. Collections excluded from the list are not monitored. By default, all collections are monitored, except for those in the local and admin databases. The list must contain the singlestore.signals collection, in addition to the specified regular expressions. Also, the MongoDB® user must have write permissions to the singlestore.signals collection.
signal.data.collection (Optional): A collection in the remote source that is used by SingleStore to generate special markings for snapshotting and synchronization. By default, this parameter is set to singlestore.signals_xxxxxx, where xxxxxx is an automatically generated character sequence. The default signal collection is in the database named singlestore. The MongoDB® user must have write permissions to this collection. Once the pipelines are started, any change to the value of this parameter is ignored, and the pipelines use the latest value specified before the pipelines started.
max.queue.size: Specifies the size of the queue inside the extractor process for records that are ready for ingestion. The default queue size is 1024. This variable also specifies the number of rows for each partition. Increasing the queue size results in an increase in the memory consumption by the replication process and you may need to increase the pipelines_cdc_java_heap_size.
max.batch.size: Specifies the maximum number of rows of data fetched from the remote source in a single iteration (batch). The default batch size is 512. max.batch.size must be lower than max.queue.size.
poll.interval.ms: Specifies the interval for polling of remote sources if there were no new records in the previous iteration in the replication process. The default interval is 500 milliseconds.
snapshot.mode: Specifies the snapshot mode for the pipeline. It can have one of the following values:
- "initial" (Default): Perform a full snapshot first and replay CDC events created during the snapshot. Then, continue ingestion using CDC.
- "incremental": Start the snapshot operation and CDC simultaneously.
- "never": Skip the snapshot, and ingest changes using CDC.
Refer to CDC Snapshot Strategies for more information.

Replication Strategies

Use one of the following methods to create the required components for data ingestion.

Replicate MongoDB® Collections As Is

To replicate or migrate MongoDB® collections as is, use the CREATE {TABLES | TABLE} AS INFER PIPELINE SQL statement. This method automatically creates the required tables, pipelines, and stored procedures. Refer to Syntax for more information.

Apply Transformations or Ingest a Subset of Columns

To apply transformations or ingest only a subset of collections, manually create the required tables, stored procedure, and pipelines:

Run the CREATE {TABLES | TABLE} AS INFER PIPELINE SQL statement to infer the schema of the MongoDB® collection(s) and automatically generate templates for the relevant table(s), stored procedure(s), and aggregator pipeline(s).
Use the automatically-generated templates as a base to create a new table(s), stored procedure(s), and pipeline(s) for custom transformations. To inspect the generated table(s), stored procedure(s), and pipeline(s), use the SHOW CREATE TABLE , SHOW CREATE PROCEDURE, and SHOW CREATE PIPELINE commands, respectively. After running the SHOW commands, you can drop the templates and then recreate the same components with custom transformations.

Using the automatically-generated templates:
1. Create table(s) in SingleStore with a structure that can store the ingested MongoDB® collection. Refer to CREATE TABLE for more information.
2. Create stored(s) procedure to map the MongoDB® collection to the SingleStore table and implement other transformations required. Refer to CREATE PROCEDURE for information on creating stored procedures.
3. Create pipeline(s) to ingest the MongoDB® collection(s) using the CREATE AGGREGATOR PIPELINE SQL statement.
  
  Refer to Parameters for a list of supported parameters. Refer to CREATE PIPELINE for the complete syntax and related information.
  
  Note: The CDC feature only supports AGGREGATOR pipelines.

Refer to Syntax for information on CREATE {TABLES | TABLE} AS INFER PIPELINE SQL statement.

CDC Snapshot Strategies

SingleStore supports the following strategies for creating snapshots:

Perform a full snapshot before CDC:

The pipeline captures the position in the oplog and then performs a full snapshot of the data. Once the snapshot is complete, the pipeline continues ingestion using CDC. This strategy is enabled by default. If the pipeline is restarted while the snapshot is in progress, the snapshot is restarted from the beginning.

To use this strategy, set "snapshot.mode":"initial" in the CONFIG JSON.

Requirement: The oplog retention period must be long enough to maintain the records while the snapshot is in progress. Otherwise, the pipeline will fail and the process will have to be started over.
CDC only:

The pipeline will not ingest existing data, and only the changes are captured using CDC.

To use this strategy, set "snapshot.mode":"never" in the CONFIG JSON.
Perform a snapshot in parallel to CDC:

The pipeline captures the position in the oplog and starts capturing the changes using CDC. In parallel, the pipeline performs incremental snapshots of the existing data and merges it with the CDC records. Although this strategy is slower than performing a full snapshot and then ingesting changes using CDC, it is more resilient to pipeline restarts.

To use this strategy, set "snapshot.mode":"incremental" in the CONFIG JSON.

Requirement: The oplog retention period must be long enough to compensate for unexpected pipeline downtime.
Manually perform the snapshot and capture changes using the CDC pipeline:
1. Create a pipeline and then wait for at least one batch of ingestion to capture the oplog position.
2. Stop the pipeline.
3. Snapshot the data using any of the suitable methods, for example, mongodump.
4. Restore the snapshot in SingleStore using any of the supported tools, for example mongorestore.
5. Start the CDC pipeline.
To use this strategy, set "snapshot.mode":"never" in the CONFIG JSON. This strategy provides faster data ingestion when the initial historical data is very large in size.

Requirement: The oplog retention period must be long enough to maintain the records while the snapshot is in progress. Otherwise, the pipeline will fail and the process will have to be started over.

Configure Ingestion Speed Limit using Engine Variables

Use the following engine variables to configure ingestion speed:

Variable Name	Description	Default Value
`pipelines_cdc_row_emit_delay_us`	Specifies a forced delay in row emission while migrating/replicating your tables (or collections) to your SingleStore databases. It can have a maximum value of `1000000`.	`1`
`pipelines_cdc_java_heap_size`	Specifies the JVM heap size limit (in MBs) for CDC-in pipelines.	`128`

In-Depth Variable Definitions

Use the pipelines_cdc_row_emit_delay_us engine variable to limit the impact of CDC pipelines on the master aggregator node. It specifies a forced delay in row emission during ingest. This variable can be set to a maximum value of 1000000. To disable the forced delay while ingestion, set this variable to 0.

Warning

Disabling the emit delay may result in excessive CPU usage on master aggregator nodes.

Use the max.queue.size parameter in the CONFIG JSON to control the ingestion speed. SingleStore recommends setting max.batch.size to half of max.queue.size. Increasing the queue size requires a larger Java heap limit, adjust the pipelines_cdc_java_heap_size engine variable accordingly. Query the INFORMATION_SCHEMA.PIPELINES_BATCHES_SUMMARY table for information on pipeline batch performance.

Limitations

SingleStore does not support data replication using CDC pipelines from MongoDB® Atlas standalone or serverless instances. Replication is only supported for replica set deployments.
Because MongoDB® time series collections do not support change streams, they cannot be replicated in SingleStore using CDC. Refer to Time Series Collection Limitations for more information.

Troubleshooting

If the CREATE {TABLES | TABLE} AS INFER PIPELINE SQL statement returns an error, run the SHOW WARNINGS command to view the reason behind the failure.

To view the status of all the pipelines, query the information_schema.pipelines_cursors table. Run the following SQL statement to display the status of the replication task:

SQL

SELECT SOURCE_PARTITION_ID, 
 EARLIEST_OFFSET,
 LATEST_OFFSET,
 LATEST_EXPECTED_OFFSET-LATEST_OFFSET as DIFF,
 UPDATED_UNIX_TIMESTAMP 
FROM information_schema.pipelines_cursors;

To view pipeline errors, run the following SQL statement:
SQL
```
SELECT * FROM information_schema.pipelines_errors;
```
If the replication process fails with an out of memory error in Java, increase the heap size using the pipeline_cdc_java_heap_size engine variable. Refer to Configure Ingestion Speed Limit using Engine Variables for more information.

Example

Example 1 - Create Tables with Different Names from the Source Collection

To create tables in SingleStore with names that differ from the name of the source MongoDB® collection, use the following syntax:

SQL

CREATE TABLE <new_table_name> AS INFER PIPELINE AS LOAD DATA 
LINK <link_name> '<source_db.source_collection>' FORMAT AVRO;

You can also use this command to import collections if a table with the same name already exists in SingleStore. Additionally, you can use this syntax to reimport a collection with a distinct table name.

Replicate Data from MongoDB®

On this page

Prerequisites

Source Database Requirements

Replicate MongoDB® Collections using SQL

Replicate MongoDB® Collections Example

Syntax

`CREATE TABLE ... AS INFER PIPELINE` Behavior

`CREATE TABLES AS INFER PIPELINE` Behavior

Arguments

Parameters

Replication Strategies

Replicate MongoDB® Collections As Is

Apply Transformations or Ingest a Subset of Columns

CDC Snapshot Strategies

Configure Ingestion Speed Limit using Engine Variables

In-Depth Variable Definitions

Limitations

Troubleshooting

Example

Example 1 - Create Tables with Different Names from the Source Collection

Was this article helpful?

On this page

Was this article helpful?

Replicate Data from MongoDB®

On this page

Prerequisites

Source Database Requirements

Replicate MongoDB® Collections using SQL

Replicate MongoDB® Collections Example

Syntax

CREATE TABLE ... AS INFER PIPELINE Behavior

CREATE TABLES AS INFER PIPELINE Behavior

Arguments

Parameters

Replication Strategies

Replicate MongoDB® Collections As Is

Apply Transformations or Ingest a Subset of Columns

CDC Snapshot Strategies

Configure Ingestion Speed Limit using Engine Variables

In-Depth Variable Definitions

Limitations

Troubleshooting

Example

Example 1 - Create Tables with Different Names from the Source Collection

Was this article helpful?

On this page

Was this article helpful?

`CREATE TABLE ... AS INFER PIPELINE` Behavior

`CREATE TABLES AS INFER PIPELINE` Behavior