psycopg2/python 将数据从 postgresql 复制到 Amazon RedShift(postgresql)

Posted 2023-03-30

【中文标题】psycopg2/python 将数据从 postgresql 复制到 Amazon RedShift(postgresql)

【英文标题】：psycopg2/python copy data from postgresql to Amazon RedShift(postgresql)

【发布时间】：2015-05-06 08:28:26

【问题描述】：

我想了解社区对“使用 python 2.7.x 将数据从 PostgreSQL 复制到 RedShift 的最佳方式”的意见。我不能使用 Amazon S3，RedShift 是普通的 postgresql 数据库，但只支持从 S3 复制（我不能使用）

【问题讨论】：

... 你不能使用 S3 因为...？

我必须并行执行所有操作。不支持每次执行多个复制命令。无论如何，这主要是python知识问题。

Redshift docs 说他们支持 COPY “...来自 Amazon S3 上的文件、来自 DynamoDB 表或来自来自一个或多个远程主机的文本输出”所以您应该可以根据需要通过 python/psycopg2 加载 COPY 数据。

乔希，请看我之前的评论。从技术上讲这是可能的，但要求说 10 个并行操作比我使用复制是不可能的。

【参考方案1】：

您可以使用 Python/psycopg2/boto 对其进行端到端编码。 psycopg2 的替代方案是 PosgtreSQL 客户端 (psql.exe)。

如果您使用 psycopg2，您可以：

从 PostgreSQL 假脱机到文件 上传到 S3 附加到 Redshift 表。

如果你使用 psql.exe，你可以：

将数据从 PostgreSQL 管道传输到 S3 分段上传器

in_qry=open(opt.pgres_query_file, "r").read().strip().strip(';')
db_client_dbshell=r'%s\bin\psql.exe' % PGRES_CLIENT_HOME.strip('"')
loadConf=[ db_client_dbshell ,'-U', opt.pgres_user,'-d',opt.pgres_db_name, '-h', opt.pgres_db_server]

q="""
COPY ((%s) %s)
TO STDOUT
WITH DELIMITER ','
CSV %s
""" % (in_qry, limit, quote)
#print q
p1 = Popen(['echo', q], stdout=PIPE,stderr=PIPE,env=env)

p2 = Popen(loadConf, stdin=p1.stdout, stdout=PIPE,stderr=PIPE)

p1.wait()
return p2

上传到 S3。

使用 psycopg2 附加到 Redshift 表。

fn='s3://%s' % location
conn_string = REDSHIFT_CONNECT_STRING.strip().strip('"')    
con = psycopg2.connect(conn_string);
cur = con.cursor(); 
quote=''
if opt.red_quote:
    quote='quote \'%s\'' % opt.red_quote
ignoreheader =''
if opt.red_ignoreheader:
    ignoreheader='IGNOREHEADER %s' % opt.red_ignoreheader
timeformat=''
if opt.red_timeformat:
    #timeformat=" dateformat 'auto' "
    timeformat=" TIMEFORMAT '%s'" %     opt.red_timeformat.strip().strip("'")
sql="""
COPY %s FROM '%s' 
CREDENTIALS 'aws_access_key_id=%s;aws_secret_access_key=%s' 
DELIMITER '%s' 
FORMAT CSV %s 
GZIP 
%s 
%s; 
COMMIT;
""" % (opt.red_to_table, fn, AWS_ACCESS_KEY_ID, AWS_SECRET_ACCESS_KEY,opt.red_col_delim,quote, timeformat, ignoreheader)
cur.execute(sql)    
con.close()

我已尽力将所有 3 个步骤编译成 one script。

【讨论】：

感谢您的回答，但您的脚本链接已损坏。可以转发一下吗？