在 MySQL 中增量选择记录并在 Python 中保存到 csv

Question

我需要查询数据库进行一些数据分析，我有超过 2000 万条记录。我对数据库的访问受到限制，并且我的查询在 8 分钟后超时。因此，我正在尝试将查询分解为更小的部分，并将结果保存到 Excel 以供以后处理。

这是我目前所拥有的。如何让 python 在每个 x 数（例如 1,000,000）条记录上循环查询并将它们存储在同一个 csv 中，直到搜索到所有（20 mil++）记录？

import MySQLdb
import csv

db_main = MySQLdb.connect(host="localhost", 
                     port = 1234,
                     user="user1", 
                      passwd="test123", 
                      db="mainDB") 

cur = db_main .cursor()

cur.execute("SELECT a.user_id, b.last_name, b.first_name, 
    FLOOR(DATEDIFF(CURRENT_DATE(), c.birth_date) / 365) age,
    DATEDIFF(b.left_date, b.join_date) workDays
    FROM users a
    INNER JOIN users_signup b ON a.user_id a = b.user_id
    INNER JOIN users_personal c ON a.user_id a = c.user_id
    INNER JOIN
    (
        SELECT distinct d.a.user_id FROM users_signup d
        WHERE (user_id >=1 AND user_id <1000000)
        AND d.join_date >= '2013-01-01' and d.join_date < '2014-01-01'
    ) 
    AS t ON a.user_id = t.user_id") 

result=cur.fetchall()
c = csv.writer(open("temp.csv","wb"))
for row in result:
    c.writerow(row)

Answer 1

您的代码应如下所示。你可以通过per_query变量来调整它的性能

c = csv.writer(open("temp.csv","wb"))
offset = 0
per_query = 10000
while true:
    cur.execute("__the_query__ LIMIT %s OFFSET %s", (per_query, offset))

    rows = cur.fetchall()
    if len(rows) == 0:
        break #escape the loop at the end of data

    for row in cur.fetchall():
        c.writerow(row)

    offset += per_query

Answer 2

未经测试的代码，但这应该可以帮助您入门...

SQL = """
SELECT a.user_id, b.last_name, b.first_name, 
    FLOOR(DATEDIFF(CURRENT_DATE(), c.birth_date) / 365) age,
    DATEDIFF(b.left_date, b.join_date) workDays
    FROM users a
    INNER JOIN users_signup b ON a.user_id a = b.user_id
    INNER JOIN users_personal c ON a.user_id a = c.user_id
    INNER JOIN
    (
        SELECT distinct d.a.user_id FROM users_signup d
        WHERE (user_id >=1 AND user_id <1000000)
        AND d.join_date >= '2013-01-01' and d.join_date < '2014-01-01'
    ) 
    AS t ON a.user_id = t.user_id
    OFFSET %s LIMIT %s 
    """

BATCH_SIZE = 100000

with open("temp.csv","wb") as f:
   writer = csv.writer(f)
   cursor = db_main.cursor()

   offset = 0
   limit = BATCH_SIZE


   while True:
       cursor.execute(SQL, (offset, limit))
       for row in cursor:
           writer.writerow(row)
       else:
           # no more rows, we're done
           break
       offset += BATCH_SIZE    
cursor.close()

Answer 3

这是一个可能对您有所帮助的实现示例：

from contextlib import contextmanager
import MySQLdb
import csv

connection_args = {"host": "localhost", "port": 1234, "user": "user1", "passwd": "test123", "db": "mainDB"}

@contextmanager
def get_cursor(**kwargs):
    ''' The contextmanager allow to automatically close
    the cursor.
    '''
    db = MySQLdb.connect(**kwargs)
    cursor = db.cursor()
    try:
        yield cursor
    finally:
        cursor.close()

# note the placeholders for the limits
query = """ SELECT a.user_id, b.last_name, b.first_name,
        FLOOR(DATEDIFF(CURRENT_DATE(), c.birth_date) / 365) age,
        DATEDIFF(b.left_date, b.join_date) workDays
    FROM users a
    INNER JOIN users_signup b ON a.user_id a = b.user_id
    INNER JOIN users_personal c ON a.user_id a = c.user_id
    INNER JOIN
    (
        SELECT distinct d.a.user_id FROM users_signup d
        WHERE (user_id >= 1 AND user_id < 1000000)
        AND d.join_date >= '2013-01-01' and d.join_date < '2014-01-01'
    ) AS t ON a.user_id = t.user_id OFFSET %s LIMIT %s """

csv_file = csv.writer(open("temp.csv","wb"))

# One million at the time
STEP = 1000000
for step_nb in xrange(0, 20):
    with get_cursor(**connection_args) as cursor:
        cursor.execute(query, (step_nb * STEP, (step_nb + 1) * STEP))  # query the DB
        for row in cursor:  # use the cursor instead of fetching everything in memory
            csv_file.writerow(row)

已编辑：对批次的误解（尽管它在 user_id 上）