MongoDB to Databricks: Change Streams, CDC & Delta Table Modelling Guide

MongoDB data reaches Databricks three ways: a batch export loaded into Delta tables, the MongoDB Spark Connector running on your Databricks compute, or change data capture (CDC) from change streams through a pipeline that merges every insert, update and delete into Delta tables keyed on _id. Batch fits migrations and daily reporting; streaming CDC fits collections that change all day and must reflect deletes.

The constraints sit at both ends: the oplog must outlast your longest pipeline outage, or the stream can't resume, and every Delta load runs on a SQL warehouse or cluster that bills while awake, with nested fields landing as JSON strings, VARIANT or columns. One managed option is Estuary, which captures MongoDB change streams into a durable collection and has your Databricks SQL warehouse merge them into Unity Catalog tables on a configurable schedule.

Choosing a MongoDB to Databricks method

Databricks has no native MongoDB ingestion: neither Lakeflow Connect's managed connectors nor its community connectors include MongoDB. Batch reads rescan collections and miss deletes; streaming reads catch every change but need the oplog window and a reader that keeps running. The sections below follow change streams.

How MongoDB to Databricks CDC works

Debezium and the MongoDB Spark Connector both read change streams, not local.oplog.rs. A change stream provides:

• Resume tokens, valid while the oplog still holds their point.
• One ordered stream across shards, through mongos.
• Whole documents on updates only on request (updateLookup or stored post-images); otherwise just the changed fields.
• Delete events carrying only the document key (_id, plus the shard key on sharded collections).

A new pipeline notes a stream position, copies the collections, then applies events from there so writes during the copy aren't lost. Databricks applies events in batches: MERGE on _id into a current-state table, or appends to a change table that a view or AUTO CDC flow reduces.

MongoDB requirements

• Change streams need a replica set or sharded cluster (WiredTiger, protocol version 1), never a standalone mongod.
• A user with find and changeStream on each replicated database; the built-in read role has both:

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

Network access to MongoDB Atlas

Atlas accepts connections only from its IP access list or a private endpoint. Allowlist whatever connects: Databricks serverless compute's published outbound CIDR ranges (Public Preview; the legacy stable IPs were deprecated in May 2026), your classic compute network's egress address, or a pipeline service's IPs. Private endpoints (AWS PrivateLink, Azure Private Link, GCP Private Service Connect) need an M10+ dedicated cluster. Connect with the cluster's mongodb+srv:// string.

Sizing the oplog window

Size the window in hours: longer than any reader outage, plus the initial copy if your tool opens the stream afterwards. rs.printReplicationInfo() shows it. The default WiredTiger oplog is 5% of free disk (990 MB to 50 GB); multi-document updates, deletes and in-place updates fill it fastest.

Databricks requirements

Databricks needs a Unity Catalog catalog and schema, compute for the loads (usually a SQL warehouse) and a service principal.

• Grants:USE CATALOG; USE SCHEMA, CREATE TABLE and, for staged loads, CREATE VOLUME on target schemas; SELECT and MODIFY on existing tables; Can use on the warehouse; CREATE SCHEMA only if the loader creates schemas.
• Auth: OAuth machine-to-machine. Databricks labels personal access tokens legacy and revokes unused ones after 90 days.
• Warehouse: serverless starts in about 2–6 seconds, pro and classic in about 4 minutes.
• Workspace IP access lists, if enabled, must admit the loader.

Setting up MongoDB to Databricks replication

1. Confirm change streams and size the oplog window.
2. Create the read user and open network access (allowlist, private endpoint or SSH bastion).
3. Prepare Databricks: catalog, schemas, compute, service principal and grants.
4. Choose the landing shape per collection: a current-state table merged on _id or an append-only change table, and which fields become columns.
5. Copy, then stream. Note a start time, run the initial read, then stream from that time so writes during the copy replay. With the Spark Connector: a batch read, then a streaming read with change.stream.startup.mode set to timestamp (the default is the latest event) and change.stream.lookup.full.document set to updateLookup. Leave change.stream.publish.full.document.only off; it drops deletes, which have no fullDocument.
6. Apply changes in order. Append events to a bronze table and let AUTO CDC reduce them; SEQUENCE BY handles late arrivals:

sql
CREATE OR REFRESH STREAMING TABLE shop.orders;

CREATE FLOW orders_cdc AS AUTO CDC INTO shop.orders
FROM stream(shop.orders_changes)
KEYS (_id)
APPLY AS DELETE WHEN op = 'delete'
SEQUENCE BY STRUCT(cluster_time, event_seq)
COLUMNS * EXCEPT (op, cluster_time, event_seq)
STORED AS SCD TYPE 1;

