Pipeline Architecture and Configuraton

The aggregation framework in MongoDB functions as a robust analytics engine, processing documents from one or multiple collections. It operates on the concept of a pipeline, where documents flow through a sequence of stages (see Figure 7-1). Each stage transforms the documents passing through it, taking the output of the previous stage as its input. This creates a continuous stream of document transformations.

Every stage comes with specific configurable parameters (tunables) that dictate its behavior. These parameters often take the form of operators used to reshape documents, perform arithmetic, manipulate fields, or execute accumulation tasks.

Fundamental Stages: Commmon Operations

Developing an aggregation pipeline involves chaining stages. The most frequently used stages include $match, $project, $sort, $skip, and $limit.

Consider a dataset containing corporate information, such as Facebook records. To find all companies established in 2004, a simple filter is applied:

// Filtering documents based on a specific condition
db.companies.aggregate([
    { $match: { founded_year: 2004 } }
])

To refine the output, a projection stage can be added to return only specific fields, excluding the default _id field:

// Filtering and then projecting specific fields
db.companies.aggregate([
    { $match: { founded_year: 2004 } },
    { $project: { _id: 0, name: 1, founded_year: 1 } }
])

The aggregate method accepts an array of documents, where each document defines a stage operator. In the example above, the pipeline consists of a filtering stage followed by a projection stage.

You can further constrain the results using the $limit stage. It is best practice to place $limit early in the pipeline to reduce the number of documents processed by subsequent stages, improving efficiency:

// Applying limit before projection for efficiency
db.companies.aggregate([
    { $match: { founded_year: 2004 } },
    { $limit: 5 },
    { $project: { _id: 0, name: 1 } }
])

If the order of results matters, use the $sort stage before $skip and $limit:

// Sorting, skipping, and limiting results
db.companies.aggregate([
    { $match: { founded_year: 2004 } },
    { $sort: { name: 1 } },
    { $skip: 10 },
    { $limit: 5 }
])

Expression Types

The framework supports a wide variety of expressions:

• Boolean: AND, OR, NOT.
• Set: Operations like intersection, union, and difference on arrays.
• Comparision: Range filters and value comparisons.
• Arithmetic: ceil, floor, log, sqrt, addition, subtraction, etc.
• String: Concatenation, substring search, and case conversion.
• Array: Filtering elements, slicing, and range extraction.
• Variable: Handling literals, date parsing, and conditionals.
• Accumulators: Calculating sums, averages, and statistics.

Projection and Unwinding Arrays

While basic projection is straightforward, complex operations often require extracting nested fields. When dealing with array fields, the $unwind stage is essential. It deconstructs an array field from the input documents to output a document for each element (see Figure 7-4).

If an input document contains an array with 3 elements in key3, $unwind will produce 3 separate documents, each containing one element from the array in the key3 field.

For a companies collection with a funding_rounds array:

// Unwinding a nested array to process individual elements
db.companies.aggregate([
    { $match: { 'funding_rounds.investor': 'Greylock' } },
    { $unwind: '$funding_rounds' },
    { $project: {
        _id: 0,
        company: '$name',
        amount: '$funding_rounds.raised_amount',
        year: '$funding_rounds.funded_year'
    }}
])

This pipeline matches companies with a specific investor, unwinds the funding_rounds array so each round becomes its own document, and then projects the relevant financial details.

Advanced Array Expressions

Filtering

The $filter operator selects a subset of an array based on a condition.

// Using $filter to get large funding rounds
db.companies.aggregate([
    { $match: { 'funding_rounds.investor': 'Greylock' } },
    { $project: {
        _id: 0,
        name: 1,
        founded_year: 1,
        rounds: {
            $filter: {
                input: '$funding_rounds',
                as: 'round',
                cond: { $gte: ['$$round.raised_amount', 100000000] }
            }
        }
    }}
])

Here, $$round refers to the variable defined in the as clause, iterating over the funding_rounds array.

Element Access and Slicing

• $arrayElemAt: Retrieves an element at a specific index.
• $slice: Returns a subset of an array.
• $size: Returns the number of elements in an array.

