SQLAlchemy批量插入性能的分析

灰太狼 2022-05-31 01:08 339阅读 0赞

背景：

项目中有一个数据同步的步骤（使用SQLAlchemy通过txt文件把数据放进mysql数据），在当中遇到了一些问题：

1，在同步批量数据的时候（主要是更新，和插入），会导致mysql等待超时。从而到时数据导入失败。

2，在批量导入数据的时候，使用session.add(obj)的方法，导入时间较长。

由于出现上述的问题，所以特别看了关于SQLAlchemy的插入数据库的文档。

SQLAlchemy批量导入有哪些？性能又如何？

SQLAlchemy官网有一篇文章详细介绍到：[Performance—SQLAlchemy][Performance_SQLAlchemy]

当然，也有技术达人翻译了中文的文档：[SQLAlchemy批量插入性能比较][SQLAlchemy]

在这里我也贴一下官网的代码，及其结果。然后做一些我自己的认识和简单说明

官网代码：

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_save_objects(n=100000):
        init_sqlalchemy()
        t0 = time.time()
        for chunk in range(0, n, 10000):
            DBSession.bulk_save_objects(
                [
                    Customer(name="NAME " + str(i))
                    for i in xrange(chunk, min(chunk + 10000, n))
                ]
            )
        DBSession.commit()
        print(
            "SQLAlchemy ORM bulk_save_objects(): Total time for " + str(n) +
            " records " + str(time.time() - t0) + " secs")
    
    
    def test_sqlalchemy_orm_bulk_insert(n=100000):
        init_sqlalchemy()
        t0 = time.time()
        for chunk in range(0, n, 10000):
            DBSession.bulk_insert_mappings(
                Customer,
                [
                    dict(name="NAME " + str(i))
                    for i in xrange(chunk, min(chunk + 10000, n))
                ]
            )
        DBSession.commit()
        print(
            "SQLAlchemy ORM bulk_insert_mappings(): 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_save_objects(100000)
        test_sqlalchemy_orm_bulk_insert(100000)
        test_sqlalchemy_core(100000)
        test_sqlite3(100000)

官网运行结果：

SQLAlchemy ORM: Total time for 100000 records 6.89754080772 secs
    SQLAlchemy ORM pk given: Total time for 100000 records 4.09481811523 secs
    SQLAlchemy ORM bulk_save_objects(): Total time for 100000 records 1.65821218491 secs
    SQLAlchemy ORM bulk_insert_mappings(): Total time for 100000 records 0.466513156891 secs
    SQLAlchemy Core: Total time for 100000 records 0.21024107933 secs
    sqlite3: Total time for 100000 records 0.137335062027 sec

简单说明：同样添加10W行数据，插入时间比

1，Session.add(obj) 使用时间：6.89754080772秒

2，Session.add(obj)注意：手动添加了主键id 使用时间：4.09481811523秒

3，bulk\_save\_objects(\[obj1, obj2\])  使用时间：1.65821218491秒

4，bulk\_inser\_mappings(DBmodel, \[dict1, dict2\]) 使用时间： 0.466513156781秒

5，SQLAlchemy\_core(DBmodel.\_\_table\_\_insert(), \[dict1, dict2\]) 使用时间：0.21024107933秒

6，直接执行execute(str\_sql\_insert) 直接执行sql插入语句 使用时间：0.137335062027秒

个人见解：

由上面的运行时间，我们知道直接执行sql语句是最快的，执行Session.add(obj)这样传统的方式属于比较慢。

这样是不是说明了，我们在项目中直接执行sql语句就是最优的？答案是：否认的

1，我们知道SQLAlchemy的出现就是想让我们更加方便快捷的操作数据库，而不是我们直接写sql语句（如果整个项目都直接用sql语句操作数据库的话，万一项目有漏洞让居心不良的人运行了delete table，那真的是GG了）,。

2，还有一个问题，当表新加了一个字段的时候，你需要把整个项目中的insert语句，添加该添加的字段。（bulk\_inser\_mappings, SQLAlchemy\_core中都可能存在这么一个问题）

3，很多同学都是习惯使用传统的Session.add(obj)的方式，我们需要在这基础上进行优化，避免在项目上有太多的改动。所以使用的是bulk\_save\_objects(\[obj1, obj2\])这跟Session.add(obj)的区别就是，一次性添加，跟一个个添加的区别。

值得注意：

DBSession.configure(bind=engine)

时：autoflush默认是True，expire\_on\_commit也是True，这两个参数会影响你使用中是否自动把数据刷进数据或获得数据  
应该把链接方式设置为：

DBSession.configure(bind=engine, autoflush=False, expire_on_commit=False)

具体可以看官网文档：[Session API][]

话说，在项目中没有绝对可以用什么，不可以用什么。只有应该用什么，在一个项目中适应的使用不同的数据库语句会大大提高性能。这个需要建立在自己对应用场景，语句用法充分的理解。

[Performance_SQLAlchemy]: http://docs.sqlalchemy.org/en/latest/faq/performance.html
[SQLAlchemy]: http://blog.csdn.net/five3/article/details/70786741
[Session API]: http://docs.sqlalchemy.org/en/latest/orm/session_api.html