Events from one transaction can share a clusterTime, so event_seq is a tie-breaker your reader assigns in stream order.

7. Validate. Compare countDocuments() per collection; update, delete and $unset a document and check each after one load; restart the reader and confirm it resumes without a gap.

What happens after the initial load?

• Updates carry the whole document only if requested; otherwise you patch stored JSON yourself.
• Deletes carry only the document key; a merge removes or flags the row. Deleted contents need pre-images (changeStreamPreAndPostImages, MongoDB 6.0+).
• $unset is an update: the field is missing from the next document.
• New fields need no DDL in JSON or VARIANT columns. Typed columns need ALTER TABLE … ADD COLUMN or MERGE WITH SCHEMA EVOLUTION, and older rows stay NULL until reloaded.
• Type drift breaks typed columns. Delta type widening (Databricks Runtime 15.4 LTS+) covers INT to BIGINT or DOUBLE; other changes need a new column or a rebuild.

Modelling MongoDB documents in Databricks: JSON strings, VARIANT or columns

Nested data has three homes in Databricks: STRING columns of JSON text, VARIANT (Databricks SQL, or Databricks Runtime 15.4 LTS and above for Delta tables), or typed columns including STRUCT and ARRAY. Databricks recommends VARIANT over JSON strings; both use the same : path syntax. Promote the fields you filter, join and aggregate on to typed columns; leave the rest as JSON.

Querying JSON stored as strings

For an order whose customer and items land as JSON strings:

json
{ "_id": "65f1c0a2e4b0a1b2c3d4e5f6", "status": "shipped", "placedAt": "2026-03-02T14:05:11.000Z",
  "customer": { "id": 4182, "address": { "city": "Leeds", "zip code": "LS1" } },
  "items": [ { "sku": "A-1", "qty": 2, "price": 9.5 }, { "sku": "B-7", "qty": 1, "price": 30 } ] }

sql
SELECT _id,
       customer:id::BIGINT          AS customer_id,
       customer:address.city        AS ship_city,
       customer:address.`zip code`  AS zip,
       items:[*].sku                AS skus      -- JSON array string
FROM shop.orders;

• Paths into a STRING match names case-insensitively, so a document holding both userId and userid makes the path ambiguous. Brackets (customer:['Id']) match case exactly.
• Missing fields and JSON null both return SQL NULL.

Arrays: one row per element

sql
CREATE OR REPLACE VIEW shop.order_items AS
SELECT o._id AS order_id, li.pos AS line_no,
       li.item.sku, li.item.qty,
       CAST(li.item.price AS DECIMAL(12,2)) AS unit_price
FROM shop.orders AS o,
     LATERAL posexplode_outer(
       from_json(o.items, 'array<struct<sku:string,qty:int,price:double>>')
     ) AS li(pos, item);

• from_json parses the text into ARRAY<STRUCT<…>> with only the fields you name; posexplode_outer numbers the elements and keeps an order with no array as one row of NULLs.
• Aggregate each grain separately: summing an order total over order_items counts it once per line, and exploding two sibling arrays multiplies rows.

Converting to VARIANT

parse_json (or try_parse_json, which returns NULL on malformed input) converts JSON strings to VARIANT. Three differences bite: paths become case-sensitive, [*] isn't supported, and a JSON null becomes a VARIANT null that IS NULL misses. For mixed-type fields, schema_of_variant reports each value's type; cast with try_cast and count the NULLs.

BSON to Databricks type mapping

Performance, latency and Databricks cost

Capture can run continuously, but a Delta table is only as fresh as its last commit, and every commit runs on paid compute.

Warehouse uptime drives the bill

• Each load wakes the warehouse, which bills until Auto Stop: 45 idle minutes by default on pro and classic (minimum 10), 10 on serverless (minimum 5 in the UI, 1 through the API).
• Auto Stop counts idle minutes, so loads that come more often than its setting keep the warehouse running. Serverless, starting in seconds, makes short Auto Stop practical.
• A Structured Streaming job holds its cluster for as long as it runs.

Making MERGE cheaper on Delta

• Merge less often. Repeated updates to one _id between loads can collapse into one change.
• Bound each MERGE by the batch's key range, or cluster on the key, so Delta skips files.
• Keep deletion vectors on so merges mark rows instead of rewriting files; predictive optimization runs OPTIMIZE and VACUUM on managed tables.
• Append never-updated collections; note that updateLookup costs MongoDB one read per update.

Production considerations

• Recovery. Resume tokens and Spark checkpoints point into the oplog; a lost checkpoint or an outage past the window means a new initial copy.
• Monitoring. Alert on reader lag against the oplog window, and on time since the last Delta commit (DESCRIBE HISTORY lists commits).
• Security. A least-privilege MongoDB role, an OAuth service principal, and allowlists or private endpoints on both sides.

