通过 Python 批量从 MySQL 中检索数据答案

【问题标题】：Retrieving Data from MySQL in batches via Python通过 Python 批量从 MySQL 中检索数据
【发布时间】：2015-09-17 08:20:45
【问题描述】：

由于量大，我想分批进行此过程。

这是我的代码：

 getconn = conexiones()
 con = getconn.mysqlDWconnect()
 with con:
     cur = con.cursor(mdb.cursors.DictCursor)
     cur.execute("SELECT id, date, product_id, sales FROM sales")
     rows = cur.fetchall()

如何实现索引批量获取数据？

【问题讨论】：

标签： python mysql

【解决方案1】：

第一点：python db-api.cursor 是一个迭代器，所以除非你真的需要一次在内存中加载整个批次，你可以从使用这个特性开始，即而不是：

cursor.execute("SELECT * FROM mytable")
rows = cursor.fetchall()
for row in rows:
   do_something_with(row)

你可以：

cursor.execute("SELECT * FROM mytable")
for row in cursor:
   do_something_with(row)

如果您的 db 连接器的实现仍然没有正确使用此功能，那么是时候添加 LIMIT 和 OFFSET 了：

# py2 / py3 compat
try:
    # xrange is defined in py2 only
    xrange
except NameError:
    # py3 range is actually p2 xrange
    xrange = range

cursor.execute("SELECT count(*) FROM mytable")
count = cursor.fetchone()[0]
batch_size = 42 # whatever

for offset in xrange(0, count, batch_size):
    cursor.execute(
        "SELECT * FROM mytable LIMIT %s OFFSET %s", 
        (batch_size, offset))
   for row in cursor:
       do_something_with(row)

【讨论】：

你的解决方案比我的干净多了！
只是为了给所有评论这个“xrange”的人做个说明，在 Python 3 中已经贬值了。
如何扩展此解决方案以将每次迭代中的 batch_size 行数写入 csv 文件？例如在每次迭代中写入 1000 行直到计数结束？

【解决方案2】：

要扩展 akalikin 的答案，您可以使用逐步迭代将查询拆分为块，然后使用 LIMIT 和 OFFSET 执行查询。

cur = con.cursor(mdb.cursors.DictCursor)
cur.execute("SELECT COUNT(*) FROM sales")

for i in range(0,cur.fetchall(),5):
    cur2 = con.cursor(mdb.cursors.DictCursor)
    cur2.execute("SELECT id, date, product_id, sales FROM sales LIMIT %s OFFSET %s" %(5,i))
    rows = cur2.fetchall()
    print rows

【讨论】：

【解决方案3】：

你可以使用

SELECT id, date, product_id, sales FROM sales LIMIT X OFFSET Y;

其中 X 是您需要的批次大小，Y 是当前偏移量（例如 X 倍当前迭代次数）

【讨论】：

【解决方案4】：

谢谢，这是我根据您的建议实施它的方法：

control = True
index = 0
while control==True:
   getconn = conexiones()
   con = getconn.mysqlDWconnect()
   with con:
        cur = con.cursor(mdb.cursors.DictCursor)
        query = "SELECT id, date, product_id, sales FROM sales  limit 10 OFFSET " + str(10 * (index))
        cur.execute(query)
        rows = cur.fetchall()
        index = index+1        
        if len(rows)== 0:
            control=False
   for row in rows:
        dataset.append(row)

【讨论】：

我看到你正在收集似乎是 list 或 list-like dataset 对象的行，这破坏了批处理和直接光标迭代的全部意义，即避免将整个shebang加载到内存中...请问您是如何使用此dataset对象的？这里可能有更好的解决方案，Python 对惰性求值有很强的支持。
我需要检索 18M 行数据集，之后我必须清理一些我不感兴趣的引用（子集）并将其与其他引用合并。
你真的需要在内存中获取整个数据集吗？在处理“巨大”数据集时，规范模式是使用“流/处理/写入”解决方案。
我一直在考虑它......我会尝试实现一个循环来减轻这个过程。非常感谢您的建议。 :)
您可能想看看生成器函数 - 它们是一种以极低的成本提供惰性求值的好方法。