Convert MongoDB Export to CSV

Convert mongoexport NDJSON to CSV — handles all Extended JSON types

import json
import csv
from pathlib import Path

def extract_extended_json(value):
    """Unwrap MongoDB Extended JSON types to plain Python values."""
    if not isinstance(value, dict):
        return value
    if "$oid" in value:
        return value["$oid"]
    if "$date" in value:
        # Returns ISO string; slice to date only if preferred
        return value["$date"]
    if "$numberLong" in value:
        return int(value["$numberLong"])
    if "$numberDecimal" in value:
        return float(value["$numberDecimal"])
    if "$numberInt" in value:
        return int(value["$numberInt"])
    if "$binary" in value:
        return f"<binary:{value['$binary'].get('subType', '')}>"
    # Unknown extended type — return as-is
    return value

def flatten_doc(doc, prefix="", sep="."):
    """Recursively flatten nested documents."""
    out = {}
    for key, value in doc.items():
        full_key = f"{prefix}{sep}{key}" if prefix else key
        value = extract_extended_json(value)
        if isinstance(value, dict):
            out.update(flatten_doc(value, full_key, sep))
        elif isinstance(value, list):
            # Join scalar arrays; skip nested object arrays (handle separately)
            if all(not isinstance(v, (dict, list)) for v in value):
                out[full_key] = "|".join(str(extract_extended_json(v)) for v in value)
            else:
                out[full_key] = json.dumps(value)  # keep complex arrays as JSON string
        else:
            out[full_key] = value
    return out

input_path = "export.json"   # mongoexport NDJSON output
output_path = "output.csv"

headers_seen = set()
rows = []

with open(input_path, encoding="utf-8") as f:
    for line in f:
        line = line.strip()
        if not line:
            continue
        doc = json.loads(line)
        flat = flatten_doc(doc)
        headers_seen.update(flat.keys())
        rows.append(flat)

headers = sorted(headers_seen)

with open(output_path, "w", newline="", encoding="utf-8-sig") as f:
    writer = csv.DictWriter(f, fieldnames=headers, extrasaction="ignore")
    writer.writeheader()
    for row in rows:
        writer.writerow({k: row.get(k, "") for k in headers})

print(f"Exported {len(rows)} documents to {output_path}")

Using pandas for simpler flat collections (no Extended JSON types)

import pandas as pd

# If your collection has no Extended JSON types (no ObjectIds, plain dates)
df = pd.read_json("export.json", lines=True)

# Rename _id to id if present
if "_id" in df.columns:
    df = df.rename(columns={"_id": "id"})

# Flatten nested address object
if "address" in df.columns:
    address_df = pd.json_normalize(df["address"].tolist())
    address_df.columns = [f"address_{c}" for c in address_df.columns]
    df = pd.concat([df.drop(columns=["address"]), address_df], axis=1)

# Join array columns
if "scores" in df.columns:
    df["scores"] = df["scores"].apply(
        lambda x: "|".join(str(v) for v in x) if isinstance(x, list) else x
    )

df.to_csv("output.csv", index=False, encoding="utf-8-sig")

Export directly from MongoDB using pymongo (skip mongoexport)

from pymongo import MongoClient
import pandas as pd

client = MongoClient("mongodb+srv://user:pass@cluster.mongodb.net/")
db = client["mydb"]
collection = db["users"]

# Fetch all documents, exclude internal _id if not needed
docs = list(collection.find({}, {"_id": 0}))

# Or include _id as string:
# docs = list(collection.find({}))
# for doc in docs:
#     doc["_id"] = str(doc["_id"])

df = pd.json_normalize(docs, sep=".")

# Handle list columns
for col in df.columns:
    if df[col].apply(lambda x: isinstance(x, list)).any():
        df[col] = df[col].apply(
            lambda x: "|".join(str(v) for v in x) if isinstance(x, list) else x
        )

df.to_csv("users.csv", index=False, encoding="utf-8-sig")
print(f"Exported {len(df)} documents")
client.close()