如何高效过滤含1000万条记录的Django Queryset？

核心问题分析 #

你当前的实现把1000万条记录全量拉到本地内存，再用Shapely逐条判断点是否在多边形内，IO和单线程CPU计算是主要耗时原因。最优思路是把过滤逻辑移到数据库层面，利用数据库的地理索引加速查询；如果无法改模型，再考虑本地并行优化。

一、改用Django地理字段（长期最优方案） #

如果你的数据库支持PostGIS（PostgreSQL）或其他地理扩展，这是最快的解决方式：

1. 改造模型，添加地理字段
   把分散的经纬度字段替换/补充为PointField，支持地理空间查询：

   from django.contrib.gis.db import models
   
   class Polygon(models.Model):
       # 保留原有字段...
       latitude = models.FloatField(default=0, null=True)
       longitude = models.FloatField(default=0, null=True)
       # 添加地理点字段，geography=True支持球面坐标计算
       location = models.PointField(null=True, geography=True, db_index=True)
   

   执行迁移后，需把现有经纬度数据同步到location字段（可写批量脚本完成）。

2. 创建地理索引
   手动给location字段创建GIST索引（GeoDjango迁移可能自动生成，手动确认更稳妥）：

   CREATE INDEX polygon_location_gist_idx ON polygon USING GIST (location);
   

3. 数据库层面直接过滤
   把Shapely多边形转成Django支持的地理对象，直接在Queryset中过滤：

   from django.contrib.gis.geos import Polygon as GeoPolygon
   
   # 转换Shapely多边形为Django GeoPolygon
   django_polygon = GeoPolygon(list(shapely_polygon.exterior.coords))
   # 直接在数据库完成过滤，无需拉取全量数据
   filtered_polygons = Polygon.objects.filter(
       dataset_id=dataset_id,
       location__within=django_polygon
   )
   

   这种方式能利用数据库索引，查询速度会从分钟级降到秒级。

二、不修改模型，用数据库函数组合经纬度（临时兼容方案） #

如果无法修改现有模型结构，可以用数据库函数动态组合经纬度为点，再执行地理过滤：

1. 用Django数据库函数生成点字段

   from django.contrib.gis.db.models.functions import MakePoint
   from django.contrib.gis.geos import Polygon as GeoPolygon
   from django.db.models import F
   
   django_polygon = GeoPolygon(list(shapely_polygon.exterior.coords))
   filtered_polygons = Polygon.objects.filter(dataset_id=dataset_id).annotate(
       # 注意GIS标准是「经度在前，纬度在后」
       point=MakePoint(F('longitude'), F('latitude'))
   ).filter(point__within=django_polygon)
   

2. 添加函数索引加速
   为避免全表扫描，给动态生成的点创建函数索引：

   CREATE INDEX polygon_lng_lat_gist_idx ON polygon USING GIST (ST_MakePoint(longitude, latitude));
   

三、本地并行处理优化（仅当数据库无法支持地理查询时） #

如果必须在本地处理，用多进程并行计算提升效率：

import concurrent.futures
from shapely.geometry import Point

def is_point_in_polygon(pol):
    # 注意经纬度顺序
    return shapely_polygon.contains(Point(pol.longitude, pol.latitude))

all_points_poly = []
# 增大chunk_size减少数据库IO次数
chunk_size = 100000

# 利用多进程并行处理CPU密集型判断
with concurrent.futures.ProcessPoolExecutor() as executor:
    for chunk in Polygon.objects.filter(dataset_id=dataset_id).iterator(chunk_size=chunk_size):
        # 批量提交任务
        results = executor.map(is_point_in_polygon, chunk)
        # 筛选符合条件的记录
        all_points_poly.extend([pol for pol, is_valid in zip(chunk, results) if is_valid])

这种方式能利用多核CPU，比单线程处理快2-8倍（取决于CPU核心数），但仍远不如数据库层面处理高效。

为什么之前用iterator没效果？ #

iterator(chunk_size=10000)只是减少了内存占用，避免一次性加载1000万条记录到内存，但处理逻辑还是单线程串行执行，CPU是瓶颈，所以看不到性能提升。

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