如何在中断后使用fetchmany从指定位置继续批量读取数据？

直接用fetchmany没法在进程重启后从指定位置续读，因为游标（cursor）的读取状态仅存在内存中，进程中断后就会丢失。要实现断点续传，得靠SQL语句做分页查询，同时持久化记录已完成的进度。

核心思路 #

1. 每次成功写入一批数据后，将已处理的总行数记录到本地文件（比如progress.txt）；
2. 程序重启时，先读取该进度值作为起始偏移量；
3. 借助SQL Server的OFFSET ... FETCH NEXT语法，从偏移量位置开始分批拉取数据。

修改后的代码示例 #

import pandas as pd
import os

# 配置参数
ROWS_AT_ONCE = 100000
TABLE_NAME = "Table_Name"
PROGRESS_FILE = "progress.txt"
cols = ["列1", "列2", "列3"]  # 替换为实际列名

# 读取已完成的进度
if os.path.exists(PROGRESS_FILE):
    with open(PROGRESS_FILE, "r") as f:
        processed_rows = int(f.read().strip())
else:
    processed_rows = 0

try:
    while True:
        # 构造分页查询SQL，必须加ORDER BY保证结果顺序稳定
        curr.execute(f'''
            SELECT *
            FROM Database.dbo.{TABLE_NAME}
            ORDER BY 主键列  -- 替换为你的表的主键/唯一有序列，比如id
            OFFSET {processed_rows} ROWS
            FETCH NEXT {ROWS_AT_ONCE} ROWS ONLY;
        ''')
        
        rows = curr.fetchmany(ROWS_AT_ONCE)
        if not rows:
            print("数据迁移完成")
            # 完成后清理进度文件
            if os.path.exists(PROGRESS_FILE):
                os.remove(PROGRESS_FILE)
            break
        
        df_temp = pd.DataFrame(rows, columns=cols)
        df_temp.to_sql(TABLE_NAME, index=False, if_exists='append')
        
        # 更新进度并写入文件
        processed_rows += len(rows)
        with open(PROGRESS_FILE, "w") as f:
            f.write(str(processed_rows))
        print(f"已处理 {processed_rows} 行")
except Exception as e:
    print(f"迁移中断：{str(e)}")
    # 中断时进度已写入文件，下次启动会自动续传

关键注意事项 #

• 必须添加ORDER BY：如果不指定排序，OFFSET返回的结果集可能因原表数据变动（插入、删除）出现重复读取或遗漏，务必使用主键、唯一索引列或稳定的有序列排序；
• 保证进度文件可靠性：每次写入进度后可调用flush()确保内容同步到磁盘，避免缓存未写入导致进度丢失；
• 数据一致性处理：若迁移过程中原表有数据更新，建议使用快照隔离级别或临时锁表；如果是一次性迁移，排序主键即可保证续传的正确性。

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