Data Engineering

MongoDB Reference

The document model, everyday mongosh commands, the aggregation pipeline, indexes, and writing to MongoDB from Spark.

Contents

1. Documents & Collections 2. Everyday Commands 3. Aggregation Pipeline 4. Indexes & Explain Plans 5. Writing from Spark

1. Documents & Collections

MongoDB is a document database. Records are JSON-like documents (stored as BSON) grouped into collections; documents in one collection can have different fields. In data platforms it typically serves low-latency lookups, such as the latest alert or profile for an account, while history and analytics live in the lake or warehouse.

Relational	MongoDB
Database	Database
Table	Collection
Row	Document
Column	Field (can be nested or an array)
Primary key	`_id` (unique, auto-generated if omitted)
JOIN	Embedding related data, or `$lookup`

Embed data that is read together and owned by one parent (an alert's triggered rules). Reference data that is shared or grows without bound (transactions per account).
Design documents around query patterns: one read should ideally fetch one document.
Documents are limited to 16 MB.

2. Everyday Commands

mongosh -u admin -p admin             # or: docker exec -it mongodb mongosh -u admin -p admin
show dbs
use fraud
show collections
db.transactions.countDocuments()
db.transactions.findOne()

Find

db.alerts.find({ risk_level: "HIGH" }, { _id: 0, account_id: 1, amount: 1, rules: 1 })
db.transactions.find({ amount: { $gt: 50000 } })
db.transactions.find({ channel: { $in: ["ATM", "CARD_POS"] } })
db.alerts.find({ rules: "IMPOSSIBLE_TRAVEL" })        // matches an element of an array
db.transactions.find({ account_id: "ACC000001" }).sort({ event_time: -1 }).limit(5)
db.transactions.distinct("channel")

Update

db.alerts.updateOne({ txn_id: "X" },
                    { $set: { case_status: "CLOSED", reviewed_at: new Date() } })
db.alerts.updateMany({ risk_level: "LOW" }, { $set: { case_status: "AUTO_CLOSED" } })

Practise inserts and deletes in a scratch database (use practice); deletes cannot be undone. If a streaming job upserts these documents, a replay will overwrite manual edits.

3. Aggregation Pipeline

The aggregation pipeline passes documents through a list of stages, each transforming the stream:

// transactions and total amount per channel
db.transactions.aggregate([
  { $group: { _id: "$channel", txns: { $sum: 1 }, total: { $sum: "$amount" } } },
  { $sort: { txns: -1 } }
])

// how often each fraud rule fires (rules is an array)
db.alerts.aggregate([
  { $unwind: "$rules" },
  { $group: { _id: "$rules", count: { $sum: 1 } } },
  { $sort: { count: -1 } }
])

Stage	SQL equivalent
`$match`	WHERE
`$group`	GROUP BY
`$sort` / `$limit`	ORDER BY / LIMIT
`$project`	SELECT columns
`$lookup`	JOIN
`$unwind`	One row per array element

Put $match first so the pipeline can use indexes and process fewer documents.

4. Indexes & Explain Plans

db.alerts.getIndexes()
db.transactions.createIndex({ channel: 1, amount: -1 })      // compound index
db.transactions.find({ account_id: "ACC000001" }).explain("executionStats")

In explain output, COLLSCAN means a full collection scan; IXSCAN means an index was used. Compare totalDocsExamined with nReturned.
Compound indexes follow the ESR rule: Equality fields first, then Sort fields, then Range fields.
Every index speeds reads but slows writes and uses memory, so index for real query patterns only.

5. Writing from Spark

The MongoDB Spark Connector (v10+) reads and writes collections as DataFrames, including from Structured Streaming:

(alerts_df.write.format("mongodb")
    .option("connection.uri", "mongodb://admin:admin@mongodb:27017")
    .option("database", "fraud")
    .option("collection", "alerts")
    .option("operationType", "replace")     # upsert by _id: replays don't duplicate
    .mode("append")
    .save())

Set a deterministic _id (such as the transaction ID) so that reprocessing replaces documents instead of inserting duplicates. This is what makes a streaming sink idempotent.