根据SQLAlchemy,select语句在for循环中被视为可迭代语句。其效果是,将返回大量行的select语句不会使用过多的内存。
我发现MySQL表上的以下语句:
for row in my_connections.execute(MyTable.__table__.select()):
yield row似乎没有遵循这一点,因为我溢出了可用内存,并在第一行被产生之前开始颠簸。我做错了什么?
发布于 2010-09-13 18:47:02
基本的MySQLdb游标一次从服务器获取整个查询结果。这会消耗大量的内存和时间。当您想要进行一个大型查询并从服务器中一次提取一个结果时,请使用MySQLdb.cursors.SSCursor。
因此,请尝试在创建engine时传递connect_args={'cursorclass': MySQLdb.cursors.SSCursor}
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绑定。我们的side游标是真正的fetch rows lazily by default服务器端游标。安装了oursql的情况下,如果更改
conn = mysqldb_conn()至
conn = oursql_conn()然后two_cursors_one_conn()在不挂起或引发异常的情况下运行。
https://stackoverflow.com/questions/3699532
复制相似问题