使用SQLAlchemy迭代处理MySQL千万级数据至PostgreSQL解决内存问题

我有一张包含1000万行数据的MySQL表，需要读取到本地客户端完成校验后导入PostgreSQL数据库。目前已经成功获取数据，但在处理并导入PostgreSQL时遇到了内存不足的问题。请问是否可以通过迭代器分块处理数据并插入PostgreSQL？

以下是我当前的代码：

from sqlalchemy import create_engine, MetaData, Table
# MySQL database connection
source_engine = create_engine('mysql+pymysql://user:pwd@serveraddress:3306/dbname')
source_connection = engine.connect()
# Read the entire data
data = source_connection.execute('SELECT * FROM table')
# close the MySQL connection
source_connection.close()
# function to transform data
def transform(data):
    def process_row(row):
        """do data validation on the row"""
        return row
    # process and return the incoming dataset as a list of dicts
    processed_data = [dict(zip(data.keys(), process_row(d)) for d in data]
    return processed_data
transformed_data = transform(data)
# Postgres database connection
dest_connection = create_engine('postgresql://user:pwd@serveraddress:5432/dbname')
dest_meta = MetaData(bind=dest_connection, reflect=True, schema='test')
table = Table('table_name', self.meta, autoload=True)
dest_connection.execute(table.insert().values(transformed_data))
dest_connection.close()

能否提供一种简单的实现方法？

当然可以用分块处理解决内存溢出问题！你当前的代码一次性把1000万行数据全部加载到内存，这显然会撑爆内存。我们可以用SQLAlchemy的流式查询+分块读写的思路，每次只处理一小批数据，循环完成整个迁移流程，这样内存占用就能被控制在合理范围。

下面是一个简洁的实现方案：

from sqlalchemy import create_engine, MetaData, Table

def process_row(row):
    """在这里实现你的数据校验、转换逻辑"""
    # 示例：比如校验字段格式、清理脏数据
    validated_row = dict(row)
    # 举个例子：如果某字段是字符串，去除首尾空格
    # if 'string_column' in validated_row:
    #     validated_row['string_column'] = validated_row['string_column'].strip()
    return validated_row

def migrate_in_chunks(chunk_size=10000):
    # 初始化MySQL连接，开启流式查询避免一次性加载全量数据
    source_engine = create_engine('mysql+pymysql://user:pwd@serveraddress:3306/dbname')
    with source_engine.connect() as source_conn:
        # stream_results=True 让结果集分批从MySQL服务器获取
        result = source_conn.execute('SELECT * FROM your_mysql_table', stream_results=True)
        
        # 初始化PostgreSQL连接与目标表
        dest_engine = create_engine('postgresql://user:pwd@serveraddress:5432/dbname')
        dest_meta = MetaData(bind=dest_engine, reflect=True, schema='test')
        dest_table = Table('your_pg_table', dest_meta, autoload=True)
        
        with dest_engine.connect() as dest_conn:
            while True:
                # 每次读取指定数量的行数据
                chunk = result.fetchmany(chunk_size)
                if not chunk:
                    break  # 所有数据处理完成
                
                # 批量处理当前块的行数据
                processed_chunk = [process_row(row) for row in chunk]
                
                # 批量插入到PostgreSQL
                dest_conn.execute(dest_table.insert().values(processed_chunk))
                dest_conn.commit()  # 提交当前批次的事务
                
                print(f"已完成 {len(processed_chunk)} 行数据的处理与插入")

if __name__ == "__main__":
    # 可根据自身内存情况调整chunk_size，比如内存充足可以调到20000，紧张则设为5000
    migrate_in_chunks(chunk_size=10000)

核心优化点说明： #

• 流式查询：通过stream_results=True让MySQL驱动不会一次性把所有结果拉到客户端内存，而是按需分批获取。
• 分块读写：用fetchmany(chunk_size)控制每次加载到内存的数据量，处理完一批就插入一批，避免内存堆积。
• 上下文管理器：用with语句管理数据库连接，自动处理连接的打开与关闭，避免资源泄漏。

你可以根据自己的机器内存情况调整chunk_size参数，找到内存占用和迁移效率的平衡点。

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