回答问题

我必须从几个不同的数据库引擎中提取数据。导出此数据后,我将数据发送到 AWS S3 并使用 COPY 命令将该数据复制到 Redshift。一些表格包含大量文本,在列字段中存在换行符和其他字符。当我运行以下代码时:

cursor.execute('''SELECT * FROM some_schema.some_message_log''')
rows = cursor.fetchall()
with open('data.csv', 'w', newline='') as fp:
    a = csv.writer(fp, delimiter='|', quoting=csv.QUOTE_ALL, quotechar='"', doublequote=True, lineterminator='\n')
    a.writerows(rows)

一些具有回车/换行符的列将创建新行:

"2017-01-05 17:06:32.802700"|"SampleJob"|""|"Date"|"error"|"Job.py"|"syntax error at or near ""from"" LINE 34: select *, SYSDATE, from staging_tops.tkabsences;
                                      ^
-<class 'psycopg2.ProgrammingError'>"

这会导致导入过程失败。我可以通过对异常进行硬编码来解决这个问题:

cursor.execute('''SELECT * FROM some_schema.some_message_log''')
rows = cursor.fetchall()
with open('data.csv', 'w', newline='') as fp:
    a = csv.writer(fp, delimiter='|', quoting=csv.QUOTE_ALL, quotechar='"', doublequote=True, lineterminator='\n')

for row in rows:
    list_of_rows = []
    for c in row:
        if isinstance(c, str):
            c = c.replace("\n", "\\n")
            c = c.replace("|", "\|")
            c = c.replace("\\", "\\\\")
            list_of_rows.append(c)
        else:
            list_of_rows.append(c)
    a.writerow([x.encode('utf-8') if isinstance(x, str) else x for x in list_of_rows])

但这需要很长时间来处理较大的文件,而且一般来说似乎是不好的做法。是否有更快的方法将数据从 SQL 游标导出到 CSV,当遇到包含回车/换行符的文本列时不会中断?

Answers

我怀疑这个问题就像确保 Python CSV 导出库和 Redshift 的 COPY 导入使用一个通用接口一样简单。简而言之,检查您的分隔符和引用字符,并确保 Python 输出和 Redshift COPY 命令一致。

稍微详细一点:DB 驱动程序已经完成了以易于理解的形式进入 Python 的艰苦工作。也就是说,数据库中的每一行都是一个列表(或元组、生成器等),并且每个单元格都可以单独访问。当你有一个类似列表的结构时,Python 的 CSV 导出器可以完成剩下的工作,而且——至关重要的是——Redshift 将能够从输出、嵌入的换行符和所有内容中复制。 特别是,您不需要进行任何手动转义;.writerow()或.writerows()函数应该是您需要做的所有事情。

Redshift 的 COPY 实现默认理解 CSV 最常见的方言,即

• 用逗号分隔单元格 (,),

• 用双引号引用单元格("),

• 并通过加倍 ("→"") 转义任何嵌入的双引号。

要使用来自RedshiftFORMAT AS CSV的文档来支持这一点:

... 默认引号字符是双引号 ( " )。当在字段中使用引号字符时,请使用附加引号字符转义该字符。 ...

但是,您的 Python CSV 导出代码使用管道 (|) 作为delimiter并将quotechar设置为双引号 (")。这也可以工作,但为什么偏离默认值?建议使用 CSV 的同名文件并在此过程中保持代码更简单:

cursor.execute('''SELECT * FROM some_schema.some_message_log''')
rows = cursor.fetchall()
with open('data.csv', 'w') as fp:
    csvw = csv.writer( fp )
    csvw.writerows(rows)

从那里,告诉 COPY 使用 CSV 格式(同样不需要非默认规范):

COPY  your_table  FROM  your_csv_file  auth_code  FORMAT AS CSV;

那应该这样做。