Django中高效获取多对多关联表各serial对应最新timestamp方法
批量查询优化方案
你当前的实现属于典型的N+1查询问题:待查询的serial列表有多少个元素,就会产生多少次独立的数据库请求,当列表长度达到上千、上万级别时,网络IO和数据库连接开销会被无限放大,性能极差。通过批量查询的方式可以把总请求数压缩到2次以内,性能提升可达百倍以上。
通用兼容方案(支持所有数据库)
这个方案不绑定特定数据库,适配MySQL、PostgreSQL、SQLite等所有Django支持的数据库,全程仅产生2次数据库查询:
from django.db.models import Max, Q import pandas as pd # 提取所有待查询的serial,提前去重减少无效计算 target_serials = list(df_hr1['Miss'].dropna().unique()) # 第一次数据库交互:批量查询每个serial对应的最大时间戳 serial_max_ts = dict( RFIDInventorySerials.objects .filter(serial__in=target_serials) .values('serial') .annotate(max_ts=Max('timestamp')) .values_list('serial', 'max_ts') ) # 构建组合查询条件,批量拿到每个serial对应最新时间的完整记录 q = Q() for serial, ts in serial_max_ts.items(): q |= Q(serial=serial, timestamp=ts) latest_records = RFIDInventorySerials.objects.filter(q) # 转成serial为key的字典,方便后续快速取值 serial_record_map = {r.serial: r for r in latest_records} # 按原DataFrame的顺序提取需要的timestamp值,全程无额外数据库请求 last_seen = [] for v in df_hr1['Miss']: record = serial_record_map.get(v) # 兼容serial无对应记录的边界情况,避免报错 last_seen.append(record.timestamp if record else None)
PostgreSQL专属优化方案
如果你用的数据库是PostgreSQL,可以用Django支持的distinct on语法,仅需1次数据库查询就能拿到结果,效率更高:
# 仅PostgreSQL环境可用 latest_records = ( RFIDInventorySerials.objects .filter(serial__in=target_serials) .order_by('serial', '-timestamp') .distinct('serial') ) serial_record_map = {r.serial: r for r in latest_records} # 后续取值逻辑和通用方案完全一致
额外性能优化建议
- 给查询场景加针对性索引,能进一步把查询耗时压到毫秒级,修改模型添加联合索引:
class RFIDInventorySerials(models.Model): serial = models.CharField(max_length=50, blank=True, null=True, default=0) coordinate = models.CharField(max_length=100, blank=True, null=True, default=0) timestamp = models.DateTimeField(default=datetime.now) class Meta: indexes = [ # 针对按serial查最新timestamp的场景建联合索引 models.Index(fields=['serial', '-timestamp']), ]
- 原代码中
last.start属于字段笔误,你的模型里不存在start字段,对应的时间字段为timestamp,注意修正避免报错。 - 如果待查询的serial列表存在大量重复值,一定要提前去重,重复的serial不会产生额外的查询开销,只需要最后映射取值即可。
内容的提问来源于stack exchange,提问作者Rahul Sharma
相关产品推荐
相关产品推荐

