使用 Python 与 PyMongo 的 MongoDB 聚合管道教程

处理大型数据集时，常常需要在保证性能的前提下提取有意义的模式和洞见。当您的应用将数据存储在 MongoDB 中时，直接在数据库中运行复杂查询和转换，往往比将数据迁移到外部分析工具更快。MongoDB 的聚合管道让您能够在数据所在之处进行处理、转换与分析，从而提供了解决方案。

您可以通过顺序连接简单操作来创建自定义数据处理流程。管道中的每个阶段都会转换文档，并将结果传递给下一个阶段。例如，您可能需要按日期范围筛选、按类别分组、计算统计指标并格式化输出——所有这些都可通过单个数据库操作完成，就近处理数据。

在本文中，您将学习如何构建聚合管道来解决常见的数据难题，所有示例均使用 PyMongo（MongoDB 官方的 Python 客户端）演示。

尽管本文聚焦于聚合管道，如果您是首次在 Python 中使用 MongoDB 入门课程将是一个全面的起点。通过本文，您也能获得足够的基础，将这些聚合概念自行迁移到 MongoDB 的查询语言中，或借助语言模型完成。

什么是 MongoDB 中的聚合管道？

想象一下，您需要分析多个产品的客户评价，以了解满意度趋势。传统查询可以检索数据，但并不擅长将信息进行组合、分析与转换为有用的摘要。

MongoDB 的聚合管道通过提供一系列相互构建的操作，为按步骤处理数据提供了结构化方式。

管道架构与文档流

将聚合管道想象成数据的装配线。集合中的每个文档从管道一端进入，经过多个“工位”（阶段），在其中被筛选、转换、分组或丰富。

一个阶段的输出成为下一阶段的输入，使您可以将复杂的数据转换拆解为更小、可管理的步骤。

这些管道采用声明式方法；您在每个阶段指定“要什么”，而不是“如何计算”。这种方式既能清晰表达您的数据处理意图，又让 MongoDB 负责执行细节。数据库还能基于管道结构应用多种优化。

阶段顺序在管道设计中至关重要。尽早筛选文档（在分组或复杂计算之前）能减少流经管道的数据量。

在处理大型集合时，这种方式可以显著提升性能。结构良好的管道仅处理您最终结果所需的数据。

文档流转：文档进入 → 经阶段转换 → 产出结果
声明式设计：指定“要什么”，而非“如何算”
有序处理：每个阶段的输出直接馈入下一阶段
性能收益：提前筛选能降低后续阶段的数据量

阶段类型与功能分类

MongoDB 聚合阶段按用途大致分为四类。筛选阶段如 $match 类似于查询，只选择符合特定条件的文档。这有助于在执行更复杂操作之前缩小数据集范围。

重塑阶段会改变文档结构。使用 $project 或 $addFields，您可以包含、排除或重命名字段，或基于现有值创建计算字段。这些阶段通过仅保留相关信息并添加分析所需的计算值，帮助简化文档。

当您需要基于共享特征合并多份文档时，就会用到分组阶段。$group 是其中的主力，允许您按文档组计算计数、求和、平均值以及其他聚合值。这能将成千上万条记录转化为回答分析问题的有意义摘要。

为补全数据画面，连接阶段如 $lookup 允许您合并多个集合中的信息。这使您可以用相关数据丰富文档，类似于 SQL 的连接，但适配 MongoDB 的文档模型。跨集合引用数据的能力有助于保持恰当的规范化，同时在一次操作中返回完整结果。

筛选：使用 $match 基于条件选择特定文档
重塑：使用 $project 与 $addFields 转换文档结构
分组：用 $group 进行合并与汇总以获得聚合值
连接：使用 $lookup 跨集合关联相关数据

有关完整的能力与操作符参考，请查阅 MongoDB 聚合管道手册。

为本教程搭建环境

在深入了解 MongoDB 聚合管道之前，您需要一个安装了 PyMongo 并能访问示例数据集的工作环境。本节将带您基于 sample_analytics 数据集完成设置。该数据集包含用于演示聚合概念的理想金融数据。

安装 PyMongo

PyMongo 是 MongoDB 的官方 Python 驱动。您可以在 macOS 和 Windows 上通过 pip 安装：

# Install PyMongo using pip
pip install pymongo

# Or if you're using conda
conda install -c conda-forge pymongo技术支持

