Блог

Мне нужна помощь в понимании того, как Python и postgres обрабатывают транзакции и массовые вставки, особенно при вставке нескольких наборов данных в одну транзакцию. Окружающая среда:

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

Вот мой сценарий: я конвертирую данные из одной базы данных (oracle) в строки xml и вставляю эти данные в новую базу данных (postgres). Это большой набор данных, поэтому я пытаюсь оптимизировать некоторые из моих вставок. Большую часть этих данных я рассматриваю как объекты библиотечного типа, поэтому у меня есть таблица библиотеки, а затем таблицы для моих метаданных xml и содержимого xml, поля для этих данных представляют собой текстовые типы в базе данных. Я извлекаю данные из oracle, а затем создаю словари данных, которые мне нужно вставить. У меня есть 3 оператора insert, первая вставка создает запись в таблице библиотеки с использованием последовательного идентификатора, и этот идентификатор необходим для связи в следующих двух запросах, которые вставляют 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 для 2-го и 3-го запросов.
Вот мой текущий код вставки:

 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()
  

Ранее я добавлял значения в запрос, а затем выполнял запрос для каждой вставки. Когда я это делаю, я могу поместить все это в одну транзакцию, и она находит сгенерированный идентификатор, и все в порядке. Я не понимаю, почему он не находит идентификатор, когда я выполняю массовую вставку с помощью executemany()? Есть ли способ выполнить массовую вставку и два других запроса в одной транзакции?

Я читал эту документацию и искал stackoverflow и Интернет, но не нашел ответа на свою проблему: документы pyscopg, а также документы postgres: Postgresql string docs

Здесь у вас есть два варианта. Либо сгенерируйте идентификаторы извне (что позволяет выполнять массовые вставки), либо сгенерируйте их из serial (что означает, что вам нужно выполнять вставки с одной записью). Я думаю, что довольно просто разобраться с генерацией внешнего идентификатора и массовой загрузкой (хотя я бы рекомендовал вам взглянуть на инструмент ETL, а не вручную что-то кодировать на python). Если вам нужно извлекать идентификаторы из последовательного файла, вам следует рассмотреть подготовленные инструкции на стороне сервера.

Ваш первый оператор должен выглядеть следующим образом:

 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()
  

Если вам нужно достичь уровня производительности выше этого, следующим шагом будет создание функции-обработчика вставки, которая принимает массивы строк для зависимых таблиц. Я не рекомендую делать это, так как это быстро становится кошмаром для обслуживания.