在 Django 中的大表上的内存效率（常量）和速度优化迭代

Posted 2023-02-24

【中文标题】在 Django 中的大表上的内存效率（常量）和速度优化迭代

【英文标题】：Memory efficient (constant) and speed optimized iteration over a large table in Django

【发布时间】：2012-12-18 03:21:29

【问题描述】：

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

我需要遍历表的 每个 元素以预先计算一些特定数据（也许如果我做得更好，我可以这样做，但这不是重点）。

我希望在不断使用内存的情况下尽可能快地进行迭代。

正如 Limiting Memory Use in a *Large* Django QuerySet 和 Why is iterating through a large Django QuerySet consuming massive amounts of memory? 中已经明确的那样，对 django 中所有对象的简单迭代将杀死机器，因为它将从数据库中检索所有对象。

寻求解决方案

首先，为了减少内存消耗，您应该确保 DEBUG 为 False（或猴子补丁光标：turn off SQL logging while keeping settings.DEBUG?），以确保 django 不会在 connections 中存储内容以进行调试。

但即便如此，

for model in Model.objects.all()

是不行的。

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

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

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

一个简单的解决方案

solution in the first question 是通过chunk_size 根据计数器对结果进行切片。有几种写法，但基本上都归结为 SQL 中的 OFFSET + LIMIT 查询。

类似：

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

虽然这是高效的内存（与chunk_size 成比例的恒定内存使用），但它在速度方面确实很差：随着 OFFSET 的增长，MySQL 和 PostgreSQL（以及可能大多数 DB）都将开始阻塞和减速。

更好的解决方案

Thierry Schellenbach 在this post 中提供了更好的解决方案。 它在 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 的查询集的问题。 也许使用我在其他答案中找到的东西，但仅限于纯 SQL：使用 cursors。

由于我对原始 SQL 很不擅长，尤其是在 Django 中，所以真正的问题来了：

我们如何为大表构建更好的 Django QuerySet 迭代器

从我读到的内容中，我们应该使用服务器端游标（显然（参见参考资料）使用标准 Django 游标不会达到相同的结果，因为默认情况下，python-MySQL 和 psycopg 连接器都会缓存结果）。

这真的会是一个更快（和/或更高效）的解决方案吗？

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

PostgreSQL 和 MySQL 中的服务器端光标

暂时就这么多吧……

一个 Django chunked_iterator()

当然，最好的办法是让这个方法作为queryset.iterator() 工作，而不是iterate(queryset)，并成为 django 核心的一部分，或者至少是一个可插入的应用程序。

更新感谢 cmets 中的“T”找到带有一些附加信息的 django ticket。连接器行为的差异使得最好的解决方案可能是创建一个特定的chunked 方法，而不是透明地扩展iterator（对我来说听起来是个好方法）。 一个实现存根exists，但一年内没有任何工作，而且看起来作者还没有准备好继续。

其他参考：

Why does MYSQL higher LIMIT offset slow the query down? How can I speed up a MySQL query with a large offset in the LIMIT clause? http://explainextended.com/2009/10/23/mysql-order-by-limit-performance-late-row-lookups/ postgresql: offset + limit gets to be very slow Improving OFFSET performance in PostgreSQL http://www.depesz.com/2011/05/20/pagination-with-fixed-order/ How to get a row-by-row MySQL ResultSet in python MySQL 中的服务器端游标

编辑：

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

Django Database Persistent Connections

这应该有助于在某些情况下使用游标。仍然超出了我目前的技能（和学习时间）如何实施这样的解决方案..

另外，“更好的解决方案”绝对不是在所有情况下都有效，不能用作通用方法，只能根据具体情况进行调整...

【问题讨论】：

哇，这是一个经过深入研究的问题！ :)

谢谢@DanielEriksson，我以为我自己就能搞定这一切，但我还没到...

哦，其他解决方案涉及构建自定义索引（例如，请参阅分页解决方案），但我希望有一个更通用的解决方案

我认为这实际上正在完成，检查一下：code.djangoproject.com/ticket/16614 似乎对于 .iterator 和类似的用例，光标现在将是一个 SSCursor，这就是你想要的的默认值，这将透明地发生。

@t.dubrownik 哇！我搜索的人，但我没有在最明显的地方找到那个。那里有宝贵的额外信息，尽管对于这是否可以透明地发生或通过单独的方法分块发生尚未完全达成共识；不幸的是，这听起来像是发展停滞了。让我们看看我们是否收集了一些有趣的东西。我会去更新我的问题:)

【参考方案1】：

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

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

对于 Django 1.5.2 和 MySQLdb 1.2.4，以下代码将起作用。另外，它的评论很好。

警告：这不是基于公共 API，因此它可能会在未来的 Django 版本中中断。

# 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()

【参考方案2】：

简单回答

如果您只需要迭代表本身而不做任何花哨的事情，Django 附带了一个builtin iterator：

queryset.iterator()

这会导致 Django 清理它自己的缓存以减少内存使用。请注意，对于真正的大表，这可能还不够。

---

复杂答案

如果您要对每个对象执行更复杂的操作或拥有大量数据，则必须自己编写。下面是一个查询集迭代器，它将查询集拆分为块，并且不会比基本迭代器慢多少（它将是数据库查询的线性数量，而不是 1，但每 1,000 行只有一个查询）。此函数按主键分页，这是高效实现所必需的，因为在大多数 SQL 数据库中，偏移量是线性时间操作。

def queryset_iterator(queryset, page_size=1000):
    if not queryset:
        return
    max_pk = queryset.order_by("-pk")[0].pk
    # Scale the page size up by the average density of primary keys in the queryset
    adjusted_page_size = int(page_size * max_pk / queryset.count())
    
    pages = int(max_pk / adjusted_page_size) + 1
    for page_num in range(pages):
        lower = page_num * adjusted_page_size
        page = queryset.filter(pk__gte=lower, pk__lt=lower+page_size)
        for obj in page:
            yield obj

使用看起来像：

for obj in queryset_iterator(Model.objects.all()):
    # do stuff

这段代码有三个假设：

您的主键是整数（这不适用于 UUID 主键）。 查询集的主键至少在某种程度上是均匀分布的。如果不是这样，adjusted_page_size 最终可能会太大，您可能会在迭代过程中获得一个或多个大页面。

为了了解开销，我在一个包含 40,000 个条目的 Postgres 表上进行了测试。与原始迭代相比，queryset_iterator 增加了大约 80% 的迭代时间（2.2 秒对 1.2 秒）。对于 200 到 10,000 之间的页面大小，该开销并没有太大变化，尽管它开始上升到 200 以下。

【参考方案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

仅将 id 加载到内存中，并根据需要检索和丢弃对象。请注意增加的数据库负载和较慢的运行时间，这都是减少内存使用量的权衡。

我在工作程序实例上运行异步计划任务时使用过它，如果它们很慢并不重要，但如果它们尝试使用太多内存，它们可能会使实例崩溃并因此中止进程.

【讨论】：

感谢 Clay，这确实是我忘记添加的一个选项，但在我的情况下这不起作用。我的数据库大约有 3000 万行，PK 是一个 12 个字符的字符串（不是我的选择！）。这使得检索 ID 列表非常耗费内存，并且遍历每个对象会使检索变得非常缓慢！

作为参考，使用“更好的解决方案”的改进版本，它只检索在 512Mb 双核实例上运行的 30M 行所需的值，运行时间不到 30 分钟，即已经不错了。天真的解决方案会继续放慢速度，我从来没有足够的耐心等待结束（几个小时！）。正如您所说，您的解决方案会更慢，但检索 ID 列表也需要大量内存。

是的，它本身似乎对你不起作用。但是，正如您所提到的，使用仅检索您需要的内容的选项结合其他一些技术可能会有所帮助。