为什么迭代大型 Django QuerySet 会消耗大量内存？

Question

该表包含大约一千万行。

for event in Event.objects.all():
    print event

这会导致内存使用量稳定增加到 4 GB 左右，此时行会快速打印。第一行打印之前的漫长延迟让我感到惊讶——我预计它几乎会立即打印。

我还尝试了 Event.objects.iterator() ，其行为方式相同。

我不明白 Django 正在将什么加载到内存中，也不明白它为什么要这样做。我期望 Django 在数据库级别迭代结果，这意味着结果将以大致恒定的速率打印（而不是在漫长的等待后一次全部打印）。

我误解了什么？

（我不知道这是否相关，但我正在使用 PostgreSQL。）

Answer 1

Nate C 很接近，但还不够。

来自文档：

您可以通过以下方式评估查询集：

• 迭代。 QuerySet 是可迭代的，并且它会在您第一次迭代它时执行其数据库查询。例如，这将打印数据库中所有条目的标题：

  对于 Entry.objects.all() 中的 e：
      打印电子标题
  

因此，当您第一次进入该循环并获取查询集的迭代形式时，会立即检索一千万行。您所经历的等待是 Django 加载数据库行并为每一行创建对象，然后返回您可以实际迭代的内容。然后你就将所有内容都存储在内存中，结果就会溢出。

根据我对文档的阅读，iterator() 只是绕过 QuerySet 的内部缓存机制。我认为做一件一件的事情可能是有意义的，但这反过来需要对数据库进行一千万次单独的点击。也许并不是那么理想。

有效地迭代大型数据集是我们还没有完全正确的事情，但是您可能会发现一些片段对您的目的有用：

• 内存高效的 Django QuerySet 迭代器
• 批量查询集
• QuerySet Foreach

Answer 2

可能不是更快或最有效，但作为一个现成的解决方案，为什么不使用此处记录的 django 核心的 Paginator 和 Page 对象：

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

像这样的：

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

paginator = Paginator(model.objects.all(), 1000) # chunks of 1000, you can 
                                                 # change this to desired chunk size

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

Answer 3

Django 的默认行为是在评估查询时缓存 QuerySet 的整个结果。您可以使用 QuerySet 的迭代器方法来避免这种缓存：

for event in Event.objects.all().iterator():
    print event

https:// docs.djangoproject.com/en/stable/ref/models/querysets/#iterator

iterator() 方法评估查询集，然后直接读取结果，而不在查询集级别进行缓存。当迭代大量只需要访问一次的对象时，此方法会带来更好的性能并显着减少内存。请注意，缓存仍然是在数据库级别完成的。

使用 iterator() 减少了我的内存使用量，但它仍然高于我的预期。使用 mpaf 建议的分页器方法使用的内存要少得多，但对于我的测试用例来说速度慢 2-3 倍。

from django.core.paginator import Paginator

def chunked_iterator(queryset, chunk_size=10000):
    paginator = Paginator(queryset, chunk_size)
    for page in range(1, paginator.num_pages + 1):
        for obj in paginator.page(page).object_list:
            yield obj

for event in chunked_iterator(Event.objects.all()):
    print event

Answer 4

对于大量记录，数据库游标性能更好。你确实需要 Django 中的原始 SQL，Django 游标与 SQL 游标不同。

Nate C 建议的 LIMIT - OFFSET 方法可能足以满足您的情况。对于大量数据，它比游标慢，因为它必须一遍又一遍地运行相同的查询，并且必须跳过越来越多的结果。

Answer 5

Django 没有很好的解决方案来从数据库中获取大项目。

import gc
# Get the events in reverse order
eids = Event.objects.order_by("-id").values_list("id", flat=True)

for index, eid in enumerate(eids):
    event = Event.object.get(id=eid)
    # do necessary work with event
    if index % 100 == 0:
       gc.collect()
       print("completed 100 items")

values_list 可以用来获取数据库中所有的id，然后获取每个对象单独。随着时间的推移，大对象将在内存中创建，并且在退出 for 循环之前不会被垃圾收集。上面的代码在每消耗第 100 个项目后进行手动垃圾回收。

Answer 6

这里有很多过时的结果。不确定何时添加，但 Django 的 QuerySet.iterator() 方法 使用具有块大小的服务器端游标，从数据库传输结果。因此，如果您使用的是 postgres，现在应该可以为您处理开箱即用的问题。

Answer 7

这是来自文档：
http://docs.djangoproject.com/en/dev/ref/models/查询集/

在您执行某些操作来评估查询集之前，不会真正发生数据库活动。

因此，当运行 print 事件 时，查询会触发（根据您的命令，这是一次全表扫描。）并加载结果。您要求所有对象，并且如果不获得所有对象就无法获得第一个对象。

但如果你这样做：

Event.objects.all()[300:900]

http://docs.djangoproject .com/en/dev/topics/db/queries/#limiting-querysets

然后它会在内部为sql添加偏移量和限制。

Answer 8

在迭代查询集之前会消耗大量内存，因为整个查询的所有数据库行都会立即处理成对象，并且根据行数可能会进行大量处理。

您可以将查询集分成更小的易消化的位。我将这种模式称为“勺子喂养”。这是我在管理命令中使用的带有进度条的实现，首先 pip3 install tqdm

from tqdm import tqdm


def spoonfeed(qs, func, chunk=1000, start=0):
    """
    Chunk up a large queryset and run func on each item.

    Works with automatic primary key fields.

    chunk -- how many objects to take on at once
    start -- PK to start from

    >>> spoonfeed(Spam.objects.all(), nom_nom)
    """
    end = qs.order_by('pk').last()
    progressbar = tqdm(total=qs.count())
    if not end:
        return
    while start < end.pk:
        for o in qs.filter(pk__gt=start, pk__lte=start+chunk):
            func(o)
            progressbar.update(1)
        start += chunk
    progressbar.close()

要使用它，您需要编写一个对对象执行操作的函数：

def set_population(town):
    town.population = calculate_population(...)
    town.save()

然后在查询集上运行该函数：

spoonfeed(Town.objects.all(), set_population)

Answer 9

这是一个包括 len 和 count 的解决方案：

class GeneratorWithLen(object):
    """
    Generator that includes len and count for given queryset
    """
    def __init__(self, generator, length):
        self.generator = generator
        self.length = length

    def __len__(self):
        return self.length

    def __iter__(self):
        return self.generator

    def __getitem__(self, item):
        return self.generator.__getitem__(item)

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

    def count(self):
        return self.__len__()

def batch(queryset, batch_size=1024):
    """
    returns a generator that does not cache results on the QuerySet
    Aimed to use with expected HUGE/ENORMOUS data sets, no caching, no memory used more than batch_size

    :param batch_size: Size for the maximum chunk of data in memory
    :return: generator
    """
    total = queryset.count()

    def batch_qs(_qs, _batch_size=batch_size):
        """
        Returns a (start, end, total, queryset) tuple for each batch in the given
        queryset.
        """
        for start in range(0, total, _batch_size):
            end = min(start + _batch_size, total)
            yield (start, end, total, _qs[start:end])

    def generate_items():
        queryset.order_by()  # Clearing... ordering by id if PK autoincremental
        for start, end, total, qs in batch_qs(queryset):
            for item in qs:
                yield item

    return GeneratorWithLen(generate_items(), total)

用法：

events = batch(Event.objects.all())
len(events) == events.count()
for event in events:
    # Do something with the Event

Answer 10

我通常使用原始 MySQL 原始查询而不是 Django ORM 来完成此类任务。

MySQL支持流模式，因此我们可以安全快速地循环所有记录，而不会出现内存不足错误。

import MySQLdb
db_config = {}  # config your db here
connection = MySQLdb.connect(
        host=db_config['HOST'], user=db_config['USER'],
        port=int(db_config['PORT']), passwd=db_config['PASSWORD'], db=db_config['NAME'])
cursor = MySQLdb.cursors.SSCursor(connection)  # SSCursor for streaming mode
cursor.execute("SELECT * FROM event")
while True:
    record = cursor.fetchone()
    if record is None:
        break
    # Do something with record here

cursor.close()
connection.close()

参考：

1. 从 MySQL 检索百万行
2. MySQL 结果集流式传输与一次获取整个 JDBC 结果集的性能如何