SQLAlchemy Core查询SQLite性能逊于原生sqlite3的优化求助

问题背景 #

从SQLite读取500万条整数数据时，原生sqlite3实现耗时约1.2秒，SQLAlchemy Core实现却耗时约3.2秒。经性能分析确认，SQLite内部查询耗时约1秒（与原生实现一致），性能瓶颈集中在SQLAlchemy将原生行转换为Row对象的逻辑上。手动修改内部make_row=None后，耗时降至2.2秒，但需找到无需修改源码的配置方式。

使用环境：SQLAlchemy 2.0.20、libsqlite 3.37.2、Python 3.10.4

执行的查询语句：

SELECT data.x
FROM data

测试代码：

N = 5_000_000
# 存储500万条整数的数据库
db_path = '/tmp/ints.sqlite'

# 耗时约1.2秒
def test_sqlite():
    import sqlite3
    total = 0
    last = None
    with sqlite3.connect(db_path) as conn:
        for (x,) in conn.execute('SELECT * FROM data'):
            total += 1
            last = x
    assert total == N  # 验证数据完整性
    print(last)


# 耗时约3.2秒
def test_sqlalchemy():
    import sqlalchemy
    from sqlalchemy import Table, MetaData, Column, text
    engine = sqlalchemy.create_engine(f'sqlite:///{db_path}', echo=True)
    meta = MetaData()
    table_cache = Table('data', meta, Column('x', sqlalchemy.Integer))
    query = table_cache.select()
    total = 0
    last = None
    with engine.connect() as conn:
        rows = conn.execute(query)
        for (x,) in rows:
            total += 1
            last = x
    assert total == N  # 验证数据完整性
    print(last)

解决方案 #

要让SQLAlchemy返回原生sqlite元组而非Row对象，可通过以下两种无需修改源码的方式实现：

方法1：通过execution_options指定自定义行处理器 #

在执行查询时，传入自定义行处理器直接返回原生元组，绕过Row对象构造逻辑：

def test_sqlalchemy_fast():
    import sqlalchemy
    from sqlalchemy import Table, MetaData, Column
    engine = sqlalchemy.create_engine(f'sqlite:///{db_path}')
    meta = MetaData()
    table_cache = Table('data', meta, Column('x', sqlalchemy.Integer))
    query = table_cache.select()
    
    total = 0
    last = None
    with engine.connect() as conn:
        # 配置行处理器直接返回原生元组
        result = conn.execute(query, execution_options={"row_processor": lambda context, row: row})
        for (x,) in result:
            total += 1
            last = x
    assert total == N
    print(last)

该方法耗时约2.2秒，与手动修改make_row=None的效果一致。

方法2：使用text查询并开启native_results #

若使用text构造查询，可开启native_results选项直接返回原生结果：

def test_sqlalchemy_text_fast():
    import sqlalchemy
    from sqlalchemy import text
    engine = sqlalchemy.create_engine(f'sqlite:///{db_path}')
    
    total = 0
    last = None
    with engine.connect() as conn:
        # 使用text查询并开启native_results
        result = conn.execute(text("SELECT x FROM data").execution_options(native_results=True))
        for (x,) in result:
            total += 1
            last = x
    assert total == N
    print(last)

此方式同样能跳过Row对象转换，性能接近方法1。

额外优化建议 #

• 关闭echo=True：测试代码中开启的SQL语句输出会增加额外开销，生产环境建议关闭。
• 批量读取：业务允许的情况下，使用fetchmany(size=1000)减少循环次数，进一步提升性能。

内容的提问来源于stack exchange，提问作者karlicoss