版本兼容性：PyMongo 4.x 适配 Python 3.7+
依赖处理：安装程序会自动处理所需依赖
平台支持：在 Windows、macOS 和 Linux 上表现一致

获取示例数据集

MongoDB 提供了无需自建数据即可使用的示例数据集。最简单的方法是使用 MongoDB Atlas（云端版本）：

在 mongodb.com/cloud/atlas 创建免费的 MongoDB Atlas 账户
设置免费层集群

2.1. 选择永久免费版本

2.2. 为集群命名

2.3. 点击 “Deploy”

2.4. 将您的用户名与密码复制到安全位置

2.5. 点击 “Create database user”

2.6. 将连接方法选择为 “Drivers”

2.7. 复制连接 URI，供下一步使用

3.1. 在集群视图中点击三点菜单

3.2. 选择 “Load sample dataset”

3.3. 在列表中选择 sample_analytics 并等待加载完成。

对于本地 MongoDB 安装，您可以使用以下方式加载示例数据集：

# Download and restore the sample dataset
python -m pip install pymongo[srv]
python -c "from pymongo import MongoClient; MongoClient().admin.command('getParameter', '*')"技术支持

数据规模：sample_analytics 数据集体量较小（约 10MB），可在免费层顺畅使用
集合结构：包含 customers 与 accounts 集合及其关联数据
数据真实度：基于真实的金融模式，便于进行有意义的聚合示例

连接到 MongoDB

现在您已准备好 MongoDB 与示例数据集，接下来编写代码进行连接并验证设置：

from pymongo.mongo_client import MongoClient
from pymongo.server_api import ServerApi

uri = "mongodb+srv://bex:[email protected]/?retryWrites=true&w=majority&appName=Cluster0"

# Create a new client and connect to the server
client = MongoClient(uri, server_api=ServerApi('1'))

# Send a ping to confirm a successful connection
try:
   client.admin.command('ping')
   print("Pinged your deployment. You successfully connected to MongoDB!")
except Exception as e:
   print(e)
Pinged your deployment. You successfully connected to MongoDB!
# Access the sample_analytics database
db = client.sample_analytics

# Verify connection by counting documents in collections
customer_count = db.customers.count_documents({})
account_count = db.accounts.count_documents({})

print(f"Found {customer_count} customers and {account_count} accounts in sample_analytics")

# Preview one document from each collection
print("\nSample customer document:")
print(db.customers.find_one())

print("\nSample account document:")
print(db.accounts.find_one())技术支持

输出：

Found 500 customers and 1746 accounts in sample_analytics

Sample customer document:
{'_id': ObjectId('5ca4bbcea2dd94ee58162a68'), 'username': 'fmiller', 'name': 'Elizabeth Ray', 'address': '9286 Bethany Glens\nVasqueztown, CO 22939', 'birthdate': datetime.datetime(1977, 3, 2, 2, 20, 31), 'email': '[email protected]', 'active': True, 'accounts': [371138, 324287, 276528, 332179, 422649, 387979], 'tier_and_details': {'0df078f33aa74a2e9696e0520c1a828a': {'tier': 'Bronze', 'id': '0df078f33aa74a2e9696e0520c1a828a', 'active': True, 'benefits': ['sports tickets']}, '699456451cc24f028d2aa99d7534c219': {'tier': 'Bronze', 'benefits': ['24 hour dedicated line', 'concierge services'], 'active': True, 'id': '699456451cc24f028d2aa99d7534c219'}}}

Sample account document:
{'_id': ObjectId('5ca4bbc7a2dd94ee5816238c'), 'account_id': 371138, 'limit': 9000, 'products': ['Derivatives', 'InvestmentStock']}技术支持

验证：代码确认了您的连接与数据集访问
文档预览：展示了您将要处理的文档结构

运行此代码后，您应能看到文档计数以及客户与账户文档的预览。这表明您的环境已准备好进行接下来要探索的聚合管道示例。

如需了解 PyMongo 与本地 MongoDB 的入门，请参阅我们的 PyMongo 新手教程。

深入理解 MongoDB 聚合管道

既然我们已确认连接并浏览了文档结构，接下来深入 MongoDB 的聚合框架。这一强大特性能让我们在数据库内直接处理并转换数据。本节将结合示例数据集，探索最常见的管道阶段与实用示例。

使用 $match 筛选数据

$match 阶段根据指定条件筛选文档。您可以把它看作在进行更复杂操作前聚焦所关心数据的一种方式。

