ETL with MongoDB

Etlworks reads from MongoDB collections, writes into them, and captures document-level changes via CDC. This article covers every supported pattern with step-by-step setup.

MongoDB stores data in BSON — a binary form of JSON that supports nested documents, arrays, and variable schemas. Etlworks can either flatten MongoDB documents into a tabular shape or preserve their nested structure, depending on where the data is going.

Extract data from a MongoDB collection

Two paths: CSV when the destination is relational and you want flat data, JSON when you need to preserve the nested document structure.

Path 1: extract as CSV (flat / tabular)

Use CSV when the destination is a relational database or a data warehouse (Snowflake, Redshift, Synapse, …). Nested fields are emitted as JSON-encoded strings so they can be loaded into a single column.

1. Create a MongoDB connection.
2. Create a CSV format. Defaults are fine for most cases.
3. Create a flow where the source is NoSQL. Type nosql to in Select Flow Type and pick the flow that matches your destination.

   

4. Set FROM to the MongoDB connection + CSV format. Enter the MongoDB collection name; wildcards are supported (e.g. customer*). If FROM uses a wildcard, TO can use one too.

   

5. Set TO to the destination connection + format (if needed) and the destination object name.
6. Add mapping or transformations as needed. For filtering and shaping, use SQL with MongoDB.

Path 2: extract as JSON (preserve nesting)

Use JSON when the destination should keep the nested document structure — for example, exporting as JSON files, or streaming into Kafka / Kinesis.

1. Create a MongoDB connection.
2. Create a JSON format. Defaults are fine.
3. Create a flow where the source is NoSQL. Type nosql to and pick the flow type that matches your destination.
4. Set FROM to the MongoDB connection + JSON format. Enter the MongoDB collection name; wildcards are supported.

   

5. Set TO to the destination connection + format and object name.
6. Add mapping or transformations. For filtering and shaping, use SQL with MongoDB.

Filtering during extraction

Use SQL with MongoDB for filtering and shaping — better readability and long-term support than the older MongoDB-query-in-collection-name approach.

Deprecated: the legacy collection.{filter} syntax in the FROM field (e.g. customers.{"STORE_ID":1}) still works for backward compatibility but is not recommended for new flows. See the MongoDB query language reference.

Automatically partition a large MongoDB collection

Starting with Etlworks 9.8.5, a regular ETL flow with a MongoDB/NoSQL source can automatically split one large collection into non-overlapping _id ranges and read them in parallel. It is disabled by default, does not apply to the MongoDB Document connection, and a customer-managed Integration Agent must also run the matching 9.8.5 version. It can improve throughput when one large collection is the source-side bottleneck and the source, Integration Agent, and destination all have capacity — it is not a guaranteed speedup.

Configuration:

1. In the transformation settings, enable Automatically partition MongoDB collections by _id.
2. Set the flow's Maximum Number of Parallel Threads to a value greater than 1.
3. Leave Automatic partition multiplier at 1 initially. The requested range count is approximately Maximum Number of Parallel Threads × Automatic partition multiplier. Raising the multiplier creates more queued ranges for better balance when ranges finish unevenly — it does not add worker threads, and it adds planning and scheduling overhead.
4. For a sharded cluster, partitioning stays disabled by default: _id range queries can turn into scatter/gather queries across shards when _id is not aligned with the shard key, making performance worse. To opt in, add automatic.partition.sharded.enabled=true to the MongoDB connection's Other Parameters — and only after testing with representative data.

Note: Use Parallel Threads when processing sources by a wildcard controls concurrency across collections matched by a wildcard. It is separate from splitting one collection into _id ranges and is not required merely to partition one collection. With a wildcard source, collection-level concurrency and range-level concurrency share the flow's maximum-thread ceiling.

How ranges are planned. Etlworks counts the matching documents, then scans only sorted, projected _id values to calculate approximately even boundaries and build non-overlapping typed ranges (each lower-exclusive and upper-inclusive; the first has no lower bound and the last no upper bound). Supported _id BSON types are ObjectId, string, numeric types, date, timestamp, and binary. The planning is not free — an exact count plus a full sorted _id scan runs before any documents are read — so use this for one or a few genuinely large collections, not small ones. The process only reads MongoDB; it never modifies or partitions the source collection.

Source Query behavior:

• No Source Query — the entire collection is eligible for partitioning.
• JSON MongoDB filter — Etlworks combines your filter with each generated _id range using $and, without changing its meaning.
• SQL-style MongoDB Source Query, JavaScript, or another non-JSON query — the flow remains valid, but Etlworks falls back to a single cursor, because it cannot safely compose the query with BSON _id ranges. This matters here: the filtering section above recommends SQL for readability, and a SQL-style query is not automatically partitioned in 9.8.5. To combine filtering with partitioning, express the filter as a JSON MongoDB query.
• Max Records, Before Prepare Source Query, or unresolved runtime tokens also cause a safe one-cursor fallback.

