Django中大表的内存效率(常量)和速度优化迭代

Question

我有张很大的桌子。它目前在一个MySQL数据库中。我用django。

我需要在上迭代表中的每个元素，以预先计算一些特定的数据(也许如果我更好的话，我可以这样做，但这不是重点)。

我希望尽可能快地保持迭代，并不断地使用内存。

因为在姜戈QuerySet和为什么迭代一个大型Django QuerySet要消耗大量内存？中已经很清楚了，对django中的所有对象进行简单的迭代就会杀死机器，因为它将从数据库中检索所有对象。

走向解决

首先，为了减少内存消耗，您应该确保调试是假的(或者对游标：在保持settings.DEBUG的同时关闭SQL日志记录？进行猴子补丁)，以确保django没有在connections中存储用于调试的内容。

但即便如此，

for model in Model.objects.all()

是一次拒绝。

即使是稍微改进过的形式也不行：

for model in Model.objects.all().iterator()

使用iterator()将节省一些内存，因为它不会在内部存储缓存的结果(尽管不一定在PostgreSQL上！)；但是显然仍然会从数据库中检索整个对象。

天真的解决方案

第一个问题的解决办法是根据一个chunk_size计数器对结果进行切片。编写它有几种方法，但基本上它们都可以归结为SQL中的OFFSET + LIMIT查询。

类似于：

qs = Model.objects.all()
counter = 0
count = qs.count()
while counter < count:     
    for model in qs[counter:counter+chunk_size].iterator()
        yield model
    counter += chunk_size

虽然这是内存效率(内存使用与chunk_size成正比)，但在速度方面确实很差:随着偏移量的增加，MySQL和PostgreSQL (很可能大多数DBs)都会开始窒息和减速。

更好的解决办法

一个更好的解决方案是由蒂埃里·谢伦巴赫在这个职位中提供的。它在PK上过滤，这比抵消快得多(多快可能取决于DB)

pk = 0
last_pk = qs.order_by('-pk')[0].pk
queryset = qs.order_by('pk')
while pk < last_pk:
    for row in qs.filter(pk__gt=pk)[:chunksize]:
        pk = row.pk
        yield row
    gc.collect()

这开始令人满意了。现在内存= O(C)，并加速~= O(N)

与“更好”解决方案有关的问题

只有当PK在QuerySet中可用时，更好的解决方案才能工作。不幸的是，情况并不总是如此，特别是当QuerySet包含不同(group_by)和/或值(ValueQuerySet)的组合时。

在这种情况下，不能使用“更好的解决办法”。

我们能做得更好吗？

现在，我想知道我们是否可以更快地解决关于没有PK的QuerySets的问题。也许使用我在其他答案中找到的东西，但只在纯SQL中使用:使用游标。

由于我对原始SQL (特别是Django中的SQL)处理得很差，这里有一个真正的问题：

如何为大型表构建一个更好的Django QuerySet Iterator

我所读到的是，我们应该使用服务器端游标(显然(参见引用)，使用标准Django游标不会达到相同的结果，因为默认情况下，python和心理学连接器都缓存结果)。

这真的是一个更快(和/或更有效)的解决方案吗？

这可以使用django中的原始SQL来完成吗？还是应该根据数据库连接器编写特定的python代码？

PostgreSQL和MySQL中的服务器端游标

这是我目前所能得到的.

一个Django chunked_iterator()

当然，最好的方法应该是queryset.iterator()，而不是iterate(queryset)，并且是django核心的一部分，或者至少是一个可插拔的应用程序。

更新，这要感谢注释中的"T“，用于查找包含一些附加信息的django票。连接器行为的差异使得它可能是最好的解决方案，那就是创建一个特定的chunked方法，而不是透明地扩展iterator (听起来是一种很好的方法)。一个实现存根存在，但是已经有一年没有任何工作了，而且看起来作者还没有准备好跳到上面。

其他参考文献：

1. 为什么MYSQL更高的限制抵消了查询慢下来？
2. 如何通过限制子句中的大偏移量加速MySQL查询？
3. http://explainextended.com/2009/10/23/mysql-order-by-limit-performance-late-row-lookups/
4. postgresql:偏移+限制变得非常慢
5. 提高PostgreSQL中的偏置性能
6. http://www.depesz.com/2011/05/20/pagination-with-fixed-order/
7. 如何在python中获得逐行的MySQL ResultSet服务器端游标在MySQL中

编辑：

Django 1.6正在添加持久数据库连接

Django数据库持久连接

在某些情况下，这应便于使用游标。不过，我目前的技能(以及学习的时间)还不足以实现这样的解决方案。

而且，“更好的解决方案”肯定不是在所有情况下都起作用，也不能作为一种通用的方法使用，只有一个存根可以逐个进行调整.

Answer 1

简短回答

如果您使用的是PostgreSQL或Oracle，您可以使用Django的内建迭代器

queryset.iterator(chunk_size=1000)

这导致Django在遍历queryset时使用服务器端游标而不是缓存模型。从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的性能分析

我在一个PostgreSQL表上描述了一些不同的方法，在Django 3.2和Postgres 13上大约有20万行。对于每个查询，我都将ids之和相加，这两个方法都是为了确保Django实际检索对象，并且可以验证查询之间迭代的正确性。所有的时间都是在对表进行几次迭代之后才得到的，以尽量减少后续测试的缓存优势。

基本迭代

基本方法只是对表进行迭代。这种方法的主要问题是，使用的内存量不是固定的，而是随着表的大小而增加的，而且我在较大的表上看到内存不足。

x = sum(i.id for i in MyModel.objects.all())

墙时间:3.53s，内存22 of (坏)

Django Iterator

Django迭代器(至少在Django 3.2时)修复了内存问题，但性能方面的好处不大。这大概是Django花费更少的时间来管理缓存的结果。

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

墙时间:3.11s，内存<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.65s，<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.49s，内存<1MB

可选自定义寻呼功能

鉴于Django的queryset迭代器比自己分页更快，所以可以交替实现queryset寻呼机来使用它。它比我们自己进行分页要快一些，但是实现比较混乱。可读性很重要，这就是为什么我的个人偏好是前面的分页函数，但是如果查询集没有主键(不管是什么原因)，这个函数会更好。

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

不良做法

以下是您不应该使用的方法(其中许多方法在问题中提出)以及原因。

不要在无序的Queryset上使用切片

不管你做什么，都不要分割无序的查询集。这不能正确地迭代表。的原因是，片操作基于查询集执行SQL +偏移查询，而django查询集除非使用order_by，否则没有订单保证。此外，PostgreSQL没有默认的order和Postgres文档特别警告不要使用限制+偏移量而不使用。因此，每次您取一个片时，都会得到表的一个不确定的部分，这意味着你的切片可能不是重叠的，并且不会覆盖表之间的所有行。在我的经验中，只有在执行迭代时其他东西正在修改表中的数据时才会发生这种情况，这只会使这个问题更加有害，因为这意味着如果孤立地测试代码，错误可能不会出现。

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偏移量是一个线性时间操作，这意味着表的限制+偏移量分页迭代将是二次时间，这是您绝对不想要的。

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

墙时间: 15s (坏)，<1MB内存

不要在全表迭代中使用Django的分页器

Django附带了一个内置的分页器。人们可能会倾向于认为，这样做适合于对数据库进行分页迭代，但事实并非如此。分页器的目的是将结果的单个页面返回到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.1s(坏)，内存<1MB

Answer 2

基本答案是：与服务器端游标一起使用原始SQL。

可悲的是，在Django 1.5.2之前，还没有正式的方法来创建服务器端的MySQL游标(不确定其他数据库引擎)。所以我写了一些魔法代码来解决这个问题。

对于Django 1.5.2和MySQLdb 1.2.4，下面的代码可以工作。而且，这也是很好的评论。

警告：--这不是基于公共API，因此它可能会在以后的版本中崩溃。

# This script should be tested under a Django shell, e.g., ./manage.py shell

from types import MethodType

import MySQLdb.cursors
import MySQLdb.connections
from django.db import connection
from django.db.backends.util import CursorDebugWrapper


def close_sscursor(self):
    """An instance method which replace close() method of the old cursor.

    Closing the server-side cursor with the original close() method will be
    quite slow and memory-intensive if the large result set was not exhausted,
    because fetchall() will be called internally to get the remaining records.
    Notice that the close() method is also called when the cursor is garbage 
    collected.

    This method is more efficient on closing the cursor, but if the result set
    is not fully iterated, the next cursor created from the same connection
    won't work properly. You can avoid this by either (1) close the connection 
    before creating a new cursor, (2) iterate the result set before closing 
    the server-side cursor.
    """
    if isinstance(self, CursorDebugWrapper):
        self.cursor.cursor.connection = None
    else:
        # This is for CursorWrapper object
        self.cursor.connection = None


def get_sscursor(connection, cursorclass=MySQLdb.cursors.SSCursor):
    """Get a server-side MySQL cursor."""
    if connection.settings_dict['ENGINE'] != 'django.db.backends.mysql':
        raise NotImplementedError('Only MySQL engine is supported')
    cursor = connection.cursor()
    if isinstance(cursor, CursorDebugWrapper):
        # Get the real MySQLdb.connections.Connection object
        conn = cursor.cursor.cursor.connection
        # Replace the internal client-side cursor with a sever-side cursor
        cursor.cursor.cursor = conn.cursor(cursorclass=cursorclass)
    else:
        # This is for CursorWrapper object
        conn = cursor.cursor.connection
        cursor.cursor = conn.cursor(cursorclass=cursorclass)
    # Replace the old close() method
    cursor.close = MethodType(close_sscursor, cursor)
    return cursor


# Get the server-side cursor
cursor = get_sscursor(connection)

# Run a query with a large result set. Notice that the memory consumption is low.
cursor.execute('SELECT * FROM million_record_table')

# Fetch a single row, fetchmany() rows or iterate it via "for row in cursor:"
cursor.fetchone()

# You can interrupt the iteration at any time. This calls the new close() method,
# so no warning is shown.
cursor.close()

# Connection must be close to let new cursors work properly. see comments of
# close_sscursor().
connection.close()

Answer 3

还有另一种选择。它不会使迭代速度更快(实际上，它可能会减慢迭代速度)，但它会使它使用的内存少得多。根据您的需要，这可能是合适的。

large_qs = MyModel.objects.all().values_list("id", flat=True)
for model_id in large_qs:
    model_object = MyModel.objects.get(id=model_id)
    # do whatever you need to do with the model here

只将ids加载到内存中，并根据需要检索和丢弃对象。请注意增加的数据库负载和较慢的运行时，这两者都是为了减少内存使用而作出的权衡。

在工作者实例上运行异步调度任务时，我使用了这种方法，如果这些任务运行速度慢并不重要，但是如果它们试图使用过多的内存，它们可能会导致实例崩溃，从而中止进程。