python - Flask 应用程序内存泄漏与 sql 炼金术

Question

我正在通过流式处理大量 http 响应内容(100 万条记录)。从请求中读取 10000 个元素后，我调用一个方法将它们插入数据库，然后继续处理响应。我注意到 RAM 使用量在不断增加。下面是示例代码。如果我评论 db.session.add() 部分试图将数据添加到临时列表，则不会发生内存泄漏。那么 SQLAlchemy 在插入数据库后没有正确释放内存？请提出建议。

示例代码。

for chunk in response.iter_content(2048):
    #decode chunk
    xyz_json_list = list()
    xyz_json = json.loads(chunk_string)
    xyz_json_list.append(xyz_json)
    if len(xyz_json_list) == 10000:
        add_elements_to_db(xyz_json_list)
        xyz_json_list = list()
    ...



func add_elements_to_db(xyz_list):
    for xyz in xyz_list:
       db.sesssion.add(create_object_from_json(xyz))
    db.session.commit()
    db.session.expunge_all()
    del xyz_list
    gc.collect()

每次调用 add_elements_to_db 后，内存呈指数增长。

Answer 1

从 v1.0.0 开始，SQLAlchemy 支持 Bulk Opertions

• Session.bulk_save_objects()
• Session.bulk_insert_mappings()
• Session.bulk_update_mappings()

s = Session()
objects = [
    User(name="u1"),
    User(name="u2"),
    User(name="u3")
]
s.bulk_save_objects(objects)

因此您可以在 add_elements_to_db() 函数中做一些小改动:

[..]

def add_elements_to_db(xyz_list):

    objects = []
   
    for xyz in xyz_list:
        objects.append(create_object_from_json(xyz))

    db.session.bulk_save_objects(objects)

[..]

只是好奇，为什么你不使用 Redis 来执行异步任务，你可以将 add_elements_to_db() 函数的执行作为异步任务委托(delegate)给 Redis 因为它是一个阻塞任务，而且速度特别慢，你肯定会看到良好的性能。