带有 postgres 的 Python 使用命名变量和批量插入

Question

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

• Windows 7 64 位
• python 3.2
• PostgreSQL 9.1
• 心理战2

这是我的场景:我正在将一个数据库(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

如有任何帮助、建议或评论，我们将不胜感激。谢谢，米奇

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]

这将返回生成的序列号。展望 future ，我强烈建议您摆脱目前正在尝试抽象数据库通信的模式(您的 sqlDict 方法)并采用非常直接的编码模式(聪明是您的敌人，它会提高性能更努力地调整)。

您需要将插入批处理成适合性能的 block 大小。这意味着根据您的实际行为调整 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()

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