使用 SQLAlchemy ORM 批量插入

Question

有没有办法让 SQLAlchemy 进行批量插入而不是插入每个单独的对象。即，

在做：

INSERT INTO `foo` (`bar`) VALUES (1), (2), (3)

而不是：

INSERT INTO `foo` (`bar`) VALUES (1)
INSERT INTO `foo` (`bar`) VALUES (2)
INSERT INTO `foo` (`bar`) VALUES (3)

我刚刚将一些代码转换为使用 sqlalchemy 而不是原始 sql，虽然现在使用起来好多了，但现在似乎更慢了（高达 10 倍），我想知道这是否是原因。

也许我可以更有效地使用会话来改善这种情况。目前我有autoCommit=False 并在我添加了一些东西后做一个session.commit()。尽管如果在其他地方更改数据库，这似乎会导致数据过时，例如即使我执行新查询，我仍然会返回旧结果？

感谢您的帮助！

Answer 1

Sqlalchemy 支持批量插入

bulk_list = [
    Foo(
        bar=1,
    ),
    Foo(
        bar=2,
    ),
    Foo(
        bar=3,
    ),
]
db.session.bulk_save_objects(bulk_list)
db.session.commit()

Answer 2

SQLAlchemy 在1.0.0 版本中引入了这一点：

Bulk operations - SQLAlchemy docs

通过这些操作，您现在可以进行批量插入或更新！

例如（如果您希望简单表插入的开销最低），您可以使用Session.bulk_insert_mappings()：

loadme = [(1, 'a'),
          (2, 'b'),
          (3, 'c')]
dicts = [dict(bar=t[0], fly=t[1]) for t in loadme]

s = Session()
s.bulk_insert_mappings(Foo, dicts)
s.commit()

或者，如果您愿意，可以跳过 loadme 元组并将字典直接写入 dicts（但我发现将所有冗长的内容从数据中剔除并在循环中加载字典列表更容易） .

Answer 3

据我所知，没有办法让 ORM 发出批量插入。我相信根本原因是 SQLAlchemy 需要跟踪每个对象的身份（即新的主键），而批量插入会干扰这一点。例如，假设您的foo 表包含id 列并映射到Foo 类：

x = Foo(bar=1)
print x.id
# None
session.add(x)
session.flush()
# BEGIN
# INSERT INTO foo (bar) VALUES(1)
# COMMIT
print x.id
# 1

由于 SQLAlchemy 在没有发出另一个查询的情况下获取了 x.id 的值，我们可以推断它直接从 INSERT 语句中获取了值。如果您不需要后续通过 same 实例访问创建的对象，则可以跳过 ORM 层进行插入：

Foo.__table__.insert().execute([{'bar': 1}, {'bar': 2}, {'bar': 3}])
# INSERT INTO foo (bar) VALUES ((1,), (2,), (3,))

SQLAlchemy 无法将这些新行与任何现有对象匹配，因此您必须重新查询它们以进行任何后续操作。

就陈旧数据而言，记住会话没有内置方法可以知道何时在会话之外更改数据库是有帮助的。为了通过现有实例访问外部修改的数据，必须将实例标记为过期。这在session.commit() 上默认发生，但可以通过调用session.expire_all() 或session.expire(instance) 手动完成。一个例子（SQL省略）：

x = Foo(bar=1)
session.add(x)
session.commit()
print x.bar
# 1
foo.update().execute(bar=42)
print x.bar
# 1
session.expire(x)
print x.bar
# 42

session.commit() 过期 x，因此第一个 print 语句隐式打开一个新事务并重新查询 x 的属性。如果您注释掉第一个打印语句，您会注意到第二个现在获取了正确的值，因为直到更新之后才会发出新查询。

从事务隔离的角度来看，这是有道理的——您应该只在事务之间进行外部修改。如果这给您带来了麻烦，我建议您澄清或重新考虑您的应用程序的事务边界，而不是立即联系session.expire_all()。

Answer 4

sqlalchemy 文档有一个writeup，关于可用于批量插入的各种技术的性能：

ORM 基本上不适用于高性能批量插入 - 这就是 SQLAlchemy 除了提供 Core 之外的全部原因 ORM 作为一流的组件。

对于快速批量插入的用例，SQL 生成和 ORM 在其上构建的执行系统是核心的一部分。 直接使用这个系统，我们可以产生一个 INSERT 是 与直接使用原始数据库 API 相比具有竞争力。

或者，SQLAlchemy ORM 提供批量操作套件 方法，提供工作单元子部分的挂钩 过程以发出核心级别的 INSERT 和 UPDATE 构造 少量基于 ORM 的自动化。

下面的示例说明了几种不同的基于时间的测试 插入行的方法，从最自动化到最不自动化。 使用 cPython 2.7，观察到的运行时：

classics-MacBook-Pro:sqlalchemy classic$ python test.py
SQLAlchemy ORM: Total time for 100000 records 12.0471920967 secs
SQLAlchemy ORM pk given: Total time for 100000 records 7.06283402443 secs
SQLAlchemy ORM bulk_save_objects(): Total time for 100000 records 0.856323003769 secs
SQLAlchemy Core: Total time for 100000 records 0.485800027847 secs
sqlite3: Total time for 100000 records 0.487842082977 sec

脚本：

import time
import sqlite3

from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy import Column, Integer, String,  create_engine
from sqlalchemy.orm import scoped_session, sessionmaker

Base = declarative_base()
DBSession = scoped_session(sessionmaker())
engine = None


class Customer(Base):
    __tablename__ = "customer"
    id = Column(Integer, primary_key=True)
    name = Column(String(255))


def init_sqlalchemy(dbname='sqlite:///sqlalchemy.db'):
    global engine
    engine = create_engine(dbname, echo=False)
    DBSession.remove()
    DBSession.configure(bind=engine, autoflush=False, expire_on_commit=False)
    Base.metadata.drop_all(engine)
    Base.metadata.create_all(engine)


def test_sqlalchemy_orm(n=100000):
    init_sqlalchemy()
    t0 = time.time()
    for i in xrange(n):
        customer = Customer()
        customer.name = 'NAME ' + str(i)
        DBSession.add(customer)
        if i % 1000 == 0:
            DBSession.flush()
    DBSession.commit()
    print(
        "SQLAlchemy ORM: Total time for " + str(n) +
        " records " + str(time.time() - t0) + " secs")


def test_sqlalchemy_orm_pk_given(n=100000):
    init_sqlalchemy()
    t0 = time.time()
    for i in xrange(n):
        customer = Customer(id=i+1, name="NAME " + str(i))
        DBSession.add(customer)
        if i % 1000 == 0:
            DBSession.flush()
    DBSession.commit()
    print(
        "SQLAlchemy ORM pk given: Total time for " + str(n) +
        " records " + str(time.time() - t0) + " secs")


def test_sqlalchemy_orm_bulk_insert(n=100000):
    init_sqlalchemy()
    t0 = time.time()
    n1 = n
    while n1 > 0:
        n1 = n1 - 10000
        DBSession.bulk_insert_mappings(
            Customer,
            [
                dict(name="NAME " + str(i))
                for i in xrange(min(10000, n1))
            ]
        )
    DBSession.commit()
    print(
        "SQLAlchemy ORM bulk_save_objects(): Total time for " + str(n) +
        " records " + str(time.time() - t0) + " secs")


def test_sqlalchemy_core(n=100000):
    init_sqlalchemy()
    t0 = time.time()
    engine.execute(
        Customer.__table__.insert(),
        [{"name": 'NAME ' + str(i)} for i in xrange(n)]
    )
    print(
        "SQLAlchemy Core: Total time for " + str(n) +
        " records " + str(time.time() - t0) + " secs")


def init_sqlite3(dbname):
    conn = sqlite3.connect(dbname)
    c = conn.cursor()
    c.execute("DROP TABLE IF EXISTS customer")
    c.execute(
        "CREATE TABLE customer (id INTEGER NOT NULL, "
        "name VARCHAR(255), PRIMARY KEY(id))")
    conn.commit()
    return conn


def test_sqlite3(n=100000, dbname='sqlite3.db'):
    conn = init_sqlite3(dbname)
    c = conn.cursor()
    t0 = time.time()
    for i in xrange(n):
        row = ('NAME ' + str(i),)
        c.execute("INSERT INTO customer (name) VALUES (?)", row)
    conn.commit()
    print(
        "sqlite3: Total time for " + str(n) +
        " records " + str(time.time() - t0) + " sec")

if __name__ == '__main__':
    test_sqlalchemy_orm(100000)
    test_sqlalchemy_orm_pk_given(100000)
    test_sqlalchemy_orm_bulk_insert(100000)
    test_sqlalchemy_core(100000)
    test_sqlite3(100000)

Answer 5

SQLAlchemy 在1.0.0 版本中引入了这一点：

Bulk operations - SQLAlchemy docs

通过这些操作，您现在可以进行批量插入或更新！

例如，你可以这样做：

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

在这里，将进行批量插入。

Answer 6

条条大路通罗马，但其中一些穿越山脉，需要渡轮，但如果您想快速到达那里，就走高速公路。

---

在这种情况下，高速公路将使用psycopg2 的execute_batch() 功能。文档说得最好：

executemany() 的当前实现（使用极其仁慈的轻描淡写）表现不佳。这些函数可用于加速针对一组参数的语句的重复执行。通过减少服务器往返次数，性能可以比使用executemany() 好几个数量级。

在我自己的测试中，execute_batch() 的速度大约是 executemany() 的两倍，并且提供了配置 page_size 以进行进一步调整的选项（如果您想压缩最后的 2-3%驱动程序的性能）。

如果您使用 SQLAlchemy，则可以通过在使用 create_engine() 实例化引擎时将 use_batch_mode=True 设置为参数来轻松启用相同的功能

Answer 7

到目前为止我找到的最佳答案是在 sqlalchemy 文档中：

http://docs.sqlalchemy.org/en/latest/faq/performance.html#i-m-inserting-400-000-rows-with-the-orm-and-it-s-really-slow

有一个完整的可能解决方案基准示例。

如文档所示：

bulk_save_objects 不是最好的解决方案，但它的性能是正确的。

就可读性而言，我认为第二好的实现是使用 SQLAlchemy Core：

def test_sqlalchemy_core(n=100000):
    init_sqlalchemy()
    t0 = time.time()
    engine.execute(
        Customer.__table__.insert(),
            [{"name": 'NAME ' + str(i)} for i in xrange(n)]
    )

此函数的上下文在文档文章中给出。

Answer 8

我通常使用add_all。

from app import session
from models import User

objects = [User(name="u1"), User(name="u2"), User(name="u3")]
session.add_all(objects)
session.commit()

Answer 9

Piere 的回答是正确的，但一个问题是 bulk_save_objects 默认情况下不会返回对象的主键，如果您对此感到担心的话。将return_defaults 设置为True 以获取此行为。

文档是here。

foos = [Foo(bar='a',), Foo(bar='b'), Foo(bar='c')]
session.bulk_save_objects(foos, return_defaults=True)
for foo in foos:
    assert foo.id is not None
session.commit()

Answer 10

这是一种方式：

values = [1, 2, 3]
Foo.__table__.insert().execute([{'bar': x} for x in values])

这将像这样插入：

INSERT INTO `foo` (`bar`) VALUES (1), (2), (3)

参考：SQLAlchemy FAQ 包括各种提交方法的基准。

Answer 11

从 0.8 版起，SQLAlchemy 添加了直接支持

根据docs，connection.execute(table.insert().values(data)) 应该可以解决问题。 （请注意，这不与connection.execute(table.insert(), data) 相同，后者通过调用executemany 导致许多单独的行插入）。除了本地连接之外，在任何情况下，性能差异都可能是巨大的。