Common problems and troubleshooting

ChangeStreamHistoryLost after an outage

A reader stopped longer than the oplog window fails with ChangeStreamHistoryLost (error code 286): its position points at discarded history.

1. Lengthen the window (Atlas Minimum Oplog Window, or minRetentionHours) so the retry holds.
2. Re-copy the affected collections and open a new stream. Spark Structured Streaming needs a fresh checkpoint location; Debezium re-snapshots by itself only with snapshot.mode: when_needed.
3. Reconcile deletes. A re-copy finds only surviving documents, so rows deleted during the gap stay until you compare _id sets.

SQL warehouse that never stops

When loads come more often than Auto Stop allows, the warehouse never idles long enough to stop: a pro or classic warehouse at the default 45 minutes, loaded every 30, runs almost continuously. Confirm it on the warehouse's Monitoring tab. Fixes, cheapest first: lower Auto Stop, switch to serverless, lengthen the interval or confine short ones to business hours, and share one warehouse across pipelines.

Column-name collisions in Unity Catalog

MongoDB field names can hold any character except null, and userId and userid are different fields. Unity Catalog matches column names case-insensitively, and Delta tables without column mapping reject spaces, ,;{}(), newlines, tabs and =. So those two, or order total (once its space becomes _) and order_total, compete for one column, and the loader must fail, rename or drop one. Find top-level case collisions:

js
db.users.aggregate([
  { $project: { f: { $objectToArray: "$$ROOT" } } },
  { $unwind: "$f" },
  { $group: { _id: { $toLower: "$f.k" }, names: { $addToSet: "$f.k" } } },
  { $match: { "names.1": { $exists: true } } }
])

Rename at the source or in the pipeline, or keep the parent as JSON so both survive.

Using Estuary for MongoDB to Databricks

The MongoDB capture feeds the Databricks materialization: the capture opens one change stream per database and writes each MongoDB collection to a collection keyed on _id, and the materialization stages each transaction as compressed JSON in a flow_staging volume per target schema, then applies it through your SQL warehouse: MERGE when keys exist, COPY INTO when all rows are new.

• Load interval: 30 minutes by default once caught up, with optional fast-sync windows and 4-hour syncs outside them; backfills ignore it. Timing is keyed to the warehouse, so materializations sharing one warehouse and schedule wake it together.
• Change events: updates always carry the full document (updateLookup, or stored post-images when every captured collection in a database has them). Only Change Stream Incremental mode captures deletes.
• Document shape: each collection becomes one Delta table keyed on a STRING_id. At the default Depth 1, top-level fields become columns; nested objects, arrays and multi-type fields become STRING columns of JSON text, not VARIANT, so the : queries above work as written. Deeper selections add path-named columns such as `customer/address/city`.
• Types: dates become TIMESTAMP. Decimal128 values arrive as exact strings, but schema inference marks numeric strings, so they land as DOUBLE (NUMERIC(38,0) while every value is whole), and digit-only text such as US ZIP codes can lose leading zeros. castToString keeps the exact text.
• Deletes: with Hard Delete off (the default), deleted documents keep their rows, marked _meta/op = 'd'. Delta-update bindings append with COPY INTO and never delete.
• Limits: spaces and ,;{}() in field names become _, and case-only differences are flagged as conflicts. Inference tracks up to 1,000 fields per collection. There's no SSH tunnel to Databricks, and the DocumentDB variant is batch-only.

Implementing MongoDB to Databricks with Estuary

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

1. Complete the prerequisites in MongoDB requirements, with the oplog window sized, and Databricks requirements. Add Estuary's IP addresses to the Atlas IP access list (and the Databricks one, if enabled), or use an SSH tunnel to a self-managed MongoDB. Private and BYOC deployments can reach Atlas over PrivateLink.
2. Create the MongoDB capture. On the Sources page, click New Capture, choose MongoDB, name the capture, and fill in the endpoint properties:
   • Address: the mongodb+srv:// string or mongodb://host:port, without credentials; add ?authSource=admin if the user authenticates against 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 for views and time series.

3. Select collections. Click Next. Each collection becomes a binding under Output Collections. Keep Capture Mode at Change Stream Incremental wherever deletes matter (capture modes) and Automatically keep schemas up to date on. Click Save and publish: the backfill and change streams start together.

4. Create the Databricks materialization. On the capture's details page, click Materialize, choose Databricks, and fill in the endpoint properties from the warehouse's Connection Details tab:
   • Address (port 443 by default) and HTTP path of the SQL warehouse.
   • Catalog Name and Schema Name (default if blank).
   • Hard Delete: on deletes rows; off keeps soft deletes.
   • Authentication: Personal Access Token (the default) or OAuth2 M2M, with Client ID and Client Secret.

