Как я могу профилировать приложение на основе SQLAlchemy?
У кого-нибудь есть опыт профилирования приложения Python/SQLAlchemy? И как лучше найти узкие места и недостатки в дизайне?
У нас есть приложение Python, где уровень базы данных обрабатывается SQLAlchemy. Приложение использует пакетный дизайн, поэтому многие запросы к базе данных выполняются последовательно и за ограниченный промежуток времени. В настоящее время это занимает слишком много времени для запуска, поэтому необходима некоторая оптимизация. Мы не используем функциональность ORM, а база данных - PostgreSQL.
5 ответов
Иногда просто ведение журнала SQL (включается через модуль регистрации Python или через echo=True спор о 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().
Если вы чувствуете себя амбициозно, есть и более сложный пример профилирования SQLAlchemy в модульных тестах SQLAlchemy, если вы покопаетесь в http://www.sqlalchemy.org/trac/browser/sqlalchemy/trunk/test/aaa_profiling. Там у нас есть тесты, использующие декораторы, которые утверждают максимальное количество вызовов методов, используемых для определенных операций, так что, если что-то неэффективное будет проверено, тесты обнаружат это (важно отметить, что в Python вызовы функций имеют самое высокое накладные расходы на любую операцию, и количество вызовов чаще всего почти не пропорционально затраченному времени). Следует отметить тесты "zoomark", в которых используется причудливая схема "захвата SQL", которая исключает накладные расходы из DBAPI из уравнения - хотя этот метод на самом деле не является необходимым для профилирования садового сорта.
В вики SQLAlchemy есть чрезвычайно полезный рецепт профилирования
С парой мелких модификаций,
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 Stackru 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 Stackru: 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)
Я только что обнаружил библиотеку sqltap ( https://github.com/inconshreveable/sqltap). Он создает красиво стилизованные HTML-страницы, которые помогают с проверкой и профилированием запросов SQL, генерируемых SQLAlchemy.
Пример использования:
profiler = sqltap.start()
run_some_queries()
statistics = profiler.collect()
sqltap.report(statistics, "report.html")
Библиотека не обновлялась в течение 2 лет, однако, казалось, она работала очень хорошо, когда я тестировал ее с моим приложением ранее сегодня.
У меня был некоторый успех в использовании cprofile и просмотре результатов в runnakerun. Это, по крайней мере, сказало мне, какие функции и вызовы выполнялись долго и была ли проблема с базой данных. Документация здесь. Вам нужен wxpython. Презентация о нем хороша для начала.
Это так же просто, как
import cProfile
command = """foo.run()"""
cProfile.runctx( command, globals(), locals(), filename="output.profile" )
затем
python runnake.py output.profile
Если вы хотите оптимизировать свои запросы, вам понадобится профилирование postgrsql.
Стоит также войти в систему для записи запросов, но для этого нет парсера, о котором я знаю, чтобы получить длительные запросы (и он не будет полезен для одновременных запросов).
sqlhandler = logging.FileHandler("sql.log")
sqllogger = logging.getLogger('sqlalchemy.engine')
sqllogger.setLevel(logging.info)
sqllogger.addHandler(sqlhandler)
и убедитесь, что ваш оператор создания движка имеет echo = True.
Когда я это сделал, на самом деле основной проблемой была моя программа, поэтому мне помог cprofile.
Если вы хотите профилировать только время запросов, вы можете использовать диспетчер контекста для регистрации всех запросов, выполняемых в определенном контексте:
"""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 ')