// Accessing specific elements and checking array size
db.companies.aggregate([
    { $match: { name: 'Facebook' } },
    { $project: {
        _id: 0,
        name: 1,
        first_round: { $arrayElemAt: ['$funding_rounds', 0] },
        early_rounds: { $slice: ['$funding_rounds', 1, 3] },
        total_rounds: { $size: '$funding_rounds' }
    }}
])

Accumulators

Accumulators (like $sum, $avg, $max, $min, $first, $last) compute values based on document streams. In MongoDB 3.2+, some accumulators can be used in the $project stage, but they must operate on arrays within a single document.

// Calculating total funding inside a projection
db.companies.aggregate([
    { $match: { funding_rounds: { $exists: true, $ne: [] } } },
    { $project: {
        _id: 0,
        name: 1,
        max_raised: { $max: '$funding_rounds.raised_amount' },
        total_raised: { $sum: '$funding_rounds.raised_amount' }
    }}
])

Grouping Data

The $group stage functions similarly to SQL's GROUP BY. It aggregates documents by a specified expression and outputs a document for each distinct group.

// Grouping companies by founding year to calculate average staff size
db.companies.aggregate([
    { $group: {
        _id: { year_founded: '$founded_year' },
        avg_staff: { $avg: '$number_of_employees' },
        companies: { $push: '$name' }
    }},
    { $sort: { '_id.year_founded': 1 } }
])

Best practice for the _id field in $group is to use a document that labels the grouping key (e.g., { year_founded: '$founded_year' }) rather than just the field path, ensuring clarity in the output.

You can also group by nested fields, such as the IPO year:

// Grouping by a nested field (IPO year)
db.companies.aggregate([
    { $group: {
        _id: { ipo_year: '$ipo.pub_year' },
        count: { $sum: 1 }
    }}
])

Joining Collections with $lookup

The $lookup stage performs a left outer join with another collection to filter data for the joined documents.

Syntax:

{
    $lookup: {
        from: <foreign_collection>,
        localField: <local_field>,
        foreignField: <foreign_field>,
        as: <output_array_field>
    }
}

Example: Joining orders with inventory.

// Basic lookup joining two collections
db.orders.aggregate([
    {
        $lookup: {
            from: 'inventory',
            localField: 'item',
            foreignField: 'sku',
            as: 'inventory_docs'
        }
    }
])

Example with Unwind: Joining based on an array of specifications.

// Lookup after unwinding an array field
db.orders.aggregate([
    { $unwind: '$specs' },
    {
        $lookup: {
            from: 'inventory',
            localField: 'specs',
            foreignField: 'size',
            as: 'inventory_docs'
        }
    },
    { $match: { 'inventory_docs': { $ne: [] } } }
])

Grouping and Projection Combined

Accumulators like $first and $last are typically used in the $group stage to capture boundary values.

// Using $first and $last in a group stage
db.companies.aggregate([
    { $unwind: '$funding_rounds' },
    { $sort: { 'funding_rounds.funded_year': 1 } },
    {
        $group: {
            _id: '$name',
            first_round: { $first: '$funding_rounds' },
            last_round: { $last: '$funding_rounds' },
            num_rounds: { $sum: 1 },
            total_raised: { $sum: '$funding_rounds.raised_amount' }
        }
    },
    {
        $project: {
            _id: 0,
            company: '$_id',
            first_raised: '$first_round.raised_amount',
            last_raised: '$last_round.raised_amount',
            num_rounds: 1,
            total_raised: 1
        }
    }
])

Writing Results to Collections

Two stages, $out and $merge, write pipeline results to a collection. These must be the final stage in the pipeline.

• $out: Overwrites the target collection entirely. It is limited to the same database and cannot target sharded collections.
• $merge (MongoDB 4.2+): The preferred method. It can write to any database/collection (sharded or unsharded), allows for incremental updates (inserting new docs, merging with existing ones, or custom updates), and is ideal for creating materialized views.

// Merging aggregation results into a target collection
db.companies.aggregate([
    { $match: { founded_year: { $gt: 2000 } } },
    { $group: { _id: '$category_code', count: { $sum: 1 } } },
    {
        $merge: {
            into: 'company_stats',
            on: '_id',
            whenMatched: 'replace',
            whenNotMatched: 'insert'
        }
    }
])