如何快速将大容量DataFrame数据插入Amazon Redshift数据库

现有方法的核心问题 #

1. 第二种逐行执行cur.execute+conn.commit的方式，每一行都触发一次数据库交互，60万次交互必然导致极端缓慢，这是最致命的问题。
2. 第一种executemany的方式比逐行插入好，但对于Redshift这类列式数据库，原生的COPY命令才是批量加载的最优选择，效率远高于常规插入语句。

最优方案：使用Redshift COPY命令加载 #

Redshift官方推荐的批量加载方式是COPY命令，能将加载速度提升几个数量级，60万条数据通常几分钟就能完成。以下提供两种实现方式：

方案1：从内存直接加载（无需S3中转） #

将DataFrame转为CSV内存对象，通过psycopg2的copy_expert执行COPY命令：

import psycopg2
from io import StringIO
import pandas as pd

# 初始化数据库连接
conn = psycopg2.connect(
    host='redshift-####-dev.00000.us-east-1.redshift.amazonaws.com',   
    database='*****',
    user='****',
    password='*****',
    port='5439'
)
cur = conn.cursor()

# 将DataFrame转为CSV格式的内存缓冲区（无表头，匹配表列顺序）
csv_buffer = StringIO()
final_out.to_csv(csv_buffer, sep='\t', index=False, header=False)
csv_buffer.seek(0)  # 重置缓冲区指针到开头

# 执行COPY命令
copy_sql = """
    COPY odey.sfc_ca_sit_di (case_id, column_name, split_text, split_text_cnt, load_ts)
    FROM STDIN
    DELIMITER '\t'
    NULL AS ''
"""
cur.copy_expert(sql=copy_sql, file=csv_buffer)
conn.commit()

# 关闭连接
cur.close()
conn.close()

方案2：S3中转加载（超大规模数据更稳定） #

如果数据量极大或网络不稳定，先将数据上传到S3，再让Redshift从S3读取，这是企业级场景的标准做法：

import pandas as pd
import boto3
import psycopg2

# 1. 将DataFrame转为Parquet格式（比CSV更高效的列式存储）
final_out.to_parquet('data_batch.parquet', index=False)

# 2. 上传文件到S3
s3 = boto3.client('s3')
bucket_name = 'your-s3-bucket-name'
s3_file_path = 'redshift-load/data_batch.parquet'
s3.upload_file('data_batch.parquet', bucket_name, s3_file_path)

# 3. 执行Redshift COPY命令
conn = psycopg2.connect(
    host='redshift-####-dev.00000.us-east-1.redshift.amazonaws.com',   
    database='*****',
    user='****',
    password='*****',
    port='5439'
)
cur = conn.cursor()

copy_sql = """
    COPY odey.sfc_ca_sit_di (case_id, column_name, split_text, split_text_cnt, load_ts)
    FROM 's3://{}/{}'
    IAM_ROLE 'arn:aws:iam::your-account-id:role/your-redshift-access-role'
    FORMAT AS PARQUET
""".format(bucket_name, s3_file_path)

cur.execute(copy_sql)
conn.commit()

# 关闭连接
cur.close()
conn.close()

现有方法的应急优化（若无法使用COPY） #

如果必须使用插入语句，至少做以下优化：

1. 取消逐行commit，改为批量commit（比如每1万条提交一次）
2. 使用executemany批量执行，减少数据库交互次数

import psycopg2

conn = psycopg2.connect(
    host='redshift-####-dev.00000.us-east-1.redshift.amazonaws.com',   
    database='*****',
    user='****',
    password='*****',
    port='5439'
)
cur = conn.cursor()

sql = "INSERT INTO odey.sfc_ca_sit_di (case_id,column_name,split_text,split_text_cnt,load_ts) VALUES (%s,%s,%s,%s,%s)"
# 转换DataFrame为元组列表
data_tuples = [tuple(row) for row in final_out.values.tolist()]

# 批量提交，每1万条一次
batch_size = 10000
for i in range(0, len(data_tuples), batch_size):
    batch = data_tuples[i:i+batch_size]
    cur.executemany(sql, batch)
    conn.commit()

cur.close()
conn.close()

关键注意事项 #

• 确保COPY命令中指定的列顺序与DataFrame的列顺序完全一致
• 使用S3中转时，需确保Redshift拥有读取对应S3桶的IAM权限
• 批量插入时，关闭自动commit，手动控制提交频率能大幅提升效率

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