使用 python 和 psycopg2 将数据从 S3 复制到 AWS redshift答案

【问题标题】：Copying data from S3 to AWS redshift using python and psycopg2使用 python 和 psycopg2 将数据从 S3 复制到 AWS redshift
【发布时间】：2013-03-14 03:51:38
【问题描述】：

我在执行复制命令以从 python 将数据从 S3 加载到 Amazon 的 Redshift 时遇到问题。
我有以下复制命令：

copy moves from 's3://<my_bucket_name>/moves_data/2013-03-24/18/moves'
credentials 'aws_access_key_id=<key_id>;aws_secret_access_key=<key_secret>'
removequotes
delimiter ',';

当我使用 SQL Workbench/j 执行此命令时，一切都按预期工作，但是当我尝试使用 python 和 psycopg2 执行此命令时，命令通过 OK，但没有加载数据，也没有引发错误。
尝试了以下两个选项（假设 psycopg2 连接正常，因为它是）：

cursor.execute(copy_command)  
cursor.copy_expert(copy_command, sys.stdout)

两者都通过且没有警告但未加载数据

想法？

谢谢

【问题讨论】：

需要在execute()之后调用commit()否则不会生效。

标签： python psycopg2 amazon-redshift

【解决方案1】：

语法应该类似于 DDL 语句

# Create table
c.execute('''CREATE TABLE stocks
             (date text, trans text, symbol text, qty real, price real)''')

【讨论】：

【解决方案2】：

我已经成功地使用了这个精确的设置（psycopg2 + redshift + COPY）。你后来答应了吗？ SQL Workbench 默认为自动提交，而 psycopg2 默认为打开事务，因此在您在连接上调用 commit() 之前，数据将不可见。

完整的工作流程是：

conn = psycopg2.connect(...)
cur = conn.cursor()
cur.execute("COPY...")
conn.commit()

我不相信 copy_expert() 或任何 cursor.copy_* 命令可与 Redshift 一起使用。

【讨论】：

这解决了我的问题。我试图使用我的 sqlalchemy 框架来运行复制命令，但我无法提交负载。此处的相关线程：stackoverflow.com/questions/28271049/… 不幸的是，使用 sqlalchemy 的自动提交解决方案解释了那里对我不起作用，但上面的 psycopg2 确实
@Voket 它被卡在执行上，然后无法运行 conn.commit()。有什么建议吗？

【解决方案3】：

首先，确保交易已提交。

conn = psycopg2.connect(conn_string)
cur = conn.cursor()
cur.execute(copy_cmd_str)
conn.commit()

您也可以通过以下方式确保事务提交（确保释放资源），

with psycopg2.connect(conn_string) as conn:
    with conn.cursor() as curs:
        curs.execute(copy_cmd_str)

当连接退出 with 块时，如果该块没有引发异常，则事务已提交。如果出现异常，事务将回滚。

其次，当要加载的数据需要很长时间并且超过 connect_timeout（并且无法提交）时，即使执行 commit 也无济于事。因此，当显式提交没有帮助时，请尝试增加超时。

【讨论】：

感谢@str，我为此苦苦挣扎，完全忘记检查提交！

【解决方案4】：

如果您使用的是 sqlalchemy，则复制命令不会自行自动提交。这对我有用：

from sqlalchemy import create_engine
eng = create_engine(...)
command = """
copy command here
"""
conn = eng.connect()
result = conn.execution_options(autocommit=True).execute(command)
result.close()

【讨论】：