让我们查找额度高于 $9,000 的高级账户：

pipeline = [
   {"$match": {"limit": {"$gt": 9000}}}
]

premium_accounts = list(db.accounts.aggregate(pipeline))

print(f"Found {len(premium_accounts)} premium accounts")
print(premium_accounts[0])技术支持

该管道在 $match 阶段中使用比较操作符 $gt（大于）来筛选账户。这与 find() 方法类似，但在管道上下文中执行。查询会检查 accounts 集合中的每个文档，仅保留 limit 字段大于 9000 的文档。

输出：

Found 1701 premium accounts
{'_id': ObjectId('5ca4bbc7a2dd94ee5816238d'), 'account_id': 557378, 'limit': 10000, 'products': ['InvestmentStock', 'Commodity', 'Brokerage', 'CurrencyService']}技术支持

从结果可以看到，管道在数据库中共 1746 个账户中识别出 1701 个高级账户。此筛选步骤缩小了我们的关注范围，使后续分析更高效、更有针对性。在您的项目中，您可以用同样的方法，将焦点放在活跃用户、超过特定金额的交易或特定类别的产品上，然后对这些文档进行更深入的分析。

使用 $project、$sort、$limit 与 $skip 重塑文档

这些阶段帮助您控制要包含哪些字段以及如何组织结果。下面的示例将获取额度最高的前 5 个账户，并仅展示关键信息：

pipeline = [
   {"$project": {
       "_id": 0,
       "account_id": 1,
       "limit": 1,
       "product_count": {"$size": "$products"}
   }},
   {"$sort": {"limit": -1}},
   {"$limit": 5}
]

top_accounts = list(db.accounts.aggregate(pipeline))
for account in top_accounts:
   print(account)技术支持

该管道包含三个协同工作的阶段：

1. $project 阶段通过以下方式重塑每个文档：

排除 _id 字段（设为 0）
包含 account_id 与 limit 字段（设为 1）
创建新字段 product_count，使用 $size 操作符（众多聚合管道操作符之一）来统计 products 数组中的项数

2. $sort 阶段按 limit 字段降序（-1）排序

3. $limit 阶段在排序后仅保留前 5 个文档

输出：

{'account_id': 674364, 'limit': 10000, 'product_count': 1}
{'account_id': 278603, 'limit': 10000, 'product_count': 2}
{'account_id': 383777, 'limit': 10000, 'product_count': 5}
{'account_id': 557378, 'limit': 10000, 'product_count': 4}
{'account_id': 198100, 'limit': 10000, 'product_count': 3}技术支持

输出显示额度最高的五个账户，额度均为 $10,000。我们还能通过计算字段 product_count 查看每个账户持有的产品数量。

这种方式可实现更简洁、即需即取的数据呈现，而不是返回全部字段。在为您的应用构建仪表盘或报告时，您可以采用类似技术，仅向用户展示最相关的信息，减少数据传输并简化界面。

使用 $group 分组与聚合

$group 阶段是聚合发力之处。您可以为文档分类并按每个组计算指标。让我们按产品类型计算平均账户额度：

pipeline = [
   {"$unwind": "$products"},  # First unwind the products array
   {"$group": {
       "_id": "$products",
       "avg_limit": {"$avg": "$limit"},
       "count": {"$sum": 1}
   }},
   {"$sort": {"avg_limit": -1}}
]

product_analysis = list(db.accounts.aggregate(pipeline))
for product in product_analysis:
   print(f"Product: {product['_id']}")
   print(f"  Average limit: ${product['avg_limit']:.2f}")
   print(f"  Number of accounts: {product['count']}")技术支持

此管道使用多个操作符来分析产品数据：

$unwind 操作符将具有多个产品的文档拆分为多条。例如，一个包含 ["Derivatives", "InvestmentStock"] 的账户会变成两条文档，分别对应每个产品。
$group 阶段随后：

按产品名分组（$group 中的 _id 字段决定分组键）
使用 $avg 累加器计算每种产品的平均额度
使用 $sum: 1（对每条文档加 1）统计每种产品的账户数量

3. $sort 阶段按平均额度降序排列结果

输出：

Product: Commodity
 Average limit: $9963.89
 Number of accounts: 720
Product: Brokerage
 Average limit: $9960.86
 Number of accounts: 741
