从Amazon Redshift导出千万级数据至Pandas时内存不足的解决方案求助

嘿，这个问题我之前帮不少人踩过坑——1000万行直接拉进Pandas确实容易爆内存，毕竟Pandas是把全量数据塞进内存的主儿。给你几个实用的方案，从快速修改现有代码到进阶优化都有，你可以根据自己的环境挑：

核心思路就是不一次性拉全量数据，每次读一部分（chunk），然后逐步追加到pickle文件里。这种方法不需要额外的云服务，直接改你现有的psycopg2代码就行。

代码示例： #

import psycopg2
import pandas as pd
import pickle

# 初始化Redshift连接
conn = psycopg2.connect(
    dbname='你的数据库名',
    user='你的用户名',
    password='你的密码',
    host='Redshift集群地址',
    port='5439'
)
cur = conn.cursor()

# 先获取总行数（用来跟踪进度，可选）
cur.execute("SELECT COUNT(*) FROM 你的表名")
total_rows = cur.fetchone()[0]
chunk_size = 100000  # 每次读10万行，根据你的内存调整，内存小就设成5万

# 分批读取并写入pickle
first_chunk = True
offset = 0

while offset < total_rows:
    # 用LIMIT + OFFSET分批查询
    cur.execute(f"SELECT 列1, 列2, 列3 FROM 你的表名 LIMIT {chunk_size} OFFSET {offset}")
    # 获取列名
    columns = [desc[0] for desc in cur.description]
    # 读取当前chunk的数据
    chunk_data = cur.fetchall()
    df_chunk = pd.DataFrame(chunk_data, columns=columns)
    
    # 关键优化：压缩数据类型，减少内存占用
    # 比如把字符串列转成category类型（如果重复值多）
    # df_chunk['字符串列'] = df_chunk['字符串列'].astype('category')
    # 把int64转成int32（如果数值范围允许）
    # df_chunk['整数列'] = df_chunk['整数列'].astype('int32')
    
    # 写入pickle：第一次创建文件，后续追加
    if first_chunk:
        df_chunk.to_pickle('最终数据.pkl')
        first_chunk = False
    else:
        with open('最终数据.pkl', 'ab') as f:
            pickle.dump(df_chunk, f)
    
    offset += chunk_size
    print(f"已处理 {min(offset, total_rows)}/{total_rows} 行")

cur.close()
conn.close()

# 读取合并后的pickle文件
def load_pickle_chunks(file_path):
    chunks = []
    with open(file_path, 'rb') as f:
        while True:
            try:
                chunks.append(pickle.load(f))
            except EOFError:
                break
    return pd.concat(chunks, ignore_index=True)

df = load_pickle_chunks('最终数据.pkl')

额外优化：用Server-Side Cursor #

如果你的Redshift表有主键，用Server-Side Cursor会比OFFSET更高效（OFFSET大的时候性能会下降）：

# 创建Server-Side Cursor
cur = conn.cursor(name='redshift_server_cursor')
cur.itersize = chunk_size  # 每次从服务器取多少行
cur.execute("SELECT * FROM 你的表名 ORDER BY 主键列")

for chunk in cur:
    # 处理每个chunk的逻辑和上面一样
    pass

如果数据量特别大（比如1000万行已经接近内存极限），用Redshift自带的UNLOAD命令把数据导出到S3，拆成多个小文件，再用Pandas分批读取。这种方法效率更高，因为Redshift导出数据的速度比客户端拉取快很多。

第一步：执行UNLOAD命令（在Redshift中运行） #

UNLOAD ('SELECT 列1, 列2 FROM 你的表名')
TO 's3://你的S3桶/数据路径/前缀_'
IAM_ROLE 'arn:aws:iam::你的AWS账号ID:role/Redshift访问S3的角色'
FORMAT AS CSV
DELIMITER ','
HEADER
MAXFILESIZE 64 MB  # 每个文件最大64MB，自动拆分
REGION '你的S3区域，比如us-east-1';

第二步：分批读取S3上的CSV并保存为pickle #

import pandas as pd
import s3fs
import pickle

# 初始化S3文件系统
fs = s3fs.S3FileSystem()

# 获取所有导出的CSV文件路径
file_paths = fs.glob('s3://你的S3桶/数据路径/前缀_*.csv')

# 分批读取并写入pickle
first_chunk = True
for file in file_paths:
    df_chunk = pd.read_csv(f's3://{file}')
    
    # 同样做数据类型优化
    # df_chunk['字符串列'] = df_chunk['字符串列'].astype('category')
    
    if first_chunk:
        df_chunk.to_pickle('最终数据.pkl')
        first_chunk = False
    else:
        with open('最终数据.pkl', 'ab') as f:
            pickle.dump(df_chunk, f)
    print(f"已处理文件: {file}")

如果你的数据量未来还会增长，直接换成Dask更省心。Dask是并行计算库，它的DataFrame API和Pandas几乎完全兼容，但能处理比内存大得多的数据集——它会自动把数据分成多个chunk并行处理，不需要手动分批。

代码示例： #

from dask import dataframe as dd
from sqlalchemy import create_engine

# 创建Redshift连接字符串
conn_str = 'postgresql+psycopg2://用户名:密码@Redshift地址:5439/数据库名'
engine = create_engine(conn_str)

# 用Dask读取Redshift数据，自动分批
# 注意：需要指定一个索引列（比如主键），Dask会根据这个列拆分数据
dask_df = dd.read_sql_table(
    table_name='你的表名',
    con=engine,
    index_col='主键列名'
)

# 直接保存为pickle（Dask会自动生成多个小pickle文件）
dask_df.to_pickle('dask_data.pkl')

# 如果后续需要转成Pandas DataFrame（内存允许的话）
pandas_df = dask_df.compute()
pandas_df.to_pickle('全量数据.pkl')

关键提示： #

• **别用SELECT ***：只选你需要的列，能大幅减少数据量和内存占用。
• 数据类型优化是核心：把字符串列转成category、数值列降精度（比如int64转int32），能减少50%以上的内存消耗。
• 调整chunk_size：根据你的内存大小来，比如16G内存可以设100万行，8G内存就设50万行。

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