Tradeoffs and operational guidance:

• Ranges execute as independent transformations and may write to the destination concurrently — confirm the destination can handle the same concurrency you configure for the source. If the destination is already the bottleneck, more MongoDB read concurrency will not improve end-to-end throughput.
• Document (output) order is not preserved when ranges are read concurrently.
• For file destinations, make sure each partition resolves to a unique output file name or use the destination connection's Add Suffix When Creating Files in Transformation option — otherwise parallel partitions can overwrite or corrupt the same file.
• Empty or one-document collections, duplicate boundaries, missing _id values, mixed or unsupported _id types, planning failures, and non-composable queries all fall back safely to one cursor, and the flow log records the reason.
• More workers increase source CPU, I/O, network traffic, connection-pool usage, Integration Agent resource consumption, and destination concurrency. Benchmark end-to-end throughput, not just MongoDB read speed.

Related: this is the ETL counterpart of the CDC feature that chunks a collection during snapshots — see Snapshot Management → MongoDB collections and Optimizing CDC Snapshots for Large Databases. The two features are configured independently; neither is related to Debezium incremental snapshots.

Extract specific documents (MongoDB Document connection)

When you want one document (or a small group) instead of a full collection scan, use a MongoDB Document connection.

1. Create the MongoDB Document connection.
2. Create a JSON format.
3. Create a flow where the source is NoSQL. Type nosql to.
4. Set FROM to the MongoDB Document connection + JSON format. Enter the source identifier.
5. Set TO to the destination connection + format and object name.
6. In Mapping, enter SQL in Source query to narrow down what's read. See Using SQL with MongoDB.

Capture changes in MongoDB (CDC)

Etlworks ships a native MongoDB CDC connector that monitors replica sets and sharded clusters. The connector captures document-level changes (insert / update / delete) and emits them as events. Use it for real-time pipelines, event streaming, and incremental replication.

Reference:

• Change Data Capture (CDC) from transaction log
• MongoDB-specific CDC configuration

Load data into MongoDB

Two patterns — pick based on whether each source row becomes its own document, or the whole dataset goes in as one document.

Pattern 1: one source row = one MongoDB document

Use this for tabular sources (database tables, CSV files, JSON arrays).

1. Create a MongoDB connection. Set What to do with the existing document when writing:
   • insert — inserts new documents without checking for duplicates. Fastest. Use for initial loads only.
   • update — updates only the fields present in the payload; other fields are left unchanged.
   • replace — replaces the whole document. Slowest, but guarantees full replacement.
2. Create a JSON format. If the source is array-shaped (table, CSV, JSON array):
   • Set First Object Type to variable.
   • Set Root Object Name to any valid name (required).

   

3. Create a flow where the destination is NoSQL. Type to nosql and pick the flow type that matches your source.

   

4. Set FROM to the source connection + format and the source object name.
5. Set TO to the MongoDB connection + JSON format.
6. Enter the MongoDB collection name in TO. Optionally specify a unique field for the MongoDB _id using collection_name.unique_field_name — e.g. customers.CUSTOMER_ID.

   

7. Add mapping or transformations as needed.

Pattern 2: whole dataset = one MongoDB document

Use this for storing an Excel worksheet, a full JSON array, or a configuration payload as a single document.

1. Create a MongoDB Document connection.
2. Create a JSON format. If the source is an array, set First Object Type = variable and pick a meaningful Root Object Name.
3. Create a flow where the destination is NoSQL. Type to nosql.
4. Set FROM to the source connection + format and the source object name.
5. Set TO to the MongoDB Document connection + JSON format.
6. Enter the MongoDB document _id in TO.
7. Add mapping or transformations as needed.

Improve load performance

Three settings dominate write throughput when loading into MongoDB:

Parameter	What it does
Batch Size	Documents per write round-trip. Higher batch sizes mean fewer network calls and faster loads, especially for large datasets.
Write Concern	The acknowledgment level MongoDB requires for each write. Lower acknowledgment = faster writes but weaker durability. Use minimal acknowledgment for initial loads where you can re-run on failure.
What to do with the existing document when writing	insert is fastest (no lookup); use it for the initial load. Switch to update or replace for subsequent runs to handle duplicates correctly. replace is slowest because the whole document is rewritten every time.

Convert timestamp strings to ISODate

MongoDB queries and indexes work better against native ISODate values than against timestamp strings. The MongoDB connector can convert matching strings to ISODate on write — off by default.

1. Enable Timestamp Conversion on the MongoDB connection.
2. Set a Regex Pattern that identifies timestamp strings. If you don't provide one, the connector uses the default ISO 8601 pattern: ^\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}(\.\d{3})?(Z|[+-]\d{2}:\d{2})$.
3. Any string matching the pattern is converted to ISODate on write.