Python 如何分批次导入 100W 条数据到 MySQL

对于导入百万级数据到MySQL，我通常用executemany配合分批处理。这是最直接有效的方法。

import mysql.connector
from mysql.connector import Error

def batch_insert_data(data, batch_size=10000):
    """
    分批插入数据到MySQL
    
    Args:
        data: 要插入的数据列表，每个元素是一个元组
        batch_size: 每批插入的数据量，默认10000
    """
    try:
        # 建立数据库连接
        connection = mysql.connector.connect(
            host='localhost',
            database='your_database',
            user='your_username',
            password='your_password'
        )
        
        if connection.is_connected():
            cursor = connection.cursor()
            
            # 准备插入语句
            insert_query = """
            INSERT INTO your_table (column1, column2, column3) 
            VALUES (%s, %s, %s)
            """
            
            # 分批处理数据
            total_records = len(data)
            for i in range(0, total_records, batch_size):
                batch = data[i:i + batch_size]
                cursor.executemany(insert_query, batch)
                connection.commit()
                
                print(f"已插入 {min(i + batch_size, total_records)}/{total_records} 条记录")
            
            print("数据插入完成！")
            
    except Error as e:
        print(f"数据库错误: {e}")
    finally:
        if connection.is_connected():
            cursor.close()
            connection.close()

# 示例：生成100万条测试数据
def generate_test_data(num_records=1000000):
    """生成测试数据"""
    data = []
    for i in range(num_records):
        # 假设有三列数据
        data.append((f"value1_{i}", f"value2_{i}", f"value3_{i}"))
    return data

# 使用示例
if __name__ == "__main__":
    # 生成100万条测试数据
    print("正在生成测试数据...")
    test_data = generate_test_data(1000000)
    
    # 分批插入数据
    print("开始插入数据...")
    batch_insert_data(test_data, batch_size=10000)

关键点说明：

1. executemany()方法：这是MySQL Connector/Python的批量操作方法，比逐条插入快得多。它把多个INSERT语句打包成一次数据库调用。

2. 分批处理逻辑：通过range(0, total_records, batch_size)实现数据切片，每批处理batch_size条记录。

3. 事务控制：每批数据插入后立即提交（connection.commit()），这样即使中途出错，已提交的数据也不会丢失。

4. 连接管理：使用try-except-finally确保数据库连接正确关闭。

5. 参数化查询：使用%s占位符防止SQL注入，数据通过元组列表传递。

实际使用时，你需要：

• 替换数据库连接参数
• 根据你的表结构调整INSERT语句
• 调整batch_size（通常5000-20000效果较好）

如果数据源是文件（如CSV），可以结合pandas的read_csv的chunksize参数来分批读取和处理。

简单总结：用executemany加循环分批提交最稳妥。