SQLAlchemy查询PostgreSQL数据远慢于PgAdmin的原因及优化方案

我正在使用SQLAlchemy和PostgreSQL存储股票数据，数据表结构如下：

from sqlalchemy import DateTime, Column, Integer, String, Float
from base_sql import Base

class StockData(Base):
    __tablename__ = "StocksData"

    id = Column(Integer, primary_key=True)
    date = Column(DateTime(timezone=False))
    ticker = Column(String(90))
    interval = Column(String(90))
    open = Column(Float())
    high = Column(Float())
    low = Column(Float())
    close = Column(Float())
    volume = Column(Float())

def __int__(self, id, date,ticker, interval, open, high,
            low, close, volume):
    self.id = id
    self.date = date
    self.ticker = ticker
    self.interval = interval
    self.open = open
    self.high = high
    self.low = low
    self.close = close
    self.volume = volume

数据库中已导入约400万条数据，使用以下SQLAlchemy代码查询所有数据：

from sqlalchemy import create_engine
from sqlalchemy.orm import Session
from price_data_sql import StockData
import os
from dotenv import load_dotenv
load_dotenv()

import time

from base_sql import engine, Base, Session

engine = create_engine(os.getenv("POSTGRES_CONNECTION_STRING"))
session = Session()

st = time.time()
print("Querying database...")
query = session.query(StockData).all()
print(f"Query took {time.time() - st} seconds")
session.close()

print(len(query))

代码输出：

Querying database...
Query took 58.79122018814087 seconds
4013519

但在PgAdmin中执行「查看/编辑 > 所有行」仅需约3.5秒，请问造成这种性能差异的可能原因是什么？有哪些可以提升SQLAlchemy数据查询速度的建议？

• ORM对象实例化开销：SQLAlchemy的query(StockData).all()会将每一行数据转换为StockData类的实例，涉及大量对象初始化、属性赋值操作，400万条数据的实例化开销会非常显著。而PgAdmin仅获取原始查询结果，无需做对象转换，因此速度更快。
• 数据加载方式差异：PgAdmin采用流式/分批读取的方式，不会一次性把所有数据加载到内存；而all()方法会将所有查询结果一次性拉取到本地并转换为对象，内存占用和处理时间大幅增加。
• ORM额外逻辑损耗：SQLAlchemy ORM会处理会话管理、对象状态跟踪（如脏数据检测）等额外逻辑，这些都会带来性能开销；PgAdmin直接执行原生SQL，没有此类额外负担。

• 使用SQLAlchemy Core或原生SQL：跳过ORM的对象转换步骤，直接用核心API或原生SQL获取原始数据，示例：

  from sqlalchemy import text
  
  with engine.connect() as conn:
      result = conn.execute(text("SELECT * FROM StocksData"))
      # 可选择分批获取，避免一次性加载全部数据
      for batch in result.partitions(1000):
          process_batch(batch)
  

• 分批读取数据：避免用all()一次性加载所有数据，改用yield_per()分批获取，降低内存压力：

  query = session.query(StockData).yield_per(1000)
  for row in query:
      # 处理单条数据
      pass
  

• 关闭不必要的ORM特性：如果不需要修改查询结果并回写数据库，关闭对象状态跟踪和属性加载，减少开销：

  query = session.query(StockData).options(noload('*'))
  

• 启用服务器端游标：创建引擎时开启服务器端游标，让PostgreSQL在服务器端分批处理结果，避免一次性传输全部数据：

  engine = create_engine(os.getenv("POSTGRES_CONNECTION_STRING"), server_side_cursors=True)
  

  搭配yield_per()使用效果更佳。
• 只查询所需字段：明确指定需要的列，减少数据传输量和对象初始化开销：

  query = session.query(StockData.date, StockData.ticker, StockData.close).all()
  

• 优化数据库连接配置：调整引擎的连接池参数（如pool_size、max_overflow），确保连接充足且无额外连接开销；同时可优化数据传输相关参数（如client_encoding）提升效率。

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