如何使用Python向Amazon Redshift批量插入多行数据？

嘿，针对你要把元组列表批量插入Amazon Redshift的需求，我给你整理了两种生产环境常用的实用方案——毕竟逐条插入效率太低，批量操作才是正道：

方法一：用psycopg2的executemany（简单直接，适合中小数据量） #

Redshift兼容PostgreSQL协议，所以我们可以用最常用的psycopg2驱动来实现批量插入，executemany方法能一次性处理整个元组列表，比循环逐条插入高效得多：

import psycopg2
from psycopg2 import sql

# 先建立Redshift连接，替换成你的实际配置
conn = psycopg2.connect(
    dbname='你的数据库名',
    user='你的用户名',
    password='你的密码',
    host='你的Redshift端点',
    port='5439'  # Redshift默认端口，若有修改请对应调整
)
cur = conn.cursor()

# 定义插入SQL，这里假设你的目标表名为target_table，字段顺序和元组完全匹配
# 请把col1~col10替换成你表的实际字段名
insert_sql = sql.SQL("""
    INSERT INTO {} (col1, col2, col3, col4, col5, col6, col7, col8, col9, col10)
    VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s)
""").format(sql.Identifier('target_table'))

# 执行批量插入
try:
    cur.executemany(insert_sql, my_rows)
    conn.commit()
    print(f"搞定！成功插入{len(my_rows)}条数据")
except Exception as e:
    conn.rollback()
    print(f"插入失败了，错误信息：{str(e)}")
finally:
    # 别忘了关闭游标和连接
    cur.close()
    conn.close()

方法二：用copy_from（超大数据量首选，性能拉满） #

如果你的数据量超过10万条，Redshift官方更推荐用COPY命令，因为它是专门为批量数据加载优化的，性能比executemany高一个量级。我们可以把元组列表转换成内存中的CSV文件对象，再用psycopg2的copy_from来执行：

import psycopg2
from io import StringIO

# 建立连接，同上
conn = psycopg2.connect(...)
cur = conn.cursor()

# 将元组列表转换成制表符分隔的字符串IO（用制表符避免逗号冲突，也可以用其他分隔符）
output = StringIO()
for row in my_rows:
    # 注意：如果字段里有制表符，需要额外处理，这里假设你的数据里没有
    line = '\t'.join(map(str, row)) + '\n'
    output.write(line)
output.seek(0)  # 把文件指针移到开头，不然读不到内容

# 执行COPY加载
try:
    cur.copy_from(
        file=output,
        table='target_table',
        columns=('col1', 'col2', 'col3', 'col4', 'col5', 'col6', 'col7', 'col8', 'col9', 'col10'),
        sep='\t'
    )
    conn.commit()
    print(f"批量加载完成！一共导入了{len(my_rows)}条数据")
except Exception as e:
    conn.rollback()
    print(f"加载出错：{str(e)}")
finally:
    output.close()
    cur.close()
    conn.close()

几个要注意的点 #

• 一定要确保元组的顺序和目标表的字段顺序完全一致，不然会出现数据类型不匹配或者插错列的问题。
• 你的日期时间字段已经是YYYY-MM-DD HH:MI:SS格式，刚好符合Redshift的要求，不需要额外转换。
• 如果用IAM角色认证，记得在连接时开启SSL，并且可以不用密码，改用IAM认证的方式（具体可以参考Redshift官方文档，但上面的例子用的是常规密码认证）。
• 批量操作前最好先拿几条测试数据跑一遍，确认字段类型匹配，避免大规模出错。

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