限制*大型* Django 查询集中的内存使用

Question

我有一个任务需要每隔一段时间在数据库中的“大多数”对象上运行一次（每天一次，每周一次，等等）。基本上这意味着我有一些看起来像这样在它自己的线程中运行的查询。

for model_instance in SomeModel.objects.all():
    do_something(model_instance)

（请注意，它实际上是一个 filter() 而不是 all()，但尽管如此，我最终仍然选择了一组非常大的对象。）

我遇到的问题是运行后有一段时间，该线程被我的托管提供商杀死，因为我使用了太多内存。我假设所有这些内存使用都发生了，因为即使我的查询返回的QuerySet对象最初的内存占用量非常小，但它最终会随着当我迭代每个 model_instance 时，QuerySet 对象会缓存它们。

我的问题是，“以内存有效的方式迭代数据库中几乎每个 SomeModel 的最佳方法是什么？”或者也许我的问题是“如何从 django 查询集中‘取消缓存’模型实例？”

编辑：我实际上正在使用查询集的结果来构建一系列新对象。因此，我最终根本不会更新查询的对象。

Answer 1

如何使用此处记录的 django core 的 Paginator 和 Page 对象：

https://docs.djangoproject .com/en/dev/topics/pagination/

像这样：

from django.core.paginator import Paginator
from djangoapp.models import SomeModel

queryset = SomeModel.objects.all().order_by("pk") # sorting is necessary for correctness
paginator = Paginator(queryset, 1000) # chunks of 1000

for page_idx in range(1, paginator.num_pages):
    for row in paginator.page(page_idx).object_list:
        # here you can do what you want with the row
    print "done processing page %s" % page_idx

Answer 2

所以我实际上最终做的是构建一些可以“包装”查询集的东西。它的工作原理是使用切片语法对查询集进行深度复制 - 例如，some_queryset[15:45]--但是当切片完全迭代后，它会生成原始 QuerySet 的另一个深层副本。这意味着只有“此”特定切片中返回的对象集存储在内存中。

class MemorySavingQuerysetIterator(object):

    def __init__(self,queryset,max_obj_num=1000):
        self._base_queryset = queryset
        self._generator = self._setup()
        self.max_obj_num = max_obj_num

    def _setup(self):
        for i in xrange(0,self._base_queryset.count(),self.max_obj_num):
            # By making a copy of of the queryset and using that to actually access
            # the objects we ensure that there are only `max_obj_num` objects in
            # memory at any given time
            smaller_queryset = copy.deepcopy(self._base_queryset)[i:i+self.max_obj_num]
            logger.debug('Grabbing next %s objects from DB' % self.max_obj_num)
            for obj in smaller_queryset.iterator():
                yield obj

    def __iter__(self):
        return self

    def next(self):
        return self._generator.next()

因此，而不是...

for obj in SomeObject.objects.filter(foo='bar'): <-- Something that returns *a lot* of Objects
    do_something(obj);

您会做...

for obj in MemorySavingQuerysetIterator(in SomeObject.objects.filter(foo='bar')):
    do_something(obj);

请注意，这样做的目的是为了Python解释器中节省内存。它本质上是通过进行更多数据库查询来实现这一点的。通常人们会尝试做完全相反的事情——即，尽可能地减少数据库查询，而不考虑内存使用情况。希望有人会发现这很有用。

Answer 3

您不能使用 Model.objects.all().iterator() ，因为它会立即获取表中的所有元素。您也不能使用 Model.objects.all()[offset:offset+pagesize]，因为它会缓存结果。两者都会超出您的内存限制。

我尝试混合使用这两种解决方案，并且它有效：

offset = 0
pagesize = 1000
count = Model.objects.all().count()
while offset < count:
    for m in Model.objects.all()[offset : offset + pagesize].iterator:
        do_something with m
    offset += pagesize

更改 pagesize 以满足您的要求，并可选择将 [offset : offset + pagesize] 更改为 [offset * pagesize : (offset + 1) * pagesize] 习语（如果它更适合您）。当然，还将 Model 替换为您的实际型号名称。

Answer 4

许多解决方案通过对查询集进行切片来实现 SQL OFFSET 和 LIMIT。正如斯特凡诺指出的那样，对于更大的数据集，这变得非常低效。处理此问题的正确方法是使用服务器端光标来跟踪 OFFSET。

本机服务器端光标支持正在为 django 开发。在准备就绪之前，如果您将 postgres 与 psycopg2 后端一起使用，这里有一个简单的实现：

