PyMySQL游标引发内存上涨问题的排查与解决咨询

我使用Python 3.9和PyMySQL 1.1.0版本，通过DBUtils管理数据库连接，循环运行模型时需要从MySQL获取数据。经memory_profiler检测，内存上涨发生在for row in cursor:循环执行时，相关代码如下：

def databaseSelect(dataConnect,unit, kkspoint, kksname, starttime, endtimePlus):
    conn = dataConnect.connection()
    cursor = conn.cursor()  
    backdict = {}
    data1 = {}
    tables = []
    data = []
    unit = str(unit)
    kkspoint = str(kkspoint)
    kksname = str(kksname)
    starttime = str(starttime)
    endtimePlus = int(endtimePlus)
    starttime = datetime.datetime.strptime(starttime, "%Y-%m-%d %H:%M:%S")
    endtime = (starttime + datetime.timedelta(minutes=endtimePlus)).strftime('%Y-%m-%d %H:%M:%S')
    sql = 'show tables;'
    cursor.execute(sql)

    for row in cursor:
        tables.append(str(row[0].decode('utf-8')))
        del row
    cursor = conn.close()
    conn.close()
    del (cursor,conn)
    findState = False
    for table in tables:
        a = table.split('y')[0].split('a')[2]
        if a == kkspoint:
            b = table.split('y')[1].split('m')[0]
            c = table.split('y')[1].split('m')[1].split('d')
            times = b + '-' + c[0] + '-' + c[1] + ' 00:00:00'
            originTime = datetime.datetime.strptime(times, '%Y-%m-%d %H:%M:%S')
            backdict.setdefault(a, []).append(originTime)
            del (a,b,c,times,originTime)
        del table
    for key in backdict.keys():
        backdict[key].sort()
        data1[key] = backdict[key]
        for i, timefind in enumerate(data1[key]):
            if starttime >= timefind and starttime < backdict[key][i + 1]:
                checkTimeFind = timefind
                findState = True
                break
        if findState:
            break
        del key

    if unit == 'zj':
        kkspoint = 'Data' + kkspoint + 'y' + str(checkTimeFind.year) + 'm' + str(
            checkTimeFind.month) + 'd' + str(
            checkTimeFind.day)
    elif unit == 'fw':
        kkspoint = 'fwData' + kkspoint + 'y' + str(checkTimeFind.year) + 'm' + str(
            checkTimeFind.month) + 'd' + str(
            checkTimeFind.day)
    else:
        print('#' * 100 + '\nIPdownloader.databaseSelect\nwrong unit')
        raise IndexError

    sql = "select {} from {} where time between '{}' and '{}';".format(kksname, kkspoint, starttime, endtime)
    conn = dataConnect.connection()
    cursor = conn.cursor()  
    cursor.execute(sql)
    for row in cursor:
        data.append(row)
        del row
    cursor.close()
    conn.close()
    del (backdict, data1, sql, tables, cursor, conn, unit, kkspoint, kksname, starttime, endtime, findState)
    return data

问题根源 #

PyMySQL默认游标会一次性把查询结果全部加载到内存中，当数据量较大时，内存占用会急剧上升。手动del row无法解决问题，因为游标本身已经持有了整个结果集的引用，而且Python垃圾回收不会立即释放这部分内存。

解决步骤 #

1. 改用流式游标（SSCursor） #

PyMySQL提供的SSCursor（流式游标）不会一次性加载所有结果，而是逐行从服务器获取数据，能大幅降低内存消耗。修改游标创建代码：

# 导入pymysql游标类型
import pymysql.cursors

# 创建游标时指定SSCursor
cursor = conn.cursor(pymysql.cursors.SSCursor)

注意：使用流式游标期间必须保持连接打开，处理完数据后再关闭游标和连接，否则会抛出连接关闭的错误。

2. 优化数据存储逻辑 #

• 如果模型支持逐行处理数据，直接在循环内处理，不用把所有数据存入data列表：

  # 替换原有的data.append(row)逻辑
  for row in cursor:
      # 直接调用模型处理当前行
      your_model_process_function(row)
  

• 若必须收集数据，用生成器代替列表，减少内存占用：

  def data_generator(cursor):
      for row in cursor:
          yield row
  
  # 生成器可以直接迭代，或者按需转为列表
  data = list(data_generator(cursor))
  

3. 修正连接与游标关闭逻辑 #

原代码中cursor = conn.close()是错误的，应该先关闭游标再关闭连接，规范资源释放：

cursor.close()
conn.close()

4. 移除不必要的del操作 #

代码中大量手动删除变量的操作没必要，Python的垃圾回收机制会自动处理局部变量，过度手动删除反而可能干扰回收效率，只保留游标和连接的关闭操作即可。

修改后的关键代码示例 #

import pymysql.cursors
import datetime

def databaseSelect(dataConnect,unit, kkspoint, kksname, starttime, endtimePlus):
    conn = dataConnect.connection()
    # 使用流式游标获取表列表
    cursor = conn.cursor(pymysql.cursors.SSCursor)
    sql = 'show tables;'
    cursor.execute(sql)

    tables = []
    for row in cursor:
        tables.append(str(row[0].decode('utf-8')))
    # 规范关闭游标和连接
    cursor.close()
    conn.close()

    backdict = {}
    findState = False
    checkTimeFind = None
    for table in tables:
        a = table.split('y')[0].split('a')[2]
        if a == kkspoint:
            b = table.split('y')[1].split('m')[0]
            c = table.split('y')[1].split('m')[1].split('d')
            times = b + '-' + c[0] + '-' + c[1] + ' 00:00:00'
            originTime = datetime.datetime.strptime(times, '%Y-%m-%d %H:%M:%S')
            backdict.setdefault(a, []).append(originTime)
            break

    for key in backdict.keys():
        backdict[key].sort()
        for i, timefind in enumerate(backdict[key]):
            if i + 1 >= len(backdict[key]):
                continue
            if starttime >= timefind and starttime < backdict[key][i + 1]:
                checkTimeFind = timefind
                findState = True
                break
        if findState:
            break

    if unit == 'zj':
        kkspoint = 'Data' + kkspoint + 'y' + str(checkTimeFind.year) + 'm' + str(
            checkTimeFind.month) + 'd' + str(checkTimeFind.day)
    elif unit == 'fw':
        kkspoint = 'fwData' + kkspoint + 'y' + str(checkTimeFind.year) + 'm' + str(
            checkTimeFind.month) + 'd' + str(checkTimeFind.day)
    else:
        print('#' * 100 + '\nIPdownloader.databaseSelect\nwrong unit')
        raise IndexError

    starttime_str = starttime.strftime('%Y-%m-%d %H:%M:%S')
    sql = "select {} from {} where time between '{}' and '{}';".format(kksname, kkspoint, starttime_str, endtime)
    conn = dataConnect.connection()
    # 再次使用流式游标查询数据
    cursor = conn.cursor(pymysql.cursors.SSCursor)
    cursor.execute(sql)

    # 若模型支持逐行处理，直接在这里操作
    # for row in cursor:
    #     your_model.process(row)

    # 若需要返回数据，用生成器或按需收集
    data = []
    for row in cursor:
        data.append(row)

    cursor.close()
    conn.close()
    return data

内容的提问来源于stack exchange，提问作者peanuts