如何将数据写入Redshift,这是在Python中创建的数据帧的结果？

Question

我有一个Python数据框.我可以将此数据作为新表写入Redshift吗？我已经成功创建了与Redshift的数据库连接,并且能够执行简单的SQL查询.现在我需要写一个数据帧.

Answer 1

您可以使用to_sql将数据推送到Redshift数据库.我已经能够通过SQLAlchemy引擎连接到我的数据库.一定要确定index = False你的to_sql电话.如果表不存在,将创建该表,并且您可以指定是否要调用替换表,附加到表,或者如果表已存在则失败.

from sqlalchemy import create_engine
import pandas as pd

conn = create_engine('postgresql://username:password@yoururl.com:5439/yourdatabase')

df = pd.DataFrame([{'A': 'foo', 'B': 'green', 'C': 11},{'A':'bar', 'B':'blue', 'C': 20}])

df.to_sql('your_table', conn, index=False, if_exists='replace')

请注意,您可能需要pip install psycopg2通过SQLAlchemy连接到Redshift.

to_sql文档

Answer 2

import pandas_redshift as pr

pr.connect_to_redshift(dbname = <dbname>,
                        host = <host>,
                        port = <port>,
                        user = <user>,
                        password = <password>)

pr.connect_to_s3(aws_access_key_id = <aws_access_key_id>,
                aws_secret_access_key = <aws_secret_access_key>,
                bucket = <bucket>,
                subdirectory = <subdirectory>)

# Write the DataFrame to S3 and then to redshift
pr.pandas_to_redshift(data_frame = data_frame,
                        redshift_table_name = 'gawronski.nba_shots_log')

详细信息：https : //github.com/agawronski/pandas_redshift

Answer 3

假设您可以访问 S3，这种方法应该有效：

第 1 步：将 DataFrame 作为 csv 写入 S3（为此我使用 AWS SDK boto3）
第 2 步：您知道 DataFrame 中 Redshift 表的列、数据类型和键/索引，因此您应该能够生成一个create table脚本并将其推送到 Redshift 以创建空表
第 3 步：copy从 Python 环境向 Redshift发送命令以将数据从 S3 复制到第 2 步中创建的空表

每次都像魅力一样工作。

第 4 步：在您的云存储人员开始对您大喊大叫之前，请从 S3 中删除 csv

如果您看到自己多次这样做，将所有四个步骤包装在一个函数中可以保持整洁。

Answer 4

我尝试使用熊猫，df.to_sql()但速度非常慢。我花了10多分钟才插入50行。看到这个未解决的问题（撰写本文时）

我尝试odo从炽热的生态系统中使用（根据问题讨论中的建议），但是ProgrammingError我没有去研究它。

最后有效的是：

import psycopg2

# Fill in the blanks for the conn object
conn = psycopg2.connect(user = 'user',
                              password = 'password',
                              host = 'host',
                              dbname = 'db',
                              port = 666)
cursor = conn.cursor()

args_str = b','.join(cursor.mogrify("(%s,%s,...)", x) for x in tuple(map(tuple,np_data)))
cursor.execute("insert into table (a,b,...) VALUES "+args_str.decode("utf-8"))

cursor.close()
conn.commit()
conn.close()

是的，很老psycopg2。这是用于一个numpy数组，但从a转换df为a ndarray应该不太困难。这使我大约每分钟3k行。

但是，根据其他队友的建议，最快的解决方案是在将数据帧作为TSV / CSV转储到S3群集中然后进行复制之后，使用COPY命令。如果要复制非常庞大的数据集，则应对此进行调查。（如果我尝试的话，我会在这里更新）

Answer 5

我曾经依赖pandasto_sql()函数，但它太慢了。我最近转向执行以下操作：

import pandas as pd
import s3fs # great module which allows you to read/write to s3 easily
import sqlalchemy

df = pd.DataFrame([{'A': 'foo', 'B': 'green', 'C': 11},{'A':'bar', 'B':'blue', 'C': 20}])

s3 = s3fs.S3FileSystem(anon=False)
filename = 'my_s3_bucket_name/file.csv'
with s3.open(filename, 'w') as f:
    df.to_csv(f, index=False, header=False)

con = sqlalchemy.create_engine('postgresql://username:password@yoururl.com:5439/yourdatabase')
# make sure the schema for mytable exists

# if you need to delete the table but not the schema leave DELETE mytable
# if you want to only append, I think just removing the DELETE mytable would work

con.execute("""
    DELETE mytable;
    COPY mytable
    from 's3://%s'
    iam_role 'arn:aws:iam::xxxx:role/role_name'
    csv;""" % filename)

该角色必须允许 redshift 访问 S3，请参阅此处了解更多详细信息

我发现对于 300KB 文件（12000x2 数据帧），这需要 4 秒，而我使用 pandasto_sql()函数需要 8 分钟