有没有人有分析 Python/SQLAlchemy 应用程序的经验？找到瓶颈和设计缺陷的最佳方法是什么？
我们有一个 Python 应用程序，其中数据库层由 SQLAlchemy 处理。该应用程序使用批处理设计，因此许多数据库请求是按顺序在有限的时间跨度内完成的。目前运行时间有点太长，因此需要进行一些优化。我们不使用 ORM 功能，数据库是 PostgreSQL。
问问题
18879 次
5 回答
75
有时只是简单的 SQL 日志记录（通过 python 的日志记录模块或通过echo=True参数 on 启用create_engine()）可以让您了解事情需要多长时间。例如，如果您在 SQL 操作后立即记录某些内容，您会在日志中看到类似这样的内容：
17:37:48,325 INFO [sqlalchemy.engine.base.Engine.0x...048c] SELECT ...
17:37:48,326 INFO [sqlalchemy.engine.base.Engine.0x...048c] {<params>}
17:37:48,660 DEBUG [myapp.somemessage]
如果您myapp.somemessage在操作后立即登录，您知道完成 SQL 部分需要 334 毫秒。
记录 SQL 还将说明是否正在发出数十/数百个查询，这些查询可以通过连接更好地组织成更少的查询。使用 SQLAlchemy ORM 时，提供“急切加载”功能以部分 ( contains_eager()) 或完全 ( eagerload(), eagerload_all()) 自动执行此活动，但如果没有 ORM，它仅意味着使用连接，以便可以将跨多个表的结果加载到一个结果集中而不是随着更多深度的增加而增加查询数量（即r + r*r2 + r*r2*r3......）
如果日志显示单个查询花费的时间太长，您需要细分数据库处理查询、通过网络发送结果、由 DBAPI 处理以及最终由 SQLAlchemy 的结果集接收所花费的时间和/或 ORM 层。根据具体情况，每个阶段都可能出现各自的瓶颈。
为此，您需要使用分析，例如 cProfile 或 hotshot。这是我使用的装饰器：
import cProfile as profiler
import gc, pstats, time
def profile(fn):
def wrapper(*args, **kw):
elapsed, stat_loader, result = _profile("foo.txt", fn, *args, **kw)
stats = stat_loader()
stats.sort_stats('cumulative')
stats.print_stats()
# uncomment this to see who's calling what
# stats.print_callers()
return result
return wrapper
def _profile(filename, fn, *args, **kw):
load_stats = lambda: pstats.Stats(filename)
gc.collect()
began = time.time()
profiler.runctx('result = fn(*args, **kw)', globals(), locals(),
filename=filename)
ended = time.time()
return ended - began, load_stats, locals()['result']
要分析一段代码，请将其放在带有装饰器的函数中：
@profile
def go():
return Session.query(FooClass).filter(FooClass.somevalue==8).all()
myfoos = go()
分析的输出可用于给出时间花费的想法。例如，如果您看到所有时间都花在 内cursor.execute()，那是对数据库的低级 DBAPI 调用，这意味着您的查询应该通过添加索引或重组查询和/或底层架构来优化。对于该任务，我建议使用 pgadmin 及其图形 EXPLAIN 实用程序来查看查询正在执行的工作类型。
如果您看到与获取行相关的数千个调用，这可能意味着您的查询返回的行数比预期的多 - 由于连接不完整而导致的笛卡尔积可能会导致此问题。另一个问题是在类型处理中花费的时间 - 一个 SQLAlchemy 类型，例如Unicode将对绑定参数和结果列执行字符串编码/解码，这可能并非在所有情况下都需要。
配置文件的输出可能有点令人生畏，但经过一些练习后，它们很容易阅读。曾经有人在邮件列表中声称速度很慢，在让他发布配置文件的结果后，我能够证明速度问题是由于网络延迟 - 在 cursor.execute() 以及所有 Python 中花费的时间方法非常快，而大部分时间都花在了 socket.receive() 上。
如果您有雄心壮志，如果您浏览http://www.sqlalchemy.org/trac/browser/sqlalchemy/trunk/test/aaa_profiling ，在 SQLAlchemy 单元测试中还有一个更复杂的 SQLAlchemy 分析示例。在那里，我们使用装饰器进行测试，这些装饰器断言用于特定操作的方法调用的最大数量，因此如果签入了一些低效的东西，测试将揭示它（重要的是要注意，在 Python 中，函数调用具有最高的任何操作的开销，并且调用次数通常与花费的时间几乎成正比）。值得注意的是“zoomark”测试使用了一种奇特的“SQL 捕获”方案，该方案从等式中消除了 DBAPI 的开销——尽管该技术不是
于 2009-07-24T03:54:46.533 回答
51
SQLAlchemy wiki上有一个非常有用的分析方法
通过几个小的修改，
from sqlalchemy import event
from sqlalchemy.engine import Engine
import time
import logging
logging.basicConfig()
logger = logging.getLogger("myapp.sqltime")
logger.setLevel(logging.DEBUG)
@event.listens_for(Engine, "before_cursor_execute")
def before_cursor_execute(conn, cursor, statement,
parameters, context, executemany):
context._query_start_time = time.time()
logger.debug("Start Query:\n%s" % statement)
# Modification for StackOverflow answer:
# Show parameters, which might be too verbose, depending on usage..
logger.debug("Parameters:\n%r" % (parameters,))
@event.listens_for(Engine, "after_cursor_execute")
def after_cursor_execute(conn, cursor, statement,
parameters, context, executemany):
total = time.time() - context._query_start_time
logger.debug("Query Complete!")
# Modification for StackOverflow: times in milliseconds
logger.debug("Total Time: %.02fms" % (total*1000))
if __name__ == '__main__':
from sqlalchemy import *
engine = create_engine('sqlite://')
m1 = MetaData(engine)
t1 = Table("sometable", m1,
Column("id", Integer, primary_key=True),
Column("data", String(255), nullable=False),
)
conn = engine.connect()
m1.create_all(conn)
conn.execute(
t1.insert(),
[{"data":"entry %d" % x} for x in xrange(100000)]
)
conn.execute(
t1.select().where(t1.c.data.between("entry 25", "entry 7800")).order_by(desc(t1.c.data))
)
输出类似于：
DEBUG:myapp.sqltime:Start Query:
SELECT sometable.id, sometable.data
FROM sometable
WHERE sometable.data BETWEEN ? AND ? ORDER BY sometable.data DESC
DEBUG:myapp.sqltime:Parameters:
('entry 25', 'entry 7800')
DEBUG:myapp.sqltime:Query Complete!
DEBUG:myapp.sqltime:Total Time: 410.46ms
然后，如果您发现一个奇怪的慢查询，您可以获取查询字符串，在参数中格式化（可以使用%字符串格式化运算符，至少对于 psycopg2），在其前面加上“EXPLAIN ANALYZE”并将查询计划输出推入http://explain.depesz.com/（通过这篇关于 PostgreSQL 性能的好文章找到）
于 2011-12-08T09:06:54.550 回答
3
我在使用 cprofile 并在运行中查看结果方面取得了一些成功。这至少告诉我哪些函数和调用需要很长时间以及数据库是否是问题所在。文档在这里。你需要 wxpython。关于它的演示文稿很好地帮助您入门。
它就像
import cProfile
command = """foo.run()"""
cProfile.runctx( command, globals(), locals(), filename="output.profile" )
然后
python runsnake.py 输出.profile
如果您希望优化您的查询，您将需要postgrsql 分析。
登录记录查询也是值得的，但是我知道没有解析器来获取长时间运行的查询（它对并发请求没有用处）。
sqlhandler = logging.FileHandler("sql.log")
sqllogger = logging.getLogger('sqlalchemy.engine')
sqllogger.setLevel(logging.info)
sqllogger.addHandler(sqlhandler)
并确保您的创建引擎语句具有 echo = True。
当我这样做时，实际上主要问题是我的代码，所以 cprofile 的事情有所帮助。
于 2009-07-23T12:00:32.223 回答
2
我刚刚发现了这个库sqltap（https://github.com/inconshreveable/sqltap）。它生成样式精美的 HTML 页面，有助于检查和分析 SQLAlchemy 生成的 SQL 查询。
一个示例用法：
profiler = sqltap.start()
run_some_queries()
statistics = profiler.collect()
sqltap.report(statistics, "report.html")
该库已经 2 年没有更新了，但是，当我今天早些时候用我的应用程序测试它时，它似乎工作得很好。
于 2018-10-19T20:25:33.377 回答
0
如果只是您想要分析的查询时间，您可以使用上下文管理器记录在特定上下文中执行的所有查询：
"""SQLAlchemy Query profiler and logger."""
import logging
import time
import traceback
import sqlalchemy
class QueryProfiler:
"""Log query duration and SQL as a context manager."""
def __init__(self,
engine: sqlalchemy.engine.Engine,
logger: logging.Logger,
path: str):
"""
Initialize for an engine and logger and filepath.
engine: The sqlalchemy engine for which events should be logged.
You can pass the class `sqlalchemy.engine.Engine` to capture all engines
logger: The logger that should capture the query
path: Only log the stacktrace for files in this path, use `'/'` to log all files
"""
self.engine = engine
self.logger = logger
self.path = path
def _before_cursor_execute(self, conn, cursor, statement, parameters, context, executemany):
"""Set the time on the connection to measure query duration."""
conn._sqla_query_start_time = time.time()
def _after_cursor_execute(self, conn, cursor, statement, parameters, context, executemany):
"""Listen for the 'after_cursor_execute' event and log sqlstatement and time."""
end_time = time.time()
start_time = getattr(conn, '_sqla_query_start_time', end_time)
elapsed_time = round((end_time-start_time) * 1000)
# only include the files in self.path in the stacktrace to reduce the noise
stack = [frame for frame in traceback.extract_stack()[:-1] if frame.filename.startswith(self.path)]
self.logger.debug('Query `%s` took %s ms. Stack: %s', statement, elapsed_time, traceback.format_list(stack))
def __enter__(self, *args, **kwargs):
"""Context manager."""
if isinstance(self.engine, sqlalchemy.engine.Engine):
sqlalchemy.event.listen(self.engine, "before_cursor_execute", self._before_cursor_execute)
sqlalchemy.event.listen(self.engine, "after_cursor_execute", self._after_cursor_execute)
return self
def __exit__(self, *args, **kwargs) -> None:
"""Context manager."""
if isinstance(self.engine, sqlalchemy.engine.Engine):
sqlalchemy.event.remove(self.engine, "before_cursor_execute", self._before_cursor_execute)
sqlalchemy.event.remove(self.engine, "after_cursor_execute", self._after_cursor_execute)
使用与测试：
"""Test SQLAlchemy Query profiler and logger."""
import logging
import os
import sqlalchemy
from .sqlaprofiler import QueryProfiler
def test_sqlite_query(caplog):
"""Create logger and sqllite engine and profile the queries."""
logging.basicConfig()
logger = logging.getLogger(f'{__name__}')
logger.setLevel(logging.DEBUG)
caplog.set_level(logging.DEBUG, logger=f'{__name__}')
path = os.path.dirname(os.path.realpath(__file__))
engine = sqlalchemy.create_engine('sqlite://')
metadata = sqlalchemy.MetaData(engine)
table1 = sqlalchemy.Table(
"sometable", metadata,
sqlalchemy.Column("id", sqlalchemy.Integer, primary_key=True),
sqlalchemy.Column("data", sqlalchemy.String(255), nullable=False),
)
conn = engine.connect()
metadata.create_all(conn)
with QueryProfiler(engine, logger, path):
conn.execute(
table1.insert(),
[{"data": f"entry {i}"} for i in range(100000)]
)
conn.execute(
table1.select()
.where(table1.c.data.between("entry 25", "entry 7800"))
.order_by(sqlalchemy.desc(table1.c.data))
)
assert caplog.messages[0].startswith('Query `INSERT INTO sometable (data) VALUES (?)` took')
assert caplog.messages[1].startswith('Query `SELECT sometable.id, sometable.data \n'
'FROM sometable \n'
'WHERE sometable.data BETWEEN ? AND ? '
'ORDER BY sometable.data DESC` took ')
于 2021-04-28T10:06:51.563 回答