5. Choose the sync schedule. Under Sync Schedule, Sync Frequency defaults to 30 minutes; weigh it against the warehouse's Auto Stop (cost). For business-hours freshness, set Timezone, Fast Sync Start Time, Fast Sync Stop Time and optionally Fast Sync Enabled Days; outside that window, syncs run every 4 hours (sync schedules).

6. Choose how documents land. Click Next and pick a naming convention; Match source structure gives each MongoDB database its own schema (pre-create them or grant CREATE SCHEMA). Then, per binding, on the Config tab:
   • Table and Schema override names. Enable delta updates only for append-only collections.
   • In Field Selection, set Field Depth to Depth 1 (the default), Depth 2 or Unlimited Depth, then Require or Exclude single fields (field selection).
   • Add castToString: true to Decimal128 and digit-only text fields in the binding's fields specification.
7. Publish. Click Save and publish. Estuary creates the tables and staging volumes and loads the backfill without waiting for the schedule. Schema not yet available for collection means no inferred schema exists yet: wait 5–10 minutes and publish again (details).
8. Verify. The capture details page reads PRIMARY in its Shard Information card; open a collection from the Collections list to see its Data Preview. The materialization details page charts Data Read. In Databricks, compare counts (WHERE `_meta/op` <> 'd' under soft deletes) and run the setup step 7 checks.

How Estuary handles backfills, oplog loss and Delta table rebuilds

Captured changes land first in a collection, JSON files in your own cloud storage bucket, and the materialization reads from there, not from MongoDB.

Backfill and change streams in one pass

Large collections don't stall change capture: the connector scans in slices of up to five minutes, catching every change stream up to the cluster's latest operation time between slices. Scanned documents and change events for one collection are never emitted together, so a scan can't overwrite a newer change in Delta, and resume tokens keep advancing.

When the oplog window is exceeded

After an outage longer than the oplog window, the capture errors on restart; it never skips changes it can't read. An incremental backfill re-reads the affected collections while Databricks tables stay in place; merge bindings absorb the re-read documents, delta-update tables duplicate them. Backfills never emit deletes, so rows deleted during the gap stay until you compare _id sets.

Rebuilding Delta tables without touching MongoDB

A materialization backfill rebuilds a Delta table: it truncates the table, keeping clustering and other added DDL, and refills it from the collection without querying MongoDB. Adding a destination replays that stored history, then new changes; MongoDB keeps serving one change stream per database. Replay covers what the collection retains: about 20 days on trial storage, full history in your own bucket.

Schema inference and Databricks columns

A placeholder schema blocks documents until a real inferred one is published. New fields arrive as nullable columns through ALTER TABLE … ADD COLUMN, without a backfill. Some type changes migrate in place via the column mapping set on every table (whole numbers to DOUBLE, scalars to STRING); other incompatible changes drop and recreate the table by default, losing DDL you added. Inference never narrows a type, so a column that has turned STRING keeps it until you correct the source documents and reset the dataflow.

When another approach may make more sense

• You run Spark and want to own the code: the Spark Connector streams change events, but you own the initial copy (it starts at the latest event or a timestamp), the checkpoint, deletes and the MERGE. Batch reads infer the schema from 1,000 sampled documents unless you pass schemaHints, and the current 11.x line needs Spark 4.0 or later.
• A Debezium-to-Kafka pipeline already exists: apply the topics with AUTO CDC. Debezium emits documents as Extended JSON strings, sends full documents on updates only with capture.mode: change_streams_update_full, and re-snapshots after oplog loss only with snapshot.mode: when_needed. Rebuilding a Delta table from Kafka needs topics that still hold every event.
• One-time migration or daily batch, deletes irrelevant:mongoexport writes Extended JSON, so ObjectIds and dates arrive wrapped ($oid, $date). Each refresh rescans the collection, and deleted documents linger until you replace the table. Estuary's Batch Incremental mode polls an indexed cursor field without a scheduler.
• No change streams (views, time series, or DocumentDB with streams off): poll a strictly increasing indexed field and accept missed deletes, as Estuary's batch-only DocumentDB connector does.

Related resources

• Estuary docs: MongoDB capture · Amazon DocumentDB · Databricks materialization · Sync schedules · Schema evolution · Backfilling data
• MongoDB: Change streams · Atlas oplog settings · Atlas IP access list · Spark Connector streaming options
• Databricks: JSON path expressions · VARIANT vs JSON strings · AUTO CDC · SQL warehouse settings · Serverless limitations

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