MongoDB to S3: Parquet Files, Change Streams & Nested Documents Guide

MongoDB collections reach Amazon S3 as Parquet through four kinds of tool: Atlas Data Federation's $out stage run by Atlas Scheduled Triggers, AWS DMS, a Spark, Glue or dlt job, or a managed change data capture (CDC) service. Atlas Data Federation fits teams already on Atlas, DMS fits AWS-centred teams, a code job fits custom transforms, and a managed service fits many collections or several destinations.

Two decisions matter more than the tool: how nested documents and arrays become columns (structs, flattened columns or JSON text), and how updates and deletes show up in files that can't be edited. Each change becomes a new row, and readers keep the latest row per _id. Only change-stream methods record deletes, and they need an oplog that outlasts any outage. For backups, use mongodump or Atlas snapshot export instead; neither writes Parquet. Estuary is one managed option for this route: it reads MongoDB change streams into a durable collection and appends each batch of changes to Parquet files in S3 on an upload interval you set.

Choosing a MongoDB to S3 method

A time-window copy never sees a deleted document, and every method writes immutable files, so the rest of this guide assumes change streams plus a query that keeps the latest row per _id.

How MongoDB change streams become Parquet files in S3

Change streams are MongoDB's resumable API over the oplog. A pipeline to S3 notes the stream's position, scans each collection, then replays every change made since that position, so nothing written during the scan is lost.

• Events arrive in order with a resume token. An update carries only the changed fields unless the reader requests fullDocument: "updateLookup"; a delete carries only the document key.
• Files are written in batches under a prefix per collection and never rewritten, so an update or delete is one more row, not an edit.
• Reads go through a catalog table or a glob, with a view that keeps the newest row per _id and drops deletes.

Checkpoint the resume token only after a batch is in S3. A pipeline stopped for longer than the oplog's window can't resume and must re-read its collections.

MongoDB requirements

• Replica set or sharded cluster. Change streams don't exist on a standalone server, which can only be polled.
• A read user. MongoDB's built-in read role includes the find and changeStream actions a database-level stream needs:

js
use admin
db.createUser({ user: "s3_pipeline", pwd: passwordPrompt(),
                roles: [{ role: "read", db: "shop" }] })

• Unwatchable collections. Views, time series collections and the admin, local, config and system.* namespaces emit no change events. Poll them on an indexed field or leave them out.
• Network path: the Atlas IP access list, an SSH bastion, or a private endpoint (Atlas needs M10 or larger for those).

The oplog window, oldest entry to newest, must outlast the longest time the pipeline could be stopped. rs.printReplicationInfo() shows it as "log length start to end".

Amazon S3 requirements

• Writer credentials: an IAM user's access key or an assumable role, scoped to the bucket and prefix. s3:PutObject also covers multipart uploads. If the bucket encrypts with an AWS KMS key, multipart uploads also need kms:GenerateDataKey and kms:Decrypt. AWS DMS asks for more: s3:DeleteObject, s3:PutObjectTagging and s3:ListBucket.
• A lifecycle rule with AbortIncompleteMultipartUpload: parts of an upload that never finishes are billed as storage until it's aborted.
• One prefix per collection, such as mongodb/shop/orders/. Athena reads every file in the folder a table's LOCATION names.
• A catalog for SQL engines: an AWS Glue Data Catalog table, by hand or from a crawler, if you query with Athena.

Setting up MongoDB to S3 replication

1. Confirm change streams and size the oplog for your worst outage.
2. Create the read user and open the network path.
3. Prepare the bucket: a prefix per collection, writer permissions, the lifecycle rule.
4. Choose how nested data lands per collection (see the type-mapping section below).
5. Load, then stream. In AWS DMS that's a task that migrates existing data and replicates ongoing changes. In document mode, set ExtractDocId=true (required for CDC when the source uses multi-document transactions) and UseUpdateLookUp=true, so update rows carry the whole document.
6. Register the folder and add a current-state view. DMS full-load rows have no operation letter, so treat a missing one as an insert:

sql
CREATE OR REPLACE VIEW shop.orders_current AS
SELECT * FROM (
  SELECT c.*,
         ROW_NUMBER() OVER (PARTITION BY doc_id ORDER BY changed_at DESC) AS rn
  FROM shop.orders_changes c
)
WHERE rn = 1 AND COALESCE(op, 'I') <> 'D';

changed_at has to order every change to one _id: the source commit time (DMS adds it with TimestampColumnName) or the writer's own sequence.

7. Validate. Compare countDocuments() with the view's count. Update one document, delete another and $unset a field on a third, then query each _id once the next file lands. Stop the pipeline for a few minutes and confirm it resumes without a gap.

What happens after the initial load?

• Updates append a row with the whole document, or only the changed fields if the reader skips updateLookup. Older rows stay in older files.
• Deletes append a row holding the key and an operation marker. Time-window copies and batch jobs never write one.
• $unset is an update: the new row lacks the field.
• New fields become columns in files written afterwards only, so readers merge schemas by column name. Athena does that for Parquet by default.
• Type changes give newer files a different Parquet type for the field. AWS says to test such queries, and for Parquet they work only on partitioned tables.
• Re-loads write a second copy; point readers at one, or let the view absorb duplicates.
• Ordering holds across shards in a change stream, though a writer may collapse several changes to one _id into one row.

How MongoDB documents land in Parquet: structs, flattened columns or JSON text

Each method writes nested documents, arrays and other BSON types differently, and that decides how you query the files.

DMS table mode skips fields that appear after the target was defined, and a sample can miss rare fields. Structs also need matching table definitions: Athena fails with GroupColumnIO cannot be cast to PrimitiveColumnIO when Glue declares a struct or array column as string.

Decimal128 and other exact values

Decimal128 holds 34 significant digits and a DOUBLE about 15, so money written as DOUBLE comes back rounded. Write it as a DECIMAL sized for the field, or as a string the query casts. Digit-only strings such as postcodes and order numbers need care too: a writer that infers them as integers drops leading zeros, and anything above 9,223,372,036,854,775,807 doesn't fit INT64. Keep them as strings.

Performance, latency and S3 cost

Change streams deliver events as they happen, yet readers see them only when the writer closes its next file. Cost follows file counts, bytes scanned and copies kept.

• Fewer, larger files. Each upload writes at least one file per changed collection, so short intervals multiply small files. Compact older data with a scheduled CTAS or INSERT OVERWRITE into another table.
• Row groups. Each row group keeps per-column min/max statistics that let engines skip it. DMS defaults to 10,024 rows per group; Atlas Data Federation to 128 MiB or maxFileSize, whichever is smaller, Snappy-compressed unless you choose gzip or none.
• Lifecycle rules can move cold prefixes, and copies left by re-loads, to cheaper storage classes or delete them.

Production considerations

• Retries. A writer that fails mid-batch rewrites or re-appends it; the view absorbs repeats, plain counts don't.
• Source impact and re-loads. Initial scans read whole collections: run big ones off-peak, into a new folder, and switch readers when done.
• Monitoring. Alert when lag nears the oplog window, and on the age of the newest object in each prefix.
• Security. A read-only MongoDB user, a writer limited to its prefix, and private networking where available.

Common problems and troubleshooting

Oplog window exceeded during a pipeline outage

Resuming fails with ChangeStreamHistoryLost (error code 286): the oplog no longer holds the saved token's operation. Compare rs.printReplicationInfo() with the length of the outage.

1. Enlarge the window first, so the recovery itself doesn't age out.
2. Re-read the affected collections into a new folder and switch readers once it's loaded.
3. Find deletes from the gap by comparing _id sets between MongoDB and the view. A re-read only writes documents that still exist.

Deleted or updated documents still appear in Athena results

