将 pandas 数据帧批量插入 PostgreSQL 表的最佳方法答案

【问题标题】：Optimal approach to bulk insert of pandas dataframe into PostgreSQL table将 pandas 数据帧批量插入 PostgreSQL 表的最佳方法
【发布时间】：2018-07-22 23:47:39
【问题描述】：

我需要将多个 excel 文件上传到 postgresql 表，但它们可以在多个寄存器中相互重叠，因此我需要注意 IntegrityErrors。我遵循两种方法：

cursor.copy_from：最快的方法，但由于寄存器重复，我不知道如何捕获和控制所有Integrityerrors

streamCSV = StringIO()
streamCSV.write(invoicing_info.to_csv(index=None, header=None, sep=';')) 
streamCSV.seek(0)  

with conn.cursor() as c:
    c.copy_from(streamCSV, "staging.table_name", columns=dataframe.columns, sep=';')
    conn.commit()

cursor.execute：我可以统计和处理每个异常，但它非常慢。

data = invoicing_info.to_dict(orient='records')

with cursor as c:
    for entry in data:
        try:
            c.execute(DLL_INSERT, entry)
            successful_inserts += 1
            connection.commit()
            print('Successful insert. Operation number {}'.format(successful_inserts))
        except psycopg2.IntegrityError as duplicate:
            duplicate_registers += 1
            connection.rollback()
            print('Duplicate entry. Operation number {}'.format(duplicate_registers))

在例程结束时，我需要确定以下信息：

print("Initial shape: {}".format(invoicing_info.shape))
print("Successful inserts: {}".format(successful_inserts))
print("Duplicate entries: {}".format(duplicate_registers))

如何修改第一种方法来控制所有异常？如何优化第二种方法？

【问题讨论】：

标签： postgresql python-3.6 psycopg2

【解决方案1】：

虽然您在不同的 Excel 工作表中有重复的 ID，但您必须自己回答如何决定要信任哪个 Excel 工作表中的数据？

当您使用多个表时，将使用方法从冲突对中至少有一行，您始终可以执行以下操作：

为每个 Excel 工作表创建临时表
将数据上传到 Excel 工作表的每个表格（就像您现在批量执行的操作一样）
通过选择不同的 on(id) 进行插入，方式如下：

INSERT INTO staging.table_name(id, col1, col2 ...)
SELECT DISTINCT ON(id) 
     id, col1, col2
FROM 
(
    SELECT id, col1, col2 ... 
       FROM staging.temp_table_for_excel_sheet1
    UNION
    SELECT id, col1, col2 ... 
       FROM staging.temp_table_for_excel_sheet2
    UNION
    SELECT id, col1, col2 ... 
       FROM staging.temp_table_for_excel_sheet3
) as data

使用这种插入 postgreSQL 将从非唯一 id 集中取出随机行。

如果您想信任第一条记录，您可以添加一些订单：

INSERT INTO staging.table_name(id, col1, col2 ...)
SELECT DISTINCT ON(id) 
     id, ordering_column col1, col2
FROM 
(
    SELECT id, 1 as ordering_column, col1, col2 ... 
       FROM staging.temp_table_for_excel_sheet1
    UNION
    SELECT id, 2 as ordering_column, col1, col2 ... 
       FROM staging.temp_table_for_excel_sheet2
    UNION
    SELECT id, 3 as ordering_column, col1, col2 ... 
       FROM staging.temp_table_for_excel_sheet3
) as data
ORDER BY ordering_column

对象的初始计数：

SELECT sum(count)
FROM 
( 
  SELECT count(*) as count FROM temp_table_for_excel_sheet1
  UNION
  SELECT count(*) as count FROM temp_table_for_excel_sheet2
  UNION
  SELECT count(*) as count FROM temp_table_for_excel_sheet3
) as data

完成此批量插入后，您可以运行 select count(*) FROM staging.table_name 以获取插入记录总数的结果

对于重复计数，您可以运行：

SELECT sum(count)
FROM 
(
SELECT count(*) as count 
FROM  temp_table_for_excel_sheet2 WHERE id in (select id FROM temp_table_for_excel_sheet1 )

UNION

SELECT count(*) as count 
FROM  temp_table_for_excel_sheet3 WHERE id in (select id FROM temp_table_for_excel_sheet1 )
)

UNION

SELECT count(*) as count 
FROM  temp_table_for_excel_sheet3 WHERE id in (select id FROM temp_table_for_excel_sheet2 )
) as data

【讨论】：

【解决方案2】：

如果 Excel 表包含重复记录，Pandas 似乎是识别和消除欺骗的一个可能选择：https://33sticks.com/python-for-business-identifying-duplicate-data/。还是不同工作表中的不同记录具有相同的 id/index 的问题？如果是这样，您可以使用类似的方法使用 Pandas 来隔离多次使用的 id，然后在尝试上传到 SQL 数据库之前使用唯一标识符更正它们。

对于批量上传，我会使用 ORM。 SQLAlchemy 有一些关于批量上传的重要信息：http://docs.sqlalchemy.org/en/rel_1_0/orm/persistence_techniques.html#bulk-operations，这里有一个相关的讨论：Bulk insert with SQLAlchemy ORM

【讨论】：