使用Quota Guard从Python3写入MySQL时频繁遇通信包读取错误求助

问题背景 #

我们每日运行若干部署在PythonAnywhere上的Python3脚本，通过Quota Guard（qgtunnel）向MySQL写入数据。数据量每日波动，数据量较大时频繁出现以下错误：

• Python端报错：

  pymysql.err.OperationalError: (2013, 'Lost connection to MySQL server during query')
  

• MySQL端日志：

  2022-08-17T17:40:39.754094Z 77949 [Note] Aborted connection 77949 to db: 'db' user: 'user' host: 'IP' (Got an error reading communication packets)
  

原代码逻辑为每处理1000条记录提交事务、休眠6秒并重置数据库连接，但仍频繁报错。MySQL版本为5.7.25，相关配置：

max_allowed_packet: 1073741824
interactive_timeout: 31536000
wait_timeout: 31536000
connect_timeout: 31536000

按照@Rick James的建议改用executemany后错误消失，现总结类似场景的优化经验：

核心优化建议 #

• 优先用批量操作替代单条循环写入：
  单条循环执行SQL会产生大量网络交互，尤其通过Quota Guard这类代理连接时，频繁的数据包传输更容易触发连接中断。executemany将多条操作打包发送，大幅减少网络请求次数，直接降低连接异常概率。
• 合理控制批量大小：
  不是批量越大越好，需根据单条记录大小调整。比如单条记录较大时，可将批量数从1000下调至500或200，避免单批次数据包过大触发代理层隐性限制（即使MySQL的max_allowed_packet配置足够）。
• 调整连接复用策略：
  原逻辑中每批次重置连接反而增加了连接建立的开销和风险。改用批量操作后，可保持连接复用，仅在捕获到OperationalError这类连接异常时再重新建立连接，无需主动休眠和重置。
• 适配代理层超时配置：
  通过Quota Guard代理时，需确认代理的超时规则，避免代理主动断开长时间空闲的连接。如果批量操作间隔较长，可在间隔期发送SELECT 1这类简单心跳查询维持连接。
• 添加异常重试机制：
  即使做了上述优化，仍可能因网络波动出现连接问题。建议给数据库操作添加重试逻辑，捕获2013这类连接错误时，重试2-3次后再抛出异常，提升脚本稳定性。

代码优化示例参考 #

原单条写入逻辑（简化版） #

import pymysql
import time

conn = pymysql.connect(host='xxx', user='xxx', password='xxx', db='xxx')
cursor = conn.cursor()

records = get_records()  # 获取待写入数据
count = 0
for record in records:
    sql = "INSERT INTO table (col1, col2) VALUES (%s, %s)"
    cursor.execute(sql, (record['col1'], record['col2']))
    count += 1
    if count % 1000 == 0:
        conn.commit()
        time.sleep(6)
        # 重置连接
        cursor.close()
        conn.close()
        conn = pymysql.connect(host='xxx', user='xxx', password='xxx', db='xxx')
        cursor = conn.cursor()
conn.commit()
cursor.close()
conn.close()

优化后executemany逻辑（简化版） #

import pymysql
from tenacity import retry, stop_after_attempt, retry_if_exception_type

@retry(stop=stop_after_attempt(3), retry=retry_if_exception_type(pymysql.err.OperationalError))
def batch_insert(conn, records):
    cursor = conn.cursor()
    sql = "INSERT INTO table (col1, col2) VALUES (%s, %s)"
    data = [(r['col1'], r['col2']) for r in records]
    cursor.executemany(sql, data)
    conn.commit()
    cursor.close()

def main():
    conn = pymysql.connect(host='xxx', user='xxx', password='xxx', db='xxx')
    records = get_records()
    batch_size = 1000
    # 分批次处理
    for i in range(0, len(records), batch_size):
        batch = records[i:i+batch_size]
        try:
            batch_insert(conn, batch)
        except Exception as e:
            # 记录失败批次并尝试重连
            print(f"Batch {i//batch_size} failed: {e}")
            conn.close()
            conn = pymysql.connect(host='xxx', user='xxx', password='xxx', db='xxx')
    conn.close()

if __name__ == "__main__":
    main()

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