MongoDB

Extractor

MongoDB is a popular document database that stores data as flexible, JSON-like documents grouped into collections. This connector extracts documents from a MongoDB or AWS DocumentDB database, either by incrementally reading each collection or by capturing inserts, updates, replaces and deletes as they happen through MongoDB change streams.

At a glance

Property	Value
Authentication	MongoDB connection string, or an AWS Secrets Manager-style DocumentDB credential
Sync type	Incremental (default) or log-based (CDC) via change streams
Streams	One per collection, discovered dynamically from your database
Custom queries	Not supported
Supported sources	Self-hosted MongoDB, MongoDB Atlas, AWS DocumentDB

What you can sync

Every collection in the database you configure becomes a stream. Each record wraps the source document in a single document object alongside some record metadata (such as object_id, namespace and replication_key).

Choosing a sync type

• Incremental (default). Reads documents directly from each collection and uses the _id field as the replication key. It needs no special infrastructure and works against any MongoDB deployment. Because _id is the key, it picks up new documents but not edits to existing ones, and deletes on the source aren't seen. A document inserted during a sync, or with an _id lower than the last one synced, isn't guaranteed to appear on the next run; use log-based replication, or a periodic full refresh, if you need that.
• Log-based (CDC). Opens a change stream on each collection and emits insert, update, replace and delete events with the full document (deletes carry only the _id). It requires a replica set or sharded cluster.

Prerequisites

• A MongoDB connection string (or DocumentDB credentials) for the database you want to sync. The username and password inside the connection string must be URL-encoded; the connector doesn't encode them for you.
• Read access to the collections you're syncing: the read role on the database (for example read@my_database) or readAnyDatabase@admin. Both also include the changeStream action that log-based replication needs.
• For log-based replication only: a replica set or sharded cluster. On AWS DocumentDB, change streams must be enabled per collection (see AWS DocumentDB).
• Network access from where your pipeline runs to your database. If your database restricts inbound connections, allow the address your pipeline runs from.

Setup

In MongoDB

1. Create a dedicated user for the connector and grant it read access:

   use admin
   db.createUser({
     user: "tap_mongodb",
     pwd: "<password>",
     roles: [{ role: "read", db: "<database>" }]
   })

2. For log-based replication, make sure the connection string includes the replica set name, for example mongodb://host:27017/?replicaSet=rs0.

AWS DocumentDB

DocumentDB requires TLS with the cluster's CA bundle, and change streams must be enabled on each collection you want to read log-based:

db.adminCommand({
  modifyChangeStreams: 1,
  database: "<database>",
  collection: "<collection>",
  enable: true
})

Alternatively, set allow_modify_change_streams to true and the connector runs that command for you. Doing so may incur new costs in DocumentDB.

Provide either a connection string such as:

mongodb://<user>:<password>@<cluster>.<region>.docdb.amazonaws.com:27017/?replicaSet=rs0&tls=true&tlsCAFile=rds-combined-ca-bundle.pem

or a documentdb_credential_json_string (the JSON format used by AWS Secrets Manager for DocumentDB secrets).

In Meltano Cloud

1. Add a MongoDB source and set database plus either mongodb_connection_string or documentdb_credential_json_string.

2. Optionally set filter_collections to limit discovery to specific collections.

3. Select the streams you want and set their replication method. As plain Meltano project config in meltano.yml, incremental is the default:

   plugins:
     extractors:
       - name: tap-mongodb
         config:
           database: my_database

   For log-based (CDC), use a plugin that inherits from the base tap:

   plugins:
     extractors:
       - name: tap-mongodb-cdc
         inherit_from: tap-mongodb
         config:
           database: my_database
           mongodb_connection_string: mongodb://localhost:27017/?replicaSet=rs0&directConnection=true
         metadata:
           '*':
             replication-method: LOG_BASED
             replication-key: replication_key

4. Commit and push your changes, and deploy your workspace from the settings page.

Available streams

Each collection in the configured database becomes a stream. Use prefix to add a common prefix to every stream name, and filter_collections to restrict discovery.

Log-based (CDC) output

Each log-based record includes the operation, when it happened, where it came from, and the document:

{
  "operation_type": "insert",
  "cluster_time": "2025-11-19T20:28:33+00:00",
  "namespace": {
    "database": "test",
    "collection": "users"
  },
  "document": {
    "_id": "...",
    "name": "John Doe",
    "email": "john@example.com"
  },
  "replication_key": "82691E3EFE000000012B0429296E1404"
}

operation_type is one of insert, update, replace or delete. Use operation_types to limit which are emitted.

Faster loading with Arrow

For large collections, incremental streams can write Apache Arrow batch files instead of one message per document by setting batch_config.encoding.format to arrow. Log-based streams always emit one message per event. See Arrow BATCH Support for details and compatible targets.

Settings

Setting	Type	Description
database	string	Database to extract records from. Required.
mongodb_connection_string	string	MongoDB connection string. The username and password in it must be URL-encoded.
documentdb_credential_json_string	string	JSON credential in the AWS Secrets Manager format for DocumentDB.
documentdb_credential_json_extra_options	string	JSON key-value pairs added to the connection string options when using documentdb_credential_json_string, for example {"tls":"true","tlsCAFile":"my-ca-bundle.pem"}.
datetime_conversion	string	How to handle dates outside years 1 to 9999. Default datetime, which fails on such dates. See the PyMongo docs for alternatives.
prefix	string	Prefix added to the name of each stream.
filter_collections	array	Collections to discover (default all). Case-insensitive.
start_date	date	Start date for incremental replication. Default 1970-01-01. Ignored in log-based mode.
add_record_metadata	boolean	Add _sdc metadata fields to records. Default false.
allow_modify_change_streams	boolean	On DocumentDB, enable change streams on collections that don't have them yet. Default false. May incur new costs.
operation_types	array	Change stream operation types to include. Default create, delete, insert, replace, update.
change_stream_pipeline	array	Aggregation pipeline stages appended to the change stream, for example to shrink large events server-side. Log-based mode only.
batch_config	object	Set encoding.format: arrow to write Arrow batch files from incremental streams.
document_layout	string	object (default) emits each document as a single document object. flat emits each field declared in the stream's schema as its own column.

Troubleshooting

• "Change streams not available": log-based replication needs a replica set or sharded cluster. Check that the connection string includes replicaSet=<name>. On DocumentDB, verify change streams are enabled on the collection, or set allow_modify_change_streams.
• "Resume point no longer in oplog": too much time passed between syncs. Clear the stream's state to start fresh, and consider syncing more frequently.
• No events captured: change streams only see events that happen after they open. Run an initial sync first, then make changes and sync again.
• bson.errors.InvalidBson on dates: a document contains a date outside years 1 to 9999. Set datetime_conversion to a value that handles out-of-range dates.

Need help?

If a stream or field you need isn't listed here or the connector doesn't work as expected, file it through the usual Meltano support channel.