SQLAlchemy memory hog en la instrucción select

Question

Según SQLAlchemy, las sentencias select se tratan como iterables en los bucles for. El efecto es que una instrucción de selección que devolvería una gran cantidad de filas no usa una cantidad excesiva de memoria.

Estoy descubriendo que la siguiente declaración en una tabla de MySQL:

for row in my_connections.execute(MyTable.__table__.select()): 
    yield row 

No parece seguir esto, como desbordamiento de la memoria disponible y comenzará paliza antes de que se produjo la primera fila. ¿Qué estoy haciendo mal?

Answer 1

El cursor MySQLdb básico obtiene todo el resultado de la consulta a la vez del servidor. Esto puede consumir mucha memoria y tiempo. Use MySQLdb.cursors.SSCursor cuando desee realizar una gran consulta y obtenga los resultados del servidor de uno en uno.

Por lo tanto, tratan de pasar connect_args={'cursorclass': MySQLdb.cursors.SSCursor} al crear el 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() 

Ver http://www.sqlalchemy.org/trac/ticket/1089

---

Tenga en cuenta que el uso de SSCursor bloquea la tabla hasta que la captación es completa. Esto afecta a otros cursores que utilizan la misma conexión: dos cursores de la misma conexión no pueden leerse de la tabla al mismo tiempo.

Sin embargo, los cursores de diferentes conexiones pueden leer de la misma tabla al mismo tiempo.

Aquí hay un código que demuestra el problema:

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() 

Tenga en cuenta que oursql es un conjunto alternativo de los enlaces de MySQL para Python. Los cursores oursql son verdaderos cursores del lado del servidor que fetch rows lazily by default. Con oursql instalado, si cambia

conn = mysqldb_conn() 

a

conn = oursql_conn() 

luego two_cursors_one_conn() carreras sin colgar o lanzar una excepción.