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.

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.