Python pymongo场景下优化传感器逐小时温度均值查询效率的方法

原逻辑的核心性能损耗 #

• 每次统计都独立新建Mongo客户端连接，24次查询对应24次TCP握手、认证开销，占了总耗时的大头
• 分24次逐小时查询，每次都重复扫描相同时间范围的文档，总文档扫描量是单次全量统计的24倍，大量重复计算
• 循环过程中24次网络往返Mongo服务端，网络延迟进一步放大耗时

优化思路（无需新增任何索引） #

直接将24次独立查询合并为单次聚合查询，在服务端一次性完成所有小时区间的温度均值计算，再在内存中对齐成需要的数组结构。

• 全程不需要额外创建索引，就算当前集合没有针对created_at的索引，单次扫描的总文档量（全天约8640条）也仅为原24次循环扫描量的1/24
• 去掉24次重复建连、网络往返的开销，整体耗时可以从30秒以上降到百毫秒级

具体实现 #

1. 全局复用Mongo客户端连接，避免重复初始化的开销
2. 构造聚合管道时，先用$match一次性筛出全天时间范围的所有待统计文档（总扫描文档量和原逻辑完全一致，无额外数据开销）
3. 服务端按「设备编号 + 小时截断时间」自动分组，直接计算每个小时区间的平均温度
4. 内存中预先生成24个小时区间的索引结构，将聚合返回的结果对齐到对应区间，最终生成顺序排列的结果数组

优化后完整代码 #

from datetime import datetime, timedelta
import pymongo

# 全局复用Mongo连接，仅初始化一次
myclient = pymongo.MongoClient("mongodb://localhost:27017/")
mydb = myclient["你的目标数据库名"]
mycol = mydb["你的目标集合名"]

# 定义统计时间范围
start = datetime(2020, 4, 12, 7, 0, 0, 38000)
end = datetime(2020, 4, 13, 7, 0, 0, 38000)

# 单次执行全量聚合
pipeline = [
    # 1. 一次性筛出整个统计范围的文档
    {
        "$match": {
            "created_at": {"$gte": start, "$lt": end}
        }
    },
    # 2. 预处理字段：温度转数值、时间按小时截断
    {
        "$set": {
            "numeric_temp": {"$toInt": "$air_temperature"},
            # 按小时截断时间作为分组桶，指定东八区时区避免时间偏差
            "hour_bucket": {
                "$dateTrunc": {
                    "date": "$created_at",
                    "unit": "hour",
                    "timezone": "Asia/Shanghai"
                }
            }
        }
    },
    # 3. 按 设备+小时 分组计算平均温度
    {
        "$group": {
            "_id": {
                "device": "$device_number",
                "hour_start": "$hour_bucket"
            },
            "avg_temp": {"$avg": "$numeric_temp"}
        }
    }
]
agg_result = list(mycol.aggregate(pipeline))

# 预生成24个小时的区间索引，保证结果结构完整
hour_index = {}
current_start = start
for _ in range(24):
    hour_end = current_start + timedelta(hours=1)
    hour_index[current_start] = {
        "data": [],
        "start": current_start.isoformat(),
        "end_date": hour_end.isoformat()
    }
    current_start = hour_end

# 把聚合结果填充到对应区间
for item in agg_result:
    hour_start = item["_id"]["hour_start"]
    if hour_start in hour_index:
        hour_index[hour_start]["data"].append({
            "_id": item["_id"]["device"],
            "value": item["avg_temp"]
        })

# 按时间顺序生成最终要求的avg_arr数组
avg_arr = [hour_index[key] for key in sorted(hour_index.keys())]

兼容低版本Mongo的调整 #

如果你的Mongo版本低于5.0，不支持$dateTrunc函数，只需要把$set阶段的hour_bucket计算替换为日期字符串格式化即可，逻辑完全一致：

# 替换后的hour_bucket逻辑（Mongo 4.x版本兼容）
"hour_bucket": {
    "$dateToString": {
        "format": "%Y-%m-%d %H:00:00",
        "date": "$created_at",
        "timezone": "Asia/Shanghai"
    }
}

注意替换后需要同步调整hour_index的字典key为对应的字符串格式，保持索引匹配逻辑一致即可。

额外修正的边界问题 #

原代码中$match使用$lte匹配end时间点，会把end时刻的边界文档错误统计到最后一个小时，优化后改为$lt匹配，和你定义的1小时区间逻辑完全对齐，不会出现统计偏差。

内容的提问来源于stack exchange，提问作者S_M