如何在Python中分批导出大量MySQL记录为CSV（共享主机环境）

使用NameCheap共享主机，无法使用mysqldump和OUTFILE工具，现有Python代码在迭代记录并写入CSV文件时，既耗时又占用大量内存，处理大数据量时脚本会被主机终止。

核心优化方向 #

• 避免内存累积：放弃批量缓存记录的方式，改为逐行读取并写入，控制内存占用
• 减少IO开销：直接写入压缩文件，跳过中间CSV文件的生成步骤
• 修正分页逻辑：修复原SQL查询中的参数错误，避免数据重复或遗漏
• 规范CSV生成：使用标准库csv模块处理格式，避免手动拼接的错误

优化后的完整代码 #

数据库连接函数（微调） #

import csv
import gzip
import time
from mysql import connector

def get_connection(host, user, password, db_name):
    connection = None
    try:
        connection = mysql.connector.connect(
            host=host,
            user=user,
            use_unicode=True,
            password=password,
            database=db_name,
            charset='utf8'
        )
        print('Connected')
    except Exception as ex:
        print(str(ex))
    finally:
        return connection

批量导出函数（核心优化） #

def export_batch(connection, batch_id, start_id):
    try:
        if not connection:
            return False
        
        # 修正分页逻辑：基于id连续范围查询，避免数据遗漏
        sql = "SELECT id, instrument_name, m, p, e, ts FROM options_data WHERE id >= %s LIMIT 3000"
        with connection.cursor(dictionary=True) as cursor:
            cursor.execute(sql, (start_id,))
            
            # 直接写入gzip压缩文件，无需生成中间CSV
            with gzip.open(f'{batch_id}_options.csv.gz', 'wt', encoding='utf8') as gz_file:
                writer = csv.writer(gz_file)
                # 可选：写入CSV表头
                writer.writerow(['instrument_name', 'm', 'p', 'e', 'ts'])
                
                last_id = start_id
                # 逐行读取并写入，内存仅保留单条记录
                for row in cursor:
                    formatted_ts = row['ts'].strftime('%Y-%m-%d %H:%M:%S')
                    writer.writerow([
                        row['instrument_name'],
                        str(row['m']),
                        str(row['p']),
                        str(row['e']),
                        formatted_ts
                    ])
                    last_id = row['id']
        
        print(f'{batch_id}_options.csv.gz created successfully')
        return last_id
    
    except Exception as ex:
        print(f'Exception in batch {batch_id}: {ex}')
        return None

主执行逻辑 #

DB_HOST = 'your_host'
DB_USER = 'your_user'
DB_PASSWORD = 'your_password'
DB_NAME = 'your_db'

connection = get_connection(DB_HOST, DB_USER, DB_PASSWORD, DB_NAME)
if not connection:
    print('Unable to connect MySQL Server')
    exit()

try:
    batch_id = 1
    current_start_id = 0
    total_records = 37206
    batch_size = 3000
    
    while current_start_id < total_records:
        last_id = export_batch(connection, batch_id, current_start_id)
        if not last_id:
            print(f'Batch {batch_id} failed, stopping')
            break
        
        current_start_id = last_id
        batch_id += 1
        time.sleep(1)

finally:
    if connection.is_connected():
        connection.close()
        print('Connection closed')

优化效果说明 #

• 内存占用极低：通过迭代游标逐行处理，内存占用始终维持在单条记录的大小，不会随批次累积
• IO效率提升：直接写入压缩文件，减少一次磁盘写入操作，同时节省磁盘空间
• 数据准确性：修正后的分页逻辑基于id连续范围，彻底避免原代码中offset参数错误导致的数据重复或遗漏
• 格式安全性：csv.writer自动处理字段中的特殊字符（如逗号、引号），避免手动拼接引发的CSV格式错误

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