为SQLAlchemy批量upsert寻找更好的策略

Question

我有一个带有RESTful API 的Flask应用程序.其中一个API调用是带有JSON有效负载的"大量upsert"调用.我在努力表现.

我尝试的第一件事就是merge-result在一个Query物体上使用,因为......

这是一个优化的方法,它将合并所有映射的实例,保留结果行和未映射列的结构,其方法开销少于为每个值显式调用Session.merge()的方法开销.

这是最初的代码:

class AdminApiUpdateTasks(Resource):

    """Bulk task creation / update endpoint"""

    def put(self, slug):
        taskdata = json.loads(request.data)
        existing = db.session.query(Task).filter_by(challenge_slug=slug)
        existing.merge_result(
            [task_from_json(slug, **task) for task in taskdata])
        db.session.commit()
        return {}, 200

对该端点的请求包含〜5000条记录,所有记录都已存在于数据库中,返回的时间超过11米:

real    11m36.459s
user    0m3.660s
sys 0m0.391s

由于这是一个相当典型的用例,我开始研究提高性能的替代方案.根据我更好的判断,我尝试merge了每个记录的会话:

class AdminApiUpdateTasks(Resource):

    """Bulk task creation / update endpoint"""

    def put(self, slug):
        # Get the posted data
        taskdata = json.loads(request.data)
        for task in taskdata:
           db.session.merge(task_from_json(slug, **task))
        db.session.commit()
        return {}, 200

令我惊讶的是,这个速度是原来的两倍多:

real    4m33.945s
user    0m3.608s
sys 0m0.258s

我有两个问题:

1. 为什么第二种策略的使用merge速度比使用的优化第一种策略更快merge_result？
2. 如果有的话,我应该采取哪些其他策略来进一步优化？

Answer 1

我认为这要么导致您在第一个查询中运行缓慢：

existing = db.session.query(Task).filter_by(challenge_slug=slug)

另外你应该改变这个：

    existing.merge_result(
        [task_from_json(slug, **task) for task in taskdata])

到：

    existing.merge_result(
        (task_from_json(slug, **task) for task in taskdata))

因为这应该节省您一些内存和时间，因为在将列表发送到 merge_result 方法之前不会在内存中生成列表。