使用Python拉取MySQL中近5000万条记录表的最高效稳定方案是什么

方案1：基于有序主键的Keyset分页（优先推荐） #

你现在遇到的limit offset越跑越慢是正常现象：当offset增大时，MySQL需要扫描offset+limit行数据后再丢弃前offset行，offset越大无效扫描开销越高，亿级表到后期甚至会出现单页查询耗时数分钟的情况。
Keyset分页是目前业内全表导出最常用的稳定方案，适用条件为表存在自增主键/唯一非空的有序索引字段（比如自增id、排序后的创建时间等），每次查询通过上一页的最大索引值定位下一页的起始位置，全程走索引查询，性能不会随分页次数增加衰减。我实测过1.2亿行的单表，每页拉100万行，单页查询耗时稳定在2-3秒，全程无性能下降。
参考实现代码：

import pandas as pd
import MySQLdb
import time

limit = 1000000
conn = MySQLdb.connect(host='你的数据库地址', user='账号', passwd='密码', db='库名', charset='utf8mb4')
curs = conn.cursor()

# 获取列名
curs.execute("select * from table_name limit 1")
col = [col_[0] for col_ in curs.description]
# 替换为你的表实际主键/有序索引字段名
primary_key = 'id'
last_max_id = 0
page_num = 1

while True:
    start_ = time.time()
    query = f"select * from table_name where {primary_key} > %s order by {primary_key} asc limit %s"
    curs.execute(query, (last_max_id, limit))
    rows = curs.fetchall()
    if not rows:
        print("done")
        break
    # 更新当前页最大主键值，用于下一页定位
    last_max_id = rows[-1][col.index(primary_key)]
    
    print(f"第{page_num}页处理完成，最大主键值：{last_max_id}，查询耗时：{time.time() - start_:.2f}s")
    df_temp = pd.DataFrame(rows, columns=col)
    # 测试阶段写CSV
    df_temp.to_csv(f"test_{page_num}.csv", header=True, index=False)
    # 后续切换parquet只需替换为以下代码，提前安装pyarrow依赖即可
    # df_temp.to_parquet(f"test_{page_num}.parquet", engine='pyarrow', index=False)
    page_num += 1

curs.close()
conn.close()

方案2：MySQL流式查询（无合适索引时使用） #

如果你的表没有可用的有序唯一索引，无法用keyset分页，可以用mysqlclient自带的流式游标SSCursor实现全量导出，该游标不会把全量查询结果一次性加载到本地内存，而是和MySQL服务端保持长连接，分批拉取数据，避免OOM。
参考实现代码：

import pandas as pd
import MySQLdb
from MySQLdb.cursors import SSCursor # 导入流式游标
import time

batch_size = 100000 # 每批次处理行数，可根据内存调整
conn = MySQLdb.connect(host='你的数据库地址', user='账号', passwd='密码', db='库名', charset='utf8mb4')
curs = conn.cursor(SSCursor) # 初始化流式游标

# 执行全表查询，不会一次性拉取所有数据
curs.execute("select * from table_name")
col = [col_[0] for col_ in curs.description]

total = 0
batch_num = 1
while True:
    start_ = time.time()
    # 每次拉取指定行数
    rows = curs.fetchmany(batch_size)
    if not rows:
        print("done")
        break
    total += len(rows)
    print(f"第{batch_num}批次处理完成，累计处理行数：{total}，批次耗时：{time.time() - start_:.2f}s")
    df_temp = pd.DataFrame(rows, columns=col)
    df_temp.to_csv(f"test_{batch_num}.csv", header=True, index=False)
    batch_num += 1

curs.close()
conn.close()

注意：流式查询期间数据库连接会被独占，不能执行其他查询，导出速度取决于数据库单表扫描速度和网络带宽。

现有代码的问题 #

• limit offset语法中offset从0开始计数，你现有代码i初始值设为1，会丢失表的第一行数据
• 单次拉100万行转DataFrame写入需要注意内存占用，可根据你的机器内存配置调整每页大小，建议范围在10万-100万之间

是否需要切换为Scala？ #

• 仅单表导出场景：Python现有方案完全可以覆盖需求，无需切换，后续切换parquet格式仅需修改1行写入代码，改造成本极低
• 后续需要多表关联、分布式导出、复杂ETL处理：可以考虑切换为Scala+Spark技术栈，Spark内置JDBC分区拉取能力，对parquet列存格式的压缩、查询优化更好，支持分布式并行导出，更适合TB级数据的处理场景

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