Python with postgres 使用命名变量和批量插入

Question

我需要一些帮助来了解 Python 和 postgres 在单个事务中插入多个数据集时如何处理事务和批量插入。 环境：

• Windows 7 64 位
• Python 3.2
• Postgresql 9.1
• psycopg2

这是我的场景： 我正在将来自一个数据库（oracle）的数据转换为 xml 字符串并将该数据插入到一个新的数据库（postgres）中。这是一个大型数据集，所以我正在尝试优化我的一些插入。很多这些数据我正在考虑库类型对象，所以我有一个库表，然后是我的 xml 元数据和 xml 内容的表，这些数据的字段是数据库中的文本类型。我从 oracle 中提取数据，然后创建需要插入的数据的字典。我有 3 个插入语句，第一个插入使用序列 id 在库表中创建一条记录，并且该 id 对于接下来将 xml 插入元数据和内容表的两个查询中的关系是必需的。这是我所说的一个例子：

for inputKey in libDataDict.keys():
  metaString = libDataDict[inputKey][0]
  contentString = libDataDict[inputKey][1]
  insertLibDataList.append({'objIdent':"%s" % inputKey, 'objName':"%s" % inputKey, objType':libType})
  insertMetadataDataList.append({'objIdent':inputKey,'objMetadata':metaString}) 
  insertContentDataList.append({'objIdent':inputKey, 'objContent':contentString})

dataDict['cmsLibInsert'] = insertLibDataList
dataDict['cmsLibMetadataInsert'] = insertMetadataDataList
dataDict['cmsLibContentInsert'] = insertContentDataList

sqlDict[0] = {'sqlString':"insert into cms_libraries (cms_library_ident, cms_library_name, cms_library_type_id, cms_library_status_id) \
              values (%(objIdent)s, %(objName)s, (select id from cms_library_types where cms_library_type_name = %(objType)s), \
              (select id from cms_library_status where cms_library_status_name = 'active'))", 'data':dataDict['cmsLibInsert']}

sqlDict[1] = {'sqlString':"insert into cms_library_metadata (cms_library_id, cms_library_metadata_data) values \
              ((select id from cms_libraries where cms_library_ident = %(objIdent)s), $$%(objMetadata)s$$)", \
              'data':dataDict['cmsLibMetadataInsert']}

sqlDict[2] = {'sqlString':"insert into cms_library_content (cms_library_id, cms_library_content_data) values \
              ((select id from cms_libraries where cms_library_ident = %(objIdent)s), $$%(objContent)s$$)", \
              'data':dataDict['cmsLibContentInsert']}

bulkLoadData(myConfig['pgConn'], myConfig['pgCursor'], sqlDict)

我遇到的问题是，当我运行第一个查询 (sqlDict[0]) 并执行插入时，只要我在运行接下来的两个查询之前将其分开并提交，一切都可以正常工作。理想情况下，我希望所有这些查询都在同一个事务中，但它失败了，因为它无法从 cms_libraries 表中找到第二个和第三个查询的 id。 这是我当前的插入代码：

def bulkLoadData(dbConn, dbCursor, sqlDict):
 try:
   libInsertSql = sqlDict.pop(0)
   dbSql = libInsertSql['sqlString']
   data = libInsertSql['data']
   dbCursor.executemany(dbSql, data)
   dbConn.commit()
   for sqlKey in sqlDict:
     dbSql = sqlDict[sqlKey]['sqlString']
     data = sqlDict[sqlKey]['data']
     dbCursor.executemany(dbSql, data)

   dbConn.commit()

以前我将值附加到查询中，然后为每个插入运行查询。当我这样做时，我可以将它们全部放在同一个事务中，它会找到生成的 id，一切都很好。我不明白为什么当我使用 executemany() 进行批量插入时它找不到 id？有没有办法在同一个事务中进行批量插入和其他两个查询？

我一直在阅读此文档并搜索 stackoverflow 和互联网，但没有找到我的问题的答案： pyscopg docs 以及postgres的： Postgresql string docs

任何帮助、建议或 cmets 将不胜感激。 谢谢， 米奇

Answer 1

这里有两个选择。要么在外部生成 ID（这允许您进行批量插入），要么从串行生成它们（这意味着您必须进行单项插入）。我认为弄清楚外部 ID 生成和批量加载非常简单（尽管我建议你看一下 ETL tool 而不是在 python 中手动编码）。如果需要从串口中提取ID，则应考虑服务端prepared statements。

您的第一条语句应如下所示：

dbCursor.execute("""
PREPARE cms_lib_insert (bigint, text, text) AS 
INSERT INTO cms_libraries (cms_library_ident, cms_library_name, cms_library_type_id, cms_library_status_id)
VALUES ($1, $2,
    (select id from cms_library_types where cms_library_type_name = $3), 
    (select id from cms_library_status where cms_library_status_name = 'active')
)
RETURNING cms_library.id
""")

您将在启动时运行一次。然后，您需要在每个条目级别上运行以下 EXECUTE 语句。

dbCursor.execute("""
EXECUTE cms_lib_insert(%(objIndent)s, %(objName)s, %(objType)s)
""", {'objIndent': 345, 'objName': 'foo', 'objType': 'bar'))
my_new_id = dbCursor.fetchone()[0]

这将返回生成的序列号。展望未来，我强烈建议您摆脱当前尝试抽象数据库通信的模式（您的 sqlDict 方法）并使用非常直接的编码模式（聪明是您的敌人，它可以提高性能更努力地调整）。

您需要将插入批处理成一个适合性能的块大小。这意味着根据您的实际行为调整您的 BLOCK_SIZE。您的代码应如下所示：

BLOCK_SIZE = 500
while not_done:
   dbCursor.begin()
   for junk in irange(BLOCK_SIZE):
       dbCursor.execute("EXECUTE cms_lib_insert(...)")
       cms_lib_id = dbCursor.fetchone()[0]     # you're using this below.
       dbCursor.executemany("EXECUTE metadata_insert(...)")
       dbCursor.executemany("EXECUTE library_insert(...)")
   dbCursor.commit()

如果您需要达到比这更高的性能水平，下一步是构建一个插入处理函数，该函数为相关表获取行数组。我不建议这样做，因为它很快就会成为维护的噩梦。