MongoDB 聚合管道

概述

MongoDB 聚合管道（Aggregation Pipeline）是一种强大的数据处理框架，可以将多个处理阶段串联起来，对文档进行转换、过滤、分组和计算。类似 SQL 的 GROUP BY + JOIN + HAVING。

管道核心

聚合管道的每个阶段接收上一阶段的输出作为输入，经过处理后输出到下一阶段。常见阶段: $match → $group → $sort → $limit。

基础概念

常用管道阶段

阶段	说明	SQL 类比
`$match`	过滤文档	WHERE
`$group`	分组聚合	GROUP BY + 聚合函数
`$project`	字段投影	SELECT 字段列表
`$sort`	排序	ORDER BY
`$limit`	限制数量	LIMIT
`$skip`	跳过文档	OFFSET
`$unwind`	展开数组	-
`$lookup`	左外连接	LEFT JOIN
`$count`	计数	COUNT(*)
`$addFields`	添加计算字段	-
`$facet`	多维聚合	-

语法与代码

基础聚合

php

<?php
declare(strict_types=1);

use MongoDB\Client;

$client = new Client('mongodb://localhost:27017');
$collection = $client->selectCollection('shop', 'orders');

// 示例1: 统计每个用户的订单数和总金额
$pipeline = [
    ['$match' => ['status' => 'completed']],
    ['$group' => [
        '_id'        => '$userId',
        'orderCount' => ['$sum' => 1],
        'totalAmount' => ['$sum' => '$amount'],
    ]],
    ['$sort' => ['totalAmount' => -1]],
    ['$limit' => 10],
];

$results = $collection->aggregate($pipeline)->toArray();

foreach ($results as $row) {
    printf(
        "用户: %s | 订单数: %d | 总金额: %.2f\n",
        $row['_id'],
        $row['orderCount'],
        $row['totalAmount']
    );
}

// 示例2: 按分类统计商品数和平均价格
$pipeline = [
    ['$group' => [
        '_id'          => '$category',
        'count'        => ['$sum' => 1],
        'avgPrice'     => ['$avg' => '$price'],
        'maxPrice'     => ['$max' => '$price'],
        'minPrice'     => ['$min' => '$price'],
        'totalStock'   => ['$sum' => '$stock'],
    ]],
    ['$match' => ['count' => ['$gte' => 5]]],
    ['$sort' => ['count' => -1]],
];

$categoryStats = $collection->aggregate($pipeline)->toArray();

$project — 字段投影与计算

php

<?php
declare(strict_types=1);

// 字段选择和重命名
$pipeline = [
    ['$project' => [
        'userName' => '$name',
        'userEmail' => '$email',
        'isAdult'  => ['$gte' => ['$age', 18]],
        'ageGroup' => [
            '$cond' => [
                'if'   => ['$gte' => ['$age', 60]],
                'then' => 'senior',
                'else' => [
                    '$cond' => [
                        'if'   => ['$gte' => ['$age', 18]],
                        'then' => 'adult',
                        'else' => 'minor',
                    ],
                ],
            ],
        ],
    ]],
];

// $addFields — 添加计算字段（保留所有原字段）
$pipeline = [
    ['$addFields' => [
        'fullName' => ['$concat' => ['$lastName', ' ', '$firstName']],
        'discountedPrice' => ['$multiply' => ['$price', 0.9]],
        'tagsCount' => ['$size' => ['$ifNull' => ['$tags', []]]],
    ]],
];

$lookup — 关联查询

php

<?php
declare(strict_types=1);

// 左外连接 — 关联用户信息到订单
$pipeline = [
    ['$match' => ['status' => 'completed']],
    ['$lookup' => [
        'from'         => 'users',
        'localField'   => 'userId',
        'foreignField' => '_id',
        'as'           => 'userInfo',
    ]],
    ['$unwind' => '$userInfo'], // 展开数组（因为 left join 最多一条）
    ['$project' => [
        'orderId'    => '$_id',
        'amount'     => 1,
        'userName'   => '$userInfo.name',
        'userEmail'  => '$userInfo.email',
    ]],
    ['$sort' => ['amount' => -1]],
];

$orderDetails = $collection->aggregate($pipeline)->toArray();

// 子查询 — 获取用户的最新3条订单
$pipeline = [
    ['$lookup' => [
        'from'         => 'orders',
        'let'          => ['userId' => '$_id'],
        'pipeline'     => [
            ['$match' => ['$expr' => ['$eq' => ['$userId', '$$userId']]]],
            ['$sort' => ['createdAt' => -1]],
            ['$limit' => 3],
        ],
        'as'           => 'recentOrders',
    ]],
];

$usersWithOrders = $client->selectCollection('shop', 'users')
    ->aggregate($pipeline)
    ->toArray();

$unwind — 数组展开

php

<?php
declare(strict_types=1);

// 展开标签数组，统计每个标签的文章数
$pipeline = [
    ['$unwind' => '$tags'],
    ['$group' => [
        '_id'    => '$tags',
        'count'  => ['$sum' => 1],
    ]],
    ['$sort' => ['count' => -1]],
];

// 保留索引信息
$pipeline = [
    ['$unwind' => [
        'path' => '$items',
        'includeArrayIndex' => 'itemIndex',
        'preserveNullAndEmptyArrays' => true, // 保留空数组
    ]],
    ['$project' => [
        'itemName' => '$items.name',
        'itemIndex' => 1,
    ]],
];

实战示例

销售报表聚合

php

<?php
declare(strict_types=1);

use MongoDB\Client;

class SalesReportService
{
    private MongoDB\Driver\Collection $orders;
    private MongoDB\Driver\Collection $products;

    public function __construct(Client $client)
    {
        $this->orders = $client->selectCollection('shop', 'orders');
        $this->products = $client->selectCollection('shop', 'products');
    }

    /**
     * 每日销售统计
     */
    public function dailySales(string $startDate, string $endDate): array
    {
        $start = new MongoDB\BSON\UTCDateTime(strtotime($startDate) * 1000);
        $end = new MongoDB\BSON\UTCDateTime(strtotime($endDate) * 1000);

        $pipeline = [
            ['$match' => [
                'status'    => 'completed',
                'createdAt' => ['$gte' => $start, '$lt' => $end],
            ]],
            ['$group' => [
                '_id' => [
                    '$dateToString' => [
                        'format' => '%Y-%m-%d',
                        'date'   => '$createdAt',
                    ],
                ],
                'orderCount'  => ['$sum' => 1],
                'totalAmount' => ['$sum' => '$amount'],
                'avgAmount'   => ['$avg' => '$amount'],
            ]],
            ['$sort' => ['_id' => 1]],
        ];

        return $this->orders->aggregate($pipeline)->toArray();
    }

    /**
     * 商品销售排行榜
     */
    public function productSalesRanking(int $topN = 10): array
    {
        $pipeline = [
            ['$unwind' => '$items'],
            ['$group' => [
                '_id'       => '$items.productId',
                'name'      => ['$first' => '$items.name'],
                'totalQty'  => ['$sum' => '$items.quantity'],
                'totalSales' => ['$sum' => ['$multiply' => ['$items.quantity', '$items.price']]],
            ]],
            ['$sort' => ['totalSales' => -1]],
            ['$limit' => $topN],
        ];

        return $this->orders->aggregate($pipeline)->toArray();
    }

    /**
     * 用户消费分级
     */
    public function customerSegments(): array
    {
        $pipeline = [
            ['$group' => [
                '_id'        => '$userId',
                'totalSpent' => ['$sum' => '$amount'],
                'orderCount' => ['$sum' => 1],
                'avgOrder'   => ['$avg' => '$amount'],
            ]],
            ['$addFields' => [
                'segment' => [
                    '$switch' => [
                        'branches' => [
                            ['case' => ['$gte' => ['$totalSpent', 10000]], 'then' => 'VIP'],
                            ['case' => ['$gte' => ['$totalSpent', 5000]], 'then' => 'Gold'],
                            ['case' => ['$gte' => ['$totalSpent', 1000]], 'then' => 'Silver'],
                        ],
                        'default' => 'Bronze',
                    ],
                ],
            ]],
            ['$group' => [
                '_id'    => '$segment',
                'count'  => ['$sum' => 1],
                'avgSpent' => ['$avg' => '$totalSpent'],
            ]],
            ['$sort' => ['count' => -1]],
        ];

        return $this->orders->aggregate($pipeline)->toArray();
    }
}

php

<?php
declare(strict_types=1);

// 一个管道中同时执行多个聚合
$pipeline = [
    ['$match' => ['status' => 'completed']],
    ['$facet' => [
        'dailyStats' => [
            ['$group' => [
                '_id' => ['$dateToString' => ['format' => '%Y-%m-%d', 'date' => '$createdAt']],
                'total' => ['$sum' => '$amount'],
            ]],
            ['$sort' => ['_id' => 1]],
        ],
        'topProducts' => [
            ['$unwind' => '$items'],
            ['$group' => [
                '_id' => '$items.name',
                'count' => ['$sum' => '$items.quantity'],
            ]],
            ['$sort' => ['count' => -1]],
            ['$limit' => 5],
        ],
        'totalSummary' => [
            ['$group' => [
                '_id'     => null,
                'total'   => ['$sum' => '$amount'],
                'avg'     => ['$avg' => '$amount'],
                'count'   => ['$sum' => 1],
            ]],
        ],
    ]],
];

$result = $collection->aggregate($pipeline)->toArray();
$report = $result[0];

echo "每日统计: " . count($report['dailyStats']) . " 天\n";
echo "Top产品: " . count($report['topProducts']) . " 个\n";
echo "总金额: {$report['totalSummary'][0]['total']}\n";

注意事项

性能优化

php

<?php
// 1. $match 尽早放在管道开头 — 减少后续阶段处理的数据量
// 好
$pipeline = [
    ['$match' => ['status' => 'active']],
    ['$group' => ...],
    ['$sort' => ...],
];

// 差
$pipeline = [
    ['$group' => ...],
    ['$match' => ['status' => 'active']],
];

// 2. $project 尽早使用 — 减少字段传输
$pipeline = [
    ['$match' => ['age' => ['$gte' => 18]]],
    ['$project' => ['name' => 1, 'age' => 1]], // 尽早投影
    ['$group' => ...],
];

// 3. 利用索引 — 确保 $match 和 $sort 使用索引字段
// 允许内存使用
$pipeline = [...];
$options = [
    'allowDiskUse' => true, // 允许写入临时文件（大数据集必需）
];
$results = $collection->aggregate($pipeline, $options);

内存限制

聚合操作默认限制 100MB 内存。如果超过限制，必须设置 allowDiskUse: true，否则会报错。

最佳实践

1. 聚合结果缓存

php

<?php
// 对于复杂聚合，缓存结果
class CachedAggregation
{
    private MongoDB\Driver\Collection $collection;
    private Redis $redis;

    public function getReport(string $cacheKey, array $pipeline, int $ttl = 300): array
    {
        $cached = $this->redis->get($cacheKey);
        if ($cached !== false) {
            return json_decode($cached, true);
        }

        $results = $this->collection->aggregate($pipeline, ['allowDiskUse' => true])->toArray();
        $this->redis->setex($cacheKey, $ttl, json_encode($results));

        return $results;
    }
}

2. 分阶段调试

php

<?php
// 逐阶段检查聚合结果
$pipeline = [
    ['$match' => ['status' => 'completed']],
    ['$group' => ['_id' => '$userId', 'total' => ['$sum' => '$amount']]],
    ['$sort' => ['total' => -1]],
];

// 逐阶段执行
$stage1 = $collection->aggregate(array_slice($pipeline, 0, 1))->toArray();
echo "阶段1 (match): " . count($stage1) . " 条\n";

$stage2 = $collection->aggregate(array_slice($pipeline, 0, 2))->toArray();
echo "阶段2 (group): " . count($stage2) . " 条\n";

MongoDB 聚合管道 ​

概述 ​

基础概念 ​

常用管道阶段 ​

语法与代码 ​

基础聚合 ​

$project — 字段投影与计算 ​

$lookup — 关联查询 ​

$unwind — 数组展开 ​

实战示例 ​

销售报表聚合 ​

多维聚合 ($facet) ​

注意事项 ​

性能优化 ​

最佳实践 ​

1. 聚合结果缓存 ​

2. 分阶段调试 ​

参考链接 ​