Product: InvestmentStock
 Average limit: $9955.90
 Number of accounts: 1746
Product: InvestmentFund
 Average limit: $9951.92
 Number of accounts: 728
Product: Derivatives
 Average limit: $9951.84
 Number of accounts: 706
Product: CurrencyService
 Average limit: $9946.09
 Number of accounts: 742技术支持

结果显示，Commodity 产品关联的平均账户额度最高，而 CurrencyService 最低。

此类分析有助于识别产品供给与客户消费能力之间的相关性。

在您的应用中，您可以用类似技术分析按类别的销售额、按功能的用户参与度，或按模块的错误情况——凡是需要跨组汇总数据而非逐条查看记录的场景都适用。

使用 $lookup 与 $unwind 连接集合

当您的数据分散在多个集合中时，$lookup 能将其整合。让我们找出客户及其关联账户：

pipeline = [
   {"$match": {"username": "fmiller"}},  # Find a specific customer
   {"$lookup": {
       "from": "accounts",  # Collection to join with
       "localField": "accounts",  # Field from customers collection
       "foreignField": "account_id",  # Field from accounts collection
       "as": "account_details"  # Name for the new array field
   }},
   {"$project": {
       "name": 1,
       "accounts": 1,
       "account_details.account_id": 1,
       "account_details.limit": 1,
       "account_details.products": 1
   }}
]

customer_accounts = list(db.customers.aggregate(pipeline))

print(f"Customer: {customer_accounts[0]['name']}")
print(f"Has {len(customer_accounts[0]['account_details'])} accounts:")

for account in customer_accounts[0]['account_details']:
   print(f"  Account {account['account_id']}: ${account['limit']} limit with products: {', '.join(account['products'])}")技术支持

该管道演示了如何跨集合连接相关数据：

$match 阶段通过用户名查找特定客户
$lookup 阶段对 accounts 集合执行左外连接：

from：指定要连接的集合
localField：当前集合（customers）中用于匹配的字段
foreignField：目标集合（accounts）中用于匹配的字段
as：包含匹配文档的新数组字段名称

3. $project 阶段塑形，只展示相关字段

输出：

Customer: Elizabeth Ray
Has 6 accounts:
 Account 371138: $9000 limit with products: Derivatives, InvestmentStock
 Account 324287: $10000 limit with products: Commodity, CurrencyService, Derivatives, InvestmentStock
 Account 276528: $10000 limit with products: InvestmentFund, InvestmentStock
 Account 332179: $10000 limit with products: Commodity, CurrencyService, InvestmentFund, Brokerage, InvestmentStock
 Account 422649: $10000 limit with products: CurrencyService, InvestmentStock
 Account 387979: $10000 limit with products: Brokerage, Derivatives, InvestmentFund, Commodity, InvestmentStock技术支持

输出为 Elizabeth Ray 的金融资产组合提供了全面视图，通过一次查询结果展示了她名下的六个账户及其关联产品。

这使您无需多次查询或在客户端进行连接，就能获得数据关系的 360 度全景。

在您的应用中，考虑哪些因规范化而拆分到不同集合中的数据需要为分析或展示重新组合。常见示例包括带有活动历史的用户档案、带有库存状态的产品，或带有关联评论的内容。

组合多个阶段以进行复杂分析

现在，让我们处理一个更复杂的业务问题：“按服务等级分组，客户的平均账户额度是多少？”这需要我们处理客户文档中的嵌套对象（tier_and_details），然后将该信息与 accounts 集合关联。

首先，为理解 customers 集合中的结构，我们来检查一个示例客户的 tier_and_details 字段与 accounts 字段。tier_and_details 字段是一个对象，其中每个键是某一等级订阅的标识符，值包含等级名称等详情。accounts 字段是与该客户关联的账户 ID 数组。

# First, let's look at the structure of 'tier_and_details' again
sample_customer = db.customers.find_one({"username": "fmiller"}) # Using a known customer for consistency
print("Tier structure example for customer 'fmiller':")
print(sample_customer['tier_and_details'])
print(f"Customer 'fmiller' has account IDs: {sample_customer['accounts']}")技术支持

输出：