def server_cursor_query(Table):
    table_name = Table._meta.db_table

    # There must be an existing connection before creating a server-side cursor
    if connection.connection is None:
        dummy_cursor = connection.cursor()  # not a server-side cursor

    # Optionally keep track of the columns so that we can return a QuerySet. However,
    # if your table has foreign keys, you may need to rename them appropriately
    columns = [x.name for x in Table._meta.local_fields]

    cursor = connection.connection.cursor(name='gigantic_cursor')) # a server-side
                                                                   # cursor

    with transaction.atomic():
        cursor.execute('SELECT {} FROM {} WHERE id={}'.format(
            ', '.join(columns), table_name, id))

        while True:
            rows = cursor.fetchmany(1000)

                if not rows:
                    break

                for row in rows:
                    fields = dict(zip(columns, row))
                    yield Table(**fields)

请参阅 这篇博文 很好地解释了 django 中大型查询的内存问题。

Answer 5

简短回答

如果您使用 PostgreSQL 或 Oracle，则可以使用 Django 的 内置迭代器：

queryset.iterator(chunk_size=1000)

这会导致 Django 使用 服务器端游标，而不是在迭代查询集时缓存模型。从 Django 4.1 开始，这甚至可以与 prefetch_lated 一起使用。

对于其他数据库，您可以使用以下内容：

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 内存（糟糕）

Django 迭代器

Django 迭代器（至少从 Django 3.2 开始）修复了内存问题，但性能提升较小。据推测，这是因为 Django 管理缓存的时间减少了。

assert sum(i.id for i in MyModel.objects.all().iterator(chunk_size=1000)) == x

上电时间：3.11 秒，<1MB 内存

自定义迭代器

自然的比较点是尝试通过逐渐增加对主键的查询来自己进行分页。虽然这是对简单迭代的改进，因为它具有恒定的内存，但它实际上在速度上输给了 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 秒，<1MB 内存

自定义分页功能

使用自定义迭代的主要原因是这样您可以获得分页结果。此函数对于在仅使用常量内存时插入批量更新非常有用。在我的测试中，它比 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 秒，<1MB 内存

替代自定义分页功能

鉴于 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 秒，<1MB 内存

---

不良方法

以下是您永远不应该使用的方法（问题中建议了其中许多方法）以及原因。

不要对无序查询集使用切片

无论你做什么，都不要对无序查询集进行切片。这不能正确地迭代表。原因是切片操作根据您的查询集执行 SQL 限制 + 偏移量查询，并且 django 查询集没有顺序保证，除非您使用 order_by.此外，PostgreSQL 没有默认的排序依据，并且 Postgres 文档特别警告不要使用限制 + 没有 order by 的偏移量。因此，每次获取切片时，您都会获得表的不确定性切片，这意味着您的切片可能不是重叠并且不会覆盖它们之间表格的所有行。根据我的经验，只有当您在进行迭代时有其他东西正在修改表中的数据时，才会发生这种情况，这只会使这个问题更加严重，因为这意味着如果您单独测试代码，则该错误可能不会出现。

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），<1MB 内存

不要使用 Django 的分页器进行全表迭代

Django 带有内置的 分页器。人们可能会认为这适合对数据库进行分页迭代，但事实并非如此。 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），<1MB 内存

Answer 6

有一个 django 片段：

http://djangosnippets.org/snippets/1949/

它迭代通过生成原始查询集的较小“块”的行来对查询集进行处理。它最终使用的内存显着减少，同时允许您调整速度。我在我的一个项目中使用它。

Answer 7

我正在继续研究，看起来我想做相当于 SQL OFFSET 和 LIMIT 的操作，根据 Django 文档关于限制查询集 意味着我想使用切片语法，例如 SomeModel.objects.all()[15:25]

所以现在我想也许这样的东西就是我正在寻找的：

# Figure out the number of objects I can safely hold in memory
# I'll just say 100 for right now
number_of_objects = 100 
count = SomeModel.objects.all().count():
for i in xrange(0,count,number_of_objects):
    smaller_queryset = SomeModel.objects.all()[i:i+number_of_objects]
    for model_instance in smaller_queryset:
        do_something(model_instance)

根据我的计算，这将使 smaller_queryset 永远不会变得太大。

Answer 8

以下方法不使用昂贵的数据库偏移量查询，并且避免计算页码，从而更加高效。
文档字符串中指定的限制。

def queryset_pk_iterator(queryset, batch_size=1000):
    """
    Iterator that splits the queryset into batches to reduce memory consumption.
    Useful in cases where builtin .iterator() method of the queryset skips the "prefetch_related" optimization.

    :param queryset: Queryset to iterate over. The supplied queryset must not specify order and limit/offset.
        Queryset objects must have a monotonically increasing and ordering primary key.
    :param batch_size: Size of the batches into which to split the queryset.
    :return: iterator object
    """
    pk = None
    while True:
        batch_queryset = queryset.order_by('pk')
        if pk is not None:
            batch_queryset = batch_queryset.filter(pk__gt=pk)
        batch_queryset = batch_queryset[:batch_size]
        obj = None
        for obj in batch_queryset:
            yield obj
        if obj is None:
            return
        pk = obj.pk