python - 如何从 django/sqlalchemy 处理的数据库中批量获取模型对象

Question

最近我遇到了以下问题:如何迭代真正的大数据查询以执行操作(比如为每个对象创建两个不同的对象)。如果您处理一个小查询集，这很简单:

for obj in Mymodel.objects.all():
    create_corresponding_entries(obj)

现在尝试在包含 900k 个对象的查询集中执行此操作。可能您的电脑会死机，因为它会耗尽所有内存。那么我怎样才能懒惰地实现这一目标呢？无论您使用 Django ORM 还是 SQLAlchemy，都会出现同样的问题

Answer 1

尽管 Django ORM 提供了一个“惰性”查询集，但我一直在寻找的是一个生成器，它可以为我提供一种惰性获取对象的方法。 django 中的查询集并不是真正的惰性，它们是惰性的，直到您尝试访问它们，数据库将命中并获取 1M 条目。 SQLAlchemy 做同样的事情。如果您有 oracle 或 postgre 数据库，那么您很幸运，您可以使用受支持的服务器端游标。如果您使用 mysqldb 或 pymysql 方言，SQLAlchemy 还支持这些加上 mysql。我不确定服务器端游标在幕后是如何工作的。

更多信息

• Django ORM:https://www.niwi.nz/2012/10/22/server-side-cursors-with-postgresql-and-django/
• SQLAlchemy:http://docs.sqlalchemy.org/en/latest/orm/query.html#sqlalchemy.orm.query.Query.yield_per

因此，如果您不符合上述任何一种情况，则必须想出一种方法来延迟获取这些对象。因为 Django ORM 和 SQLAlchemy 都支持通过将其转换为纯 SQL 查询来进行切片，所以我想我可以使用自定义生成器来对我需要的查询进行切片。

免责声明:该解决方案试图解决在本地转储大量数据的问题，它不会尝试最大限度地提高查询性能或与数据库相关的任何性能。

警告:与简单的 Mymodel.objects.all() 相比，这将导致对数据库的更多查询，但对您的 RAM 的挑战较小。

def lazy_bulk_fetch(max_obj, max_count, fetch_func, start=0):
    counter = start
    while counter < max_count:
        yield fetch_func()[counter:counter + max_obj]
        counter += max_obj

然后使用它例如:

fetcher = lazy_bulk_fetch(50, Mymodel.objects.count(), lambda: Mymodel.objects.order_by('id'))
for batch in fetcher:
    make_actions(batch)

这将为我每次迭代获取 50 个对象的列表，直到达到我想要的最大数量。如果您在 Django 中将 make_actions(batch) 更改为 print(batch.query)，您将看到如下内容:

SELECT "services_service"."id" FROM "services_service" LIMIT 50
SELECT "services_service"."id" FROM "services_service" LIMIT 50 OFFSET 50
SELECT "services_service"."id" FROM "services_service" LIMIT 50 OFFSET 100
SELECT "services_service"."id" FROM "services_service" LIMIT 50 OFFSET 150

---

同样的概念可以用于 slice 即 SQLAlchemy supports .在这种情况下的解决方案是相同的，但不是 python 切片，而是使用 SQLAlchemy 查询对象的 slice 函数

编辑: 据我所见，SQLAlchemy Query 类实现了 __getitem__ 函数。 因此，对于 SQLAlchemy，您可以使用我为 Django 建议的完全相同的函数。如果您想显式使用 slice 函数，您将得到如下所示的结果: p>

def lazy_bulk_fetch(max_obj, max_count, fetch_func, start=0):
    counter = start
    while counter < max_count:
        yield fetch_func().slice(counter, counter + max_obj)
        counter += max_obj

无论如何你会这样调用它:

from sqlalchemy import func
fetcher = lazy_bulk_fetch(50, session.query(func.count(Mymodel.id)), 
                          lambda: session.query(Mymodel).order_by(Mymodel.id))

这里有两个注意事项:

1. 您想使用 func.count 以便将其转换为服务器中的 COUNT 个 SQL 语句。如果您使用 len(session.query(Mymodel))，您将在本地转储所有内容，找到它的长度，然后将其丢弃

2. 我使用 lambda 这样实现就像 django 一样。我也可以有

   lazy_bulk_fetch(50, session.query(func.count(Mymodel.id)), 
                   session.query(Mymodel).order_by(Mymodel.id)) 
   

   但是我必须在我的函数中有

   yield fetch_func.slice(counter, counter + max_obj)
   

编辑 #2: 我添加了排序，否则您无法确定在第 N 次运行中不会得到相同的结果。订购保证您将获得独特的结果。最好将 id 作为排序键，否则你不能确定你错过了一个结果(因为在第 N 次命中时，可能已经添加了一个新条目并且没有 id 的排序可能会导致你错过它或得到双重条目)