Tier structure example for customer 'fmiller':
{'0df078f33aa74a2e9696e0520c1a828a': {'tier': 'Bronze', 'id': '0df078f33aa74a2e9696e0520c1a828a', 'active': True, 'benefits': ['sports tickets']}, '699456451cc24f028d2aa99d7534c219': {'tier': 'Bronze', 'benefits': ['24 hour dedicated line', 'concierge services'], 'active': True, 'id': '699456451cc24f028d2aa99d7534c219'}}
Customer 'fmiller' has account IDs: [371138, 324287, 276528, 332179, 422649, 387979]技术支持

pipeline = [
   # Step 1: Project necessary fields, including 'accounts' and convert 'tier_and_details'
   {"$project": {
       "tiers_array": {"$objectToArray": "$tier_and_details"}, # Convert object to array
       "customer_account_ids": "$accounts",  # Explicitly carry over the customer's account IDs
       "_id": 1  # Keep customer _id for later
   }},
   # Step 2: Unwind the new 'tiers_array' to process each tier object separately
   {"$unwind": "$tiers_array"},
   # Step 3: Reshape to clearly define the tier and keep customer account IDs
   {"$project": {
       "tier": "$tiers_array.v.tier",  # Extract the tier name
       "customer_account_ids": 1,  # Ensure account IDs are still present
       "customer_id": "$_id"  # Rename _id to customer_id for clarity
   }},
   # Step 4: Look up account details using the customer_account_ids
   {"$lookup": {
       "from": "accounts",  # Target collection
       "localField": "customer_account_ids",  # Array of account IDs from the customer
       "foreignField": "account_id",  # Field in the 'accounts' collection
       "as": "matched_account_details"  # New array with joined account documents
   }},
   # Step 5: Unwind the 'matched_account_details' array.
   {"$unwind": "$matched_account_details"},
   # Step 6: Group by tier to calculate statistics
   {"$group": {
       "_id": "$tier",  # Group by the tier name
       "avg_limit": {"$avg": "$matched_account_details.limit"}, # Calculate average limit
       "total_accounts_in_tier": {"$sum": 1},  # Count how many accounts fall into this tier
       "unique_customers_in_tier": {"$addToSet": "$customer_id"}  # Count unique customers in this tier
   }},
   # Step 7: Format the final output
   {"$project": {
       "tier_name": "$_id",  # Rename _id to tier_name
       "average_account_limit": "$avg_limit",
       "number_of_accounts": "$total_accounts_in_tier",
       "number_of_customers": {"$size": "$unique_customers_in_tier"},  # Get the count of unique customers
       "_id": 0  # Exclude the default _id
   }},
   # Step 8: Sort by average limit
   {"$sort": {"average_account_limit": -1}}
]

tier_analysis = list(db.customers.aggregate(pipeline))

print("\nTier Analysis Results:")

for tier_data in tier_analysis:
   print(f"Tier: {tier_data['tier_name']}")
   print(f"  Average Account Limit: ${tier_data['average_account_limit']:.2f}")
   print(f"  Number of Accounts in this Tier: {tier_data['number_of_accounts']}")
   print(f"  Number of Unique Customers in this Tier: {tier_data['number_of_customers']}")技术支持

该管道将复杂分析拆解为八个可管理的步骤：

