如何用Python高效导出SQL Server超6亿行查询结果至本地TXT文件

面对6亿行这种级别的数据，绝对不能把所有数据一次性拉到内存里——那直接会把机器内存撑爆。核心思路就是**「分批读取+流式写入」**，结合PyODBC的服务器端游标和高效文件IO操作，既能控制内存占用，又能保证导出速度。下面是具体实现方案：

关键优化点&步骤 #

1. 使用服务器端游标，避免客户端内存过载 #

默认情况下，PyODBC会用客户端游标，这意味着服务器会把所有查询结果一次性发送到本地内存，完全不适合大数据量。我们需要显式指定服务器端游标，让数据留在服务器端，我们分批拉取：

import pyodbc

# 建立连接时可以先设置一些优化参数
conn_str = (
    'Driver={ODBC Driver 17 for SQL Server};'
    'Server=你的服务器地址;'
    'Database=你的数据库;'
    'UID=用户名;'
    'PWD=密码;'
    'TrustServerCertificate=yes;'  # 如果需要跳过证书验证的话
)
connection = pyodbc.connect(conn_str)

# 使用服务器端游标（核心优化！）
cursor = connection.cursor(pyodbc.SQL_CURSOR_SERVER)

2. 开启NOCOUNT优化查询性能 #

在SQL语句开头加上SET NOCOUNT ON;，可以让SQL Server不返回查询的行数统计信息，减少网络传输的数据量，提升查询速度：

sql_query = """
SET NOCOUNT ON;
SELECT * FROM 你的目标表;
"""
cursor.execute(sql_query)

3. 分批读取+批量写入，平衡内存与IO #

每次从服务器拉取固定行数的结果（比如10000行，这个数值可以根据你的内存大小调整——内存大就调大，内存小就调小），然后把这批数据一次性写入文件，减少磁盘IO的次数：

# 打开文件时使用缓冲优化，减少IO开销
with open('导出结果.txt', 'w', encoding='utf-8', buffering=1024*1024) as f:
    # 先写入表头（如果需要的话）
    columns = [column[0] for column in cursor.description]
    f.write('\t'.join(columns) + '\n')  # 用制表符分隔字段，也可以换成逗号等

    batch_size = 10000  # 每批读取的行数
    while True:
        rows = cursor.fetchmany(batch_size)
        if not rows:
            break  # 没有更多数据，退出循环
        
        # 把一批行转换成字符串，一次性写入
        batch_lines = []
        for row in rows:
            # 处理字段中的特殊字符（比如换行符、制表符），避免破坏格式
            processed_row = [str(field).replace('\n', '').replace('\t', ' ') for field in row]
            batch_lines.append('\t'.join(processed_row))
        
        f.write('\n'.join(batch_lines) + '\n')

4. 其他细节优化 #

• 用with语句管理连接和文件：自动处理资源释放，避免连接泄漏或文件未关闭的问题。
• 字段处理：如果你的数据里有换行符、制表符这类可能破坏TXT格式的字符，一定要提前替换掉，不然导出的文件会乱。
• 编码选择：根据你的数据情况选择合适的编码（比如utf-8或gbk），避免乱码。
• 调整batch_size：如果你的内存足够大，可以把batch_size调到50000甚至100000，减少循环次数；如果内存紧张，就调到5000左右。

完整代码示例 #

import pyodbc

def export_large_data_to_txt():
    conn_str = (
        'Driver={ODBC Driver 17 for SQL Server};'
        'Server=你的服务器地址;'
        'Database=你的数据库;'
        'UID=用户名;'
        'PWD=密码;'
        'TrustServerCertificate=yes;'
    )
    
    try:
        connection = pyodbc.connect(conn_str)
        # 使用服务器端游标
        cursor = connection.cursor(pyodbc.SQL_CURSOR_SERVER)
        
        sql_query = """
        SET NOCOUNT ON;
        SELECT * FROM 你的目标表;
        """
        cursor.execute(sql_query)
        
        # 写入文件
        with open('large_data_export.txt', 'w', encoding='utf-8', buffering=1024*1024) as f:
            # 写入表头
            columns = [col[0] for col in cursor.description]
            f.write('\t'.join(columns) + '\n')
            
            batch_size = 10000
            while True:
                rows = cursor.fetchmany(batch_size)
                if not rows:
                    break
                
                # 处理每一行并拼接成批量字符串
                batch_content = []
                for row in rows:
                    processed_fields = [str(f).replace('\n', '').replace('\t', ' ') for f in row]
                    batch_content.append('\t'.join(processed_fields))
                
                f.write('\n'.join(batch_content) + '\n')
                
            print("数据导出完成！")
            
    except Exception as e:
        print(f"导出过程中出现错误: {str(e)}")
    finally:
        # 确保游标和连接关闭
        if 'cursor' in locals():
            cursor.close()
        if 'connection' in locals():
            connection.close()

if __name__ == "__main__":
    export_large_data_to_txt()

这个方案的核心是永远只在内存中保留一小批数据，既不会爆内存，又通过批量写入减少了磁盘IO的开销，同时服务器端游标让SQL Server只返回当前需要的批次数据，降低了网络传输压力。

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