简答
如果您使用 PostgreSQL 或 Oracle,则可以使用 Django 的内置迭代器 https://docs.djangoproject.com/en/dev/ref/models/querysets/#iterator:
queryset.iterator(chunk_size=1000)
这导致 Django 使用服务器端游标 https://docs.djangoproject.com/en/4.1/ref/models/querysets/#with-server-side-cursors并且在迭代查询集时不缓存模型。从 Django 4.1 开始,这甚至可以与prefetch_related
.
对于其他数据库,您可以使用以下内容:
def queryset_iterator(queryset, page_size=1000):
page = queryset.order_by("pk")[:page_size]
while page:
for obj in page:
yield obj
pk = obj.pk
page = queryset.filter(pk__gt=pk).order_by("pk")[:page_size]
如果您想要返回页面而不是单个对象以与其他优化相结合,例如bulk_update
, 用这个:
def queryset_to_pages(queryset, page_size=1000):
page = queryset.order_by("pk")[:page_size]
while page:
yield page
pk = max(obj.pk for obj in page)
page = queryset.filter(pk__gt=pk).order_by("pk")[:page_size]
PostgreSQL 性能分析
我在 Django 3.2 和 Postgres 13 上对大约 200,000 行的 PostgreSQL 表分析了多种不同的方法。对于每个查询,我将 ids 的总和相加,既确保 Django 实际检索对象,也使我能够验证查询之间迭代的正确性。所有计时都是在对相关表进行多次迭代后进行的,以最大限度地减少后续测试的缓存优势。
基本迭代
基本方法只是迭代表。这种方法的主要问题是所使用的内存量不是恒定的;它随着表的大小而增长,并且我已经看到在较大的表上内存不足。
x = sum(i.id for i in MyModel.objects.all())
挂壁时间:3.53 秒,22MB 内存(BAD)
Django迭代器
Django 迭代器(至少从 Django 3.2 开始)修复了内存问题,并带来了较小的性能提升。据推测,这是因为 Django 管理缓存的时间减少了。
assert sum(i.id for i in MyModel.objects.all().iterator(chunk_size=1000)) == x
挂载时间:3.11 秒,
自定义迭代器
自然的比较点是尝试通过逐渐增加对主键的查询来自己进行分页。虽然这是对简单迭代的改进,因为它具有恒定的内存,但它实际上在速度上输给了 Django 的内置迭代器,因为它进行了更多的数据库查询。
def queryset_iterator(queryset, page_size=1000):
page = queryset.order_by("pk")[:page_size]
while page:
for obj in page:
yield obj
pk = obj.pk
page = queryset.filter(pk__gt=pk).order_by("pk")[:page_size]
assert sum(i.id for i in queryset_iterator(MyModel.objects.all())) == x
挂载时间:3.65 秒,
自定义分页功能
使用自定义迭代的主要原因是您可以在页面中获取结果。此函数对于在仅使用常量内存时插入批量更新非常有用。在我的测试中,它比 queryset_iterator 慢一点,并且我没有一个连贯的理论来解释为什么,但速度减慢并不严重。
def queryset_to_pages(queryset, page_size=1000):
page = queryset.order_by("pk")[:page_size]
while page:
yield page
pk = max(obj.pk for obj in page)
page = queryset.filter(pk__gt=pk).order_by("pk")[:page_size]
assert sum(i.id for page in queryset_to_pages(MyModel.objects.all()) for i in page) == x
挂载时间:4.49 秒,
替代自定义分页功能
鉴于 Django 的查询集迭代器比我们自己进行分页更快,因此可以交替实现查询集分页器来使用它。它比我们自己进行分页要快一点,但实现起来比较混乱。可读性很重要,这就是为什么我个人更喜欢前一个分页功能,但如果您的查询集在结果中没有主键(无论出于何种原因),这个功能可能会更好。
def queryset_to_pages2(queryset, page_size=1000):
page = []
page_count = 0
for obj in queryset.iterator():
page.append(obj)
page_count += 1
if page_count == page_size:
yield page
page = []
page_count = 0
yield page
assert sum(i.id for page in queryset_to_pages2(MyModel.objects.all()) for i in page) == x
挂载时间:4.33 秒,
不良方法
以下是您永远不应该使用的方法(问题中建议了其中许多方法)以及原因。
不要对无序查询集使用切片
无论你做什么,都不要对无序查询集进行切片。这不能正确地迭代表。这样做的原因是切片操作根据您的查询集执行 SQL limit + offset 查询,并且 django 查询集没有顺序保证,除非您使用order_by
。此外,PostgreSQL 没有默认的 order by,并且Postgres 文档特别警告不要使用 limit + offset 而不使用 order by https://www.postgresql.org/docs/current/queries-limit.html。因此,每次获取切片时,您都会获得表的不确定性切片,这意味着你的切片可能不重叠 https://dba.stackexchange.com/a/138210并且不会覆盖它们之间表格的所有行。根据我的经验,只有当您在进行迭代时有其他东西正在修改表中的数据时,才会发生这种情况,这只会让这个问题更加严重,因为这意味着如果您单独测试代码,则该错误可能不会出现。
def very_bad_iterator(queryset, page_size=1000):
counter = 0
count = queryset.count()
while counter < count:
for model in queryset[counter:counter+page_size].iterator():
yield model
counter += page_size
assert sum(i.id for i in very_bad_iterator(MyModel.objects.all())) == x
断言错误;即计算的结果不正确!
一般情况下不要使用切片进行全表迭代
即使我们对查询集进行排序,从性能角度来看,列表切片也是很糟糕的。这是因为 SQL offset 是线性时间操作,这意味着表的 limit + offset 分页迭代将是二次时间,这是您绝对不希望的。
def bad_iterator(queryset, page_size=1000):
counter = 0
count = queryset.count()
while counter < count:
for model in queryset.order_by("id")[counter:counter+page_size].iterator():
yield model
counter += page_size
assert sum(i.id for i in bad_iterator(MyModel.objects.all())) == x
挂载时间:15 秒(BAD),
不要使用 Django 的分页器进行全表迭代
Django 带有一个内置的分页器 https://docs.djangoproject.com/en/dev/topics/pagination/。人们可能会认为这适合对数据库进行分页迭代,但事实并非如此。 Paginator 的目的是将单页结果返回到 UI 或 API 端点。它比任何迭代表的好方法都要慢得多。
from django.core.paginator import Paginator
def bad_paged_iterator(queryset, page_size=1000):
p = Paginator(queryset.order_by("pk"), page_size)
for i in p.page_range:
yield p.get_page(i)
assert sum(i.id for page in bad_paged_iterator(MyModel.objects.all()) for i in page) == x
挂载时间:13.1 秒(BAD),