$project：我们首先转换 tier_and_details 对象。 $objectToArray 操作符将该对象转换为键值对数组（tiers_array）。这很关键，因为 $unwind 等阶段作用于数组。我们还显式传递客户文档中的 accounts 数组为 customer_account_ids，以及客户的 _id。
$unwind：该阶段将 tiers_array 拆解，为客户可能拥有的每个等级条目创建独立文档。每个新文档仍包含 customer_account_ids 和原始客户 _id。
$project：我们重塑文档，从嵌套结构（$tiers_array.v.tier）中清晰提取等级名称（如 "Bronze"），并为清晰起见将客户 _id 重命名为 customer_id。customer_account_ids 会继续传递。
$lookup：在此与 accounts 集合进行连接。我们将客户文档中的账户号数组 customer_account_ids 作为 localField，foreignField 为 accounts 集合中的 account_id。MongoDB 会查找所有其 account_id 存在于 customer_account_ids 数组中的账户，并将其作为数组添加到 matched_account_details 字段。
$unwind：对 matched_account_details 进行拆解。这样，如果某一客户-等级组合关联多个账户，我们会为每个具体账户生成一条文档，并与对应的客户和等级关联。
$group：按等级对文档进行分组。对于每个等级，我们基于连接的账户详情中的 limit 使用 $avg 计算 avg_limit。使用 $sum: 1 统计 total_accounts_in_tier。还会用 $addToSet 收集每个等级内的唯一 customer_id。
$project：对最终输出进行塑形。将分组阶段中的 _id（即等级名）重命名为 tier_name。使用 $size 获取 unique_customers_in_tier 的数量。
$sort：按 average_account_limit` 降序排序，让平均额度最高的等级排在前面。

Tier Analysis Results:
Tier: Silver
 Average Account Limit: $9974.55
 Number of Accounts in this Tier: 393
 Number of Unique Customers in this Tier: 95
Tier: Bronze
 Average Account Limit: $9964.11
 Number of Accounts in this Tier: 418
 Number of Unique Customers in this Tier: 93
Tier: Platinum
 Average Account Limit: $9962.53
 Number of Accounts in this Tier: 427
 Number of Unique Customers in this Tier: 101
Tier: Gold
 Average Account Limit: $9962.44
 Number of Accounts in this Tier: 426
 Number of Unique Customers in this Tier: 99技术支持

现在的结果展示了略有不同的层级：Silver 等级客户的账户平均额度最高，其次是 Bronze、Platinum，最后是 Gold。

这类洞见有助于理解客户细分。例如，金融机构可能会据此研究为何 Silver 与 Bronze 等级的平均额度如此之高，或据此定制营销活动与增值服务。

使用 $objectToArray 是解锁嵌套等级数据的关键，同时谨慎传递 customer_account_ids 确保 $lookup 能将客户与其特定账户关联起来。

当您在自己的数据集中遇到需要用于聚合的嵌套对象（如 用户偏好、产品属性 或 配置设置）时，请记住 $objectToArray 技巧。

务必确保后续阶段（尤其是 $lookup 操作）所需的字段在 $project 阶段显式包含。用这种结构化方式拆解复杂数据，能帮助您直接在 MongoDB 中提炼出有意义的洞见。

高级管道模式

除顺序转换外，MongoDB 的聚合框架还提供了可应对复杂分析查询的高级模式。当您需要同时从多个维度分析数据，或基于滚动窗口执行计算时，这些高级模式能在数据库内直接提供强大解决方案。

$facet 阶段允许您在单个阶段内，基于同一输入文档集并行执行多个聚合子管道。想象一下，您需要同时按价格区间对产品分组，并列出同一产品集合中最受欢迎的前 5 个品牌。$facet 能并行处理这些不同的分析视角。

$facet 中的每个子管道都会独立处理输入文档，并产生自己的输出文档数组。这意味着您可以在一次数据库查询中收集多种指标或摘要——例如构建综合仪表盘所需的内容。

例如，您可以同时从同一用户数据集中获取活跃用户总数、按订阅等级的用户分布，以及最近加入的用户列表。这让您能够在不增加多次数据库调用开销的情况下，构建丰富、多面向的报告，简化复杂视图的数据获取。

使用窗口函数进行时序分析

自 MongoDB 5.0 起引入的窗口函数，可在与当前文档相关的一组文档（“窗口”）上执行计算。

这对时序数据或任何需要考虑邻近文档上下文的有序数据非常有用。例如，您可能希望计算 7 天移动平均销售额，或按时间为每位客户计算交易累计和。

窗口函数通常在 $setWindowFields 阶段中使用。该阶段允许您定义分区（如每个产品的销售）以及分区内的排序（如按日期）。

然后，您可以在定义的窗口（例如，前 3 条文档加当前文档）内应用 $avg、$sum、$min、$max 等函数，或 $derivative、$integral 等专用函数。

考虑计算产品销售的累计总额。一个使用 $setWindowFields 的管道可能如下：

pipeline = [
   {"$match": {"category": "Electronics"}}, # Filter for electronics
   {"$sort": {"sale_date": 1}}, # Sort by sale date
   {"$setWindowFields": {
       "partitionBy": "$product_id", # Calculate running total per product
       "sortBy": {"sale_date": 1},
       "output": {
           "running_total_sales": {
               "$sum": "$sale_amount",
               "window": {
                   "documents": ["unbounded", "current"] # Sum from start to current document
               }
           }
       }
   }}
]
# electronics_sales_with_running_total = list(db.sales.aggregate(pipeline))技术支持

在这个概念性管道中，对于每笔电子产品销售，都会新增一个 running_total_sales 字段。该字段表示从最早的销售到当前销售，针对该产品的 sale_amount 之和。

这些以往需要复杂客户端逻辑或多次查询的计算，现在可在数据库内直接完成，从而简化应用代码并提升趋势分析或环比分析的性能。

结论

MongoDB 聚合管道为您提供了一种在数据库中直接处理数据的结构化方法。通过串联不同的操作阶段，您可以完成复杂的数据处理，如筛选记录、重塑文档、分组汇总以及跨集合数据合并。

这种方法有助于优化您的数据工作流，正如 PyMongo 示例所示，使您更快地从大型数据集中提取意义。就近处理数据，往往能为分析型查询带来更好的性能。

熟练掌握聚合管道有助于应对多种数据分析挑战。要将所学付诸实践，您可以尝试一个动手项目，例如搭建零售数据管道。

若想从更广的视角了解数据系统以及 MongoDB 技能在更大图景中的位置， Data Engineer with Python 学习路径提供深入学习。如果您正寻求职业发展，也可查阅常见的 MongoDB 面试题，以了解典型的解题期望。

MongoDB 聚合管道常见问答

什么是 MongoDB 聚合管道？

MongoDB 聚合管道是一套用于数据分析的框架。它将文档依次通过多个阶段进行处理与转换，最终返回计算结果。

聚合管道如何改进数据处理？

它允许在 MongoDB 内直接进行复杂数据转换与分析，减少数据传输，将处理靠近数据源，从而获得更佳性能。

MongoDB 聚合管道中常见的阶段有哪些？

常见阶段包括用于筛选的 $match、用于重塑文档的 $project、用于聚合数据的 $group，以及用于连接集合的 $lookup。

我能用 Python 操作 MongoDB 聚合管道吗？

可以的，本文教程演示了如何使用 MongoDB 官方的 Python 驱动 PyMongo 来构建并执行聚合管道。

为何阶段顺序在聚合管道中很重要？

阶段顺序对性能很重要。使用 $match 提前筛选数据，可减少后续更复杂阶段需要处理的文档数量。

作者

Bexruz (Bex) Tuychiev

我是一名数据科学内容创作者，拥有超过 2 年的经验，并在 Medium 上拥有较大的读者群。我喜欢用略带讽刺的笔调撰写有关 AI 和机器学习的深度文章——毕竟总得想点办法让这些内容不那么枯燥。我已发表 130 多篇文章，并制作了一门 DataCamp 课程，另有一门正在筹备中。我的内容已被超过 500 万人次阅读，其中有 2 万人在 Medium 和 LinkedIn 上成为关注者。

主题

MongoDB 数据工程

主题

技术

使用 Python 与 PyMongo 的 MongoDB 聚合管道教程

使用 AI 探索

什么是 MongoDB 中的聚合管道？

管道架构与文档流

阶段类型与功能分类

为本教程搭建环境

安装 PyMongo

获取示例数据集

连接到 MongoDB

深入理解 MongoDB 聚合管道

使用 $match 筛选数据

使用 $project、$sort、$limit 与 $skip 重塑文档

使用 $group 分组与聚合

使用 $lookup 与 $unwind 连接集合

组合多个阶段以进行复杂分析

高级管道模式

使用 $facet 进行多维分析

使用窗口函数进行时序分析

结论

MongoDB 聚合管道常见问答

主题

DataCamp 热门课程

Python 中的 MongoDB 入门

数据工程师在 Python 中

NoSQL 概念

主题

技术

DataCamp 热门课程

Python 中的 MongoDB 入门

数据工程师在 Python 中

NoSQL 概念

什么是 MongoDB 中的聚合管道？

管道架构与文档流

阶段类型与功能分类

为本教程搭建环境

安装 PyMongo

获取示例数据集

连接到 MongoDB

深入理解 MongoDB 聚合管道

使用 $match 筛选数据

使用 $project、$sort、$limit 与 $skip 重塑文档

使用 $group 分组与聚合

使用 $lookup 与 $unwind 连接集合

组合多个阶段以进行复杂分析

高级管道模式

使用 $facet 进行多维分析

使用窗口函数进行时序分析

结论

MongoDB 聚合管道常见问答

主题

DataCamp 热门课程

Python 中的 MongoDB 入门

数据工程师 在 Python 中

NoSQL 概念

DataCamp 热门课程

Python 中的 MongoDB 入门

数据工程师 在 Python 中

NoSQL 概念

数据工程师在 Python 中

数据工程师在 Python 中