如何用Python分块从Snowflake提取400万行数据并保存为CSV？

修正后的完整代码（按要求存入数组后统一保存） #

import csv

chunksize = 1000000
# 修正SQL语法，指定要查询的字段（用*表示所有字段，也可按需指定具体字段）
# 添加LIMIT确保只提取400万行
query = "SELECT * FROM emp_table LIMIT 4000000;"
cursor = conn.cursor()
cursor.execute(query)
r = []

# 循环分批拉取数据
while True:
    # 每次获取指定大小的数据块
    chunk = cursor.fetchmany(chunksize)
    if not chunk:
        # 没有数据时终止循环
        break
    # 将数据块中的每行数据追加到数组r（用extend而非append，避免嵌套列表）
    r.extend(chunk)
    print(f"已加载{len(r)}行数据")

# 将数组中的数据写入CSV文件
with open('emp_data.csv', 'w', newline='', encoding='utf-8') as csvfile:
    csv_writer = csv.writer(csvfile)
    # 写入表头（从游标获取字段名称）
    csv_writer.writerow([col[0] for col in cursor.description])
    # 写入所有数据行
    csv_writer.writerows(r)

# 释放数据库资源
cursor.close()
conn.close()

内存优化版（分批写入CSV，避免占用过多内存） #

如果400万行数据存入数组导致内存紧张，可以直接分批写入CSV，无需先存数组：

import csv

chunksize = 1000000
query = "SELECT * FROM emp_table LIMIT 4000000;"
cursor = conn.cursor()
cursor.execute(query)

with open('emp_data.csv', 'w', newline='', encoding='utf-8') as csvfile:
    csv_writer = csv.writer(csvfile)
    # 先写入表头
    csv_writer.writerow([col[0] for col in cursor.description])
    
    total_written = 0
    while True:
        chunk = cursor.fetchmany(chunksize)
        if not chunk:
            break
        # 直接写入当前数据块
        csv_writer.writerows(chunk)
        total_written += len(chunk)
        print(f"已写入{total_written}行数据")

cursor.close()
conn.close()

关键说明 #

• 修正了原SQL的语法错误：select from需改为SELECT * FROM（或指定具体字段，指定字段能减少数据传输量，提升效率）
• 使用fetchmany()实现分块取数，降低数据库单次查询负载
• 用extend()而非append()处理数据块，确保数组r中是扁平的行数据，而非嵌套列表
• 添加LIMIT 4000000严格控制提取的总行数
• 操作完成后务必关闭游标和数据库连接，避免资源泄漏

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