The raw files hold every version. Query the current-state view, and check that it partitions by _id, orders by a value that rises with every change, and counts DMS full-load rows as inserts. Counts that doubled after a re-load mean the table's LOCATION spans both copies.

Nested columns fail with GroupColumnIO cannot be cast to PrimitiveColumnIO

Athena raises this when a Glue table declares a struct or LIST column, as Atlas Data Federation or Spark write them, as a primitive such as string. Declare struct<…> or array<…> types, or rerun the crawler over files that hold every field. JSON text columns are primitives and don't raise it.

Using Estuary for MongoDB to S3

Estuary connects its MongoDB capture to the Amazon S3 Parquet materialization. The capture opens one change stream per database and keys each collection on _id; the materialization appends what changed to Parquet files, one path per collection.

• Timing:Upload Interval (1s, 1m, 5m default, 15m, 30m, 1h) applies once the pipeline is caught up; large backfills aren't held to it.
• Rows: append-only, with no hard-delete option. Changes to an _id within one transaction combine into one row. Update rows carry the full document; a delete row holds _id, _meta/op = 'd' and flow_published_at, both selected by default. The whole-document column isn't.
• Layout:<Prefix>/<Path>/v0000000000/00000000000000000000.parquet, numbered upward. Date patterns such as %Y/%m/%d use the write time in UTC, not a date in the document.
• Shape: at Depth 1, the default, top-level fields become columns, and objects, arrays and multi-type fields each become a Parquet JSON column. Depth 2 or Unlimited Depth adds nested fields as columns named by path, such as customer/city.
• Types:_id is a string (ObjectId as hex), and dates are UTC microsecond TIMESTAMPs. Decimal128 lands as DOUBLE (INT64 if every value seen is whole); digit-only strings can be inferred as integers and written as INT64, losing leading zeros, and a value beyond INT64 stops the task. castToString fixes all three (step Set each binding's path and fields).
• Files: Snappy, row groups of up to 1,000,000 rows or 512 MiB, and an approximate 10 GiB File Size Limit. New fields join later files and a widened type changes only new files, with no backfill.
• Limits: for MongoDB collections, inference follows at most 1,000 field locations, so very wide documents need extra fields added to the schema by hand before they can be selected. DocumentDB has its own batch-only capture.
• Assurance: the platform holds SOC 2 Type II certification and is HIPAA and GDPR compliant.

Implementing MongoDB to S3 with Estuary

These steps adapt Estuary's general data flow guide to this route.

1. Complete the prerequisites in MongoDB requirements and Amazon S3 requirements. Estuary also needs:
   • Estuary's IP addresses allowed in Atlas or your firewall, or an SSH tunnel.
   • s3:PutObject for an IAM user's access key, or a role set up per the AWS IAM guide.
   • The lifecycle rule, set to 1 day or more.
2. Create the MongoDB capture. On the Sources page, click New Capture, choose MongoDB, name the capture, and fill in the endpoint properties:
   • Address: host and port, or a mongodb+srv:// string, with ?authSource=admin if the user is defined in admin.
   • User and Password: the read user.
   • Database: optional list; empty discovers every database except admin, config and local.
   • Capture Batch Collections in Addition to Change Stream Collections: check it to include views and time series.
   • Through a bastion, SSH Endpoint and SSH Private Key under Network Tunnel.

3. Select collections. Click Next. Each collection becomes a binding under Output Collections.
   • Keep Capture Mode at Change Stream Incremental where possible: it's the only capture mode that records deletes.
   • Optional: for plain column names, use Rename on _meta/op in the collection's schema to add an alternate name such as _meta_op (a projection), which is materialized as its own column.

Click Save and publish. The backfill and change streams start together.

4. Create the S3 Parquet materialization. On the capture's details page, click Materialize, choose Amazon S3 Parquet, and fill in the endpoint properties:
   • Bucket and Region.
   • Authentication: Access Key (AWS Access Key ID, AWS Secret Access key) or AWS IAM (AWS Region, AWS Role ARN).
   • Upload Interval: keep 5m unless readers need fresher files and accept more of them.
   • Prefix: optional, such as mongodb/shop, without a leading / or s3://.
   • File Size Limit, and Row Group Row Limit and Row Group Byte Limit under Parquet Configuration: keep the defaults unless your engine prefers smaller files.

Testing doesn't contact S3: a wrong bucket, Region or key shows up as a task error after publishing.

5. Set each binding's path and fields. Click Next. Each binding's Path is its folder under the prefix and may contain date patterns. In its Field Selection table:
   • Set Field Depth: Depth 1 keeps nested objects as JSON columns; Depth 2 or Unlimited Depth promotes nested fields.
   • Keep Decimal128 amounts and digit-only strings exact with a custom column type: in the materialization's specification editor, add price: { castToString: true } under the binding's fields.include. The column becomes a STRING with the source's digits; cast it to DECIMAL when you query.
6. Publish. Click Save and publish. A large backfill writes its files without waiting for the upload interval.
7. Verify that files are landing.
   • On the capture details page, the Shard Information card reads PRIMARY; open a collection from the Collections list for its Data Preview.
   • The materialization details page charts Data Read.
   • Query s3://<bucket>/<prefix>/<path>/v0000000000/ with the view from Setting up MongoDB to S3 replication: partition by _id, order by flow_published_at, and drop rows where _meta/op (or your renamed column) is 'd'. Then run that section's validation checks.

How Estuary handles commits, re-reads and extra destinations on S3

Files appear only when a transaction commits

Each file streams to S3 as a multipart upload while a transaction is processed and is completed only when the transaction commits, so readers never list a half-written file. A retried transaction re-uploads its files, under the same names unless the path has date patterns, and the latest-row view absorbs any repeat. The lifecycle rule clears parts an interruption leaves behind.

Scanning and streaming in one capture

The capture alternates between catching the change streams up and scanning collections for up to five minutes at a time. Resume tokens keep advancing during the scan, so the oplog has to cover an outage, not the whole initial load. For any one collection, scanned documents and change events are never emitted at once, so an older copy can't land after a newer change. Scan progress is checkpointed every 50,000 documents or 4 MiB.

What a re-read writes to S3

Materializations read the collection, a durable log kept as JSON files in your cloud storage bucket, rather than MongoDB.

How far back a rebuild reaches depends on the collection's storage: trial buckets hold roughly 20 days, while a bucket you own keeps everything. That's also how one capture can keep every version in S3 and current-state tables in a warehouse: add a second materialization, and MongoDB still serves one set of change streams.

When another approach may make more sense

• Atlas only, structs wanted:$out writes groups and LIST columns with no extra service. A time-window copy misses deletes, and $out overwrites a file of the same name, so give each run its own filename (MongoDB's tutorial reuses one).
• AWS-only teams running DMS: one service for full load and CDC with an operation column. You size the replication instance, and table mode needs one type per field.
• Transforms before landing: Spark, Glue or dlt reshape documents in code; you own scheduling and schema drift, and batch reads miss deletes.
• A current-state table on S3 rather than a change log: Estuary's Apache Iceberg materialization merges changes into Iceberg tables, using Spark jobs on your EMR Serverless application and a REST catalog such as AWS Glue.
• No change streams: poll an indexed, always-increasing field and accept missed deletes. Estuary's Batch Incremental mode does this, every 24 hours unless you set a schedule.

Related resources

• Estuary: MongoDB capture · Amazon S3 Parquet · Apache Iceberg · Delta updates · Field selection · Backfilling data · MongoDB to S3 Parquet integration
• MongoDB: Change streams · Oplog · $out to S3 · Triggers tutorial
• AWS: DMS MongoDB source · DMS S3 target · Glue MongoDB connections · Athena JSON functions · Aborting incomplete uploads

To try the route, follow the MongoDB CDC tutorial or create a MongoDB capture in the Estuary dashboard. Pricing: estuary.dev/pricing.