根据SQLAlchemy,在for循环中,将select语句视为可迭代。结果是将返回大量行的select语句不会使用过多的内存。
我在MySQL表上发现以下语句:
for row in my_connections.execute(MyTable.__table__.select()): yield row
似乎不遵循此要求,因为我溢出了可用内存并在产生第一行之前开始崩溃。我究竟做错了什么?
基本MySQLdb游标立即从服务器获取整个查询结果。这会消耗大量内存和时间。当您要进行大型查询并一次从服务器提取结果时,请使用MySQLdb.cursors.SSCursor。
MySQLdb
因此,connect_args={'cursorclass': MySQLdb.cursors.SSCursor} 在创建时尝试传递engine:
connect_args={'cursorclass': MySQLdb.cursors.SSCursor}
engine
from sqlalchemy import create_engine, MetaData import MySQLdb.cursors engine = create_engine('mysql://root:zenoss@localhost/e2', connect_args={'cursorclass': MySQLdb.cursors.SSCursor}) meta = MetaData(engine, reflect=True) conn = engine.connect() rs = s.execution_options(stream_results=True).execute()
参见http://www.sqlalchemy.org/trac/ticket/1089
请注意,使用SSCursor会锁定表,直到提取完成。这会影响使用同一连接的其他游标:来自同一连接的两个游标不能同时从表中读取。
但是,来自不同连接的游标可以同时从同一表中读取。
这是一些演示问题的代码:
import MySQLdb import MySQLdb.cursors as cursors import threading import logging import config logger = logging.getLogger(__name__) query = 'SELECT * FROM huge_table LIMIT 200' def oursql_conn(): import oursql conn = oursql.connect( host=config.HOST, user=config.USER, passwd=config.PASS, db=config.MYDB) return conn def mysqldb_conn(): conn = MySQLdb.connect( host=config.HOST, user=config.USER, passwd=config.PASS, db=config.MYDB, cursorclass=cursors.SSCursor) return conn def two_cursors_one_conn(): """Two SSCursors can not use one connection concurrently""" def worker(conn): cursor = conn.cursor() cursor.execute(query) for row in cursor: logger.info(row) conn = mysqldb_conn() threads = [threading.Thread(target=worker, args=(conn, )) for n in range(2)] for t in threads: t.daemon = True t.start() # Second thread may hang or raise OperationalError: # File "/usr/lib/pymodules/python2.7/MySQLdb/cursors.py", line 289, in _fetch_row # return self._result.fetch_row(size, self._fetch_type) # OperationalError: (2013, 'Lost connection to MySQL server during query') for t in threads: t.join() def two_cursors_two_conn(): """Two SSCursors from independent connections can use the same table concurrently""" def worker(): conn = mysqldb_conn() cursor = conn.cursor() cursor.execute(query) for row in cursor: logger.info(row) threads = [threading.Thread(target=worker) for n in range(2)] for t in threads: t.daemon = True t.start() for t in threads: t.join() logging.basicConfig(level=logging.DEBUG, format='[%(asctime)s %(threadName)s] %(message)s', datefmt='%H:%M:%S') two_cursors_one_conn() two_cursors_two_conn()
请注意,oursql是Python的另一套MySQL绑定。oursql游标是真正的服务器端游标,默认情况下会延迟获取行。随着oursql安装,如果你改变
oursql
conn = mysqldb_conn()
至
conn = oursql_conn()
然后two_cursors_one_conn()运行而不会挂起或引发异常。
two_cursors_one_conn()