MongoDB to Silver: Flatten Nested Arrays Before They Double Revenue

Monday’s revenue dashboard said $4.7M. The ledger said $2.5M. I assumed a timezone. It was an array.

Orders in Mongo lived as one document with items: [{ sku, qty, cents }, ...]. The first warehouse model exploded items into lines. A later “enrichment” joined that table back to a still-nested copy of the same array and grouped by order_id without re-declaring grain. Line revenue was summed at order grain after the join had already multiplied rows. Finance was not wrong. Silver was.

I have done this more than once. Document databases are honest about nesting. Warehouses are honest about grain. The pipeline is where we pretend those are the same thing.

Land raw, flatten on purpose

Bronze is BSON or JSON as the source emitted it, plus ingest metadata. No “helpful” explode in the loader. The moment you flatten in bronze, you cannot rebuild silver when the path changes.

Silver is where the document becomes tables. One order table at order grain. One items table at line grain. Keys you can join. Types you can test.

import json
from typing import Any

def iter_order_lines(doc: dict[str, Any]) -> list[dict[str, Any]]:
    """One row per line. Order totals stay off this iterator."""
    order_id = doc["_id"]
    currency = doc.get("currency", "USD")
    lines: list[dict[str, Any]] = []
    for i, item in enumerate(doc.get("items") or []):
        lines.append({
            "order_id": str(order_id),
            "line_no": i,
            "sku": item.get("sku"),
            "qty": int(item.get("qty") or 0),
            "line_cents": int(item.get("cents") or 0),
            "currency": currency,
        })
    return lines


def order_header(doc: dict[str, Any]) -> dict[str, Any]:
    items = doc.get("items") or []
    return {
        "order_id": str(doc["_id"]),
        "status": doc.get("status"),
        "header_cents": int(doc.get("total_cents") or 0),
        "line_cents_sum": sum(int(i.get("cents") or 0) for i in items),
        "n_lines": len(items),
    }

header_cents and line_cents_sum must match within a documented rule (tax, rounding). When they do not, you have a source bug or a second explode. I fail the silver run on that check.

flowchart TD
  mongo["Mongo orders"] --> bronze["Bronze: raw document"]
  bronze --> hdr["Silver: order grain"]
  bronze --> line["Silver: line grain"]
  line --> gold["Gold facts"]
  hdr --> gold
  bronze -->|"explode twice"| bad["Join nested copy"]
  bad --> dbl["Revenue x2"]

One explode, two grains. A second explode is how $2.5M becomes $4.7M.

Change streams vs a dump

Change streams (or a CDC connector) give you updates and deletes with a resume token. That is the right path for a hot collection when you already operate Kafka, as in Debezium on Postgres — different engine, same rule: the bookmark must advance or you pay for it.

A nightly mongodump or a filtered export is fine for a small, slowly changing catalog. It is a full or incremental batch. Do not mix “we stream some fields and dump the rest” without a written grain and a watermark, or you will upsert yesterday’s nest over today’s stream.

Mongo aggregations in Compass are useful for debugging. They are not a warehouse metric layer. You will lose slowly changing history, conformed keys, and anyone who is not allowed on the cluster.

Mongo is not the warehouse

I have been asked to “just BI off Mongo” to skip the pipeline. That works until someone needs last Tuesday’s document, a join to billing in another system, or a definition of revenue that is not whatever the app wrote this sprint.

The warehouse gets flattened silver and a star or a narrow fact, same as any other source. The star vs one-big-table tradeoffs do not change because the bytes started as BSON.

Tests that catch the second explode

I do not wait for finance. Silver has two assertions on every run:

-- Order grain: one row per order_id
select order_id, count(*) as n
from silver.orders
group by 1
having count(*) > 1;

-- Line vs header: same currency, documented slop for tax
select
    h.order_id,
    h.header_cents,
    h.line_cents_sum
from silver.orders h
where abs(h.header_cents - h.line_cents_sum) > 2;

The first is a unique test. The second is the $4.7M detector. Put both in dbt (testing in production) so a second explode fails CI, not Monday.

_id is cast to text once in silver. Downstream never sees ObjectId. A join of string to variant is how line items vanish and someone “fixes” it with another flatten.

CDC apply is an upsert on _id

def apply_change(op: str, doc: dict) -> None:
    oid = str(doc["_id"])
    if op in ("insert", "update", "replace"):
        upsert_header(order_header(doc))
        replace_lines(oid, iter_order_lines(doc))
    elif op == "delete":
        tombstone_order(oid)

replace_lines deletes existing lines for that order, then inserts the new set. Merging line arrays blindly is how you keep SKUs the app already removed.

Resume tokens belong in the same ops dashboard as a Debezium slot. A stream that “works” but does not advance is a silent full replay on the next restart.

Hub: MongoDB.

FAQ

Should bronze store the whole document? Yes, plus ingest time and source token. Silver can throw fields away. Bronze cannot grow them back.

What about deeply nested configs, not orders? Same rule: one table per grain you will query. A 12-level nest that only the app needs can stay a JSON column in silver with a hash for change detection. Do not explode “just in case.”

Can Atlas SQL be the warehouse? It is a query engine on the operational cluster. You still lack conformed dims, SCD2, and a place finance will sign. Land, flatten, warehouse.

How do I handle schema drift in documents? New fields land in bronze automatically. Silver adds columns in a reviewed model. A silent items[].tax that only exists on 8% of docs is a typed column plus a coverage test, not SELECT *.

Pitfalls

Exploding in bronze “to make it easier.” You lose the document you would have used to rebuild.

Summing line amounts at order grain after a join that still has arrays. That is the double count.

Treating _id as a string in one job and ObjectId in another. Joins go empty. Cast once in silver.

Change stream without a sink watermark. Restarts replay. Apply must be idempotent on _id plus a token or cluster time.

Running heavy $lookup on the primary for analytics. You just made Mongo the warehouse and the OLTP box at once.

What this means for your pipelines

Documents are a good application shape and a bad fact table. I land them raw, flatten to declared grains, and test that header totals and line totals tell the same story.

Change streams when the collection is hot and deletes matter. Dumps when it is small and quiet. Never a dashboard on a nested array that has been joined to itself. The $4.7M number was not a finance dispute. It was silver without a grain.