Как использовать простые вызовы sqlalchemy при использовании потоков / многопроцессорности

Question

Задача

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

1. Основная процедура создает новую базу данных и настраивает несколько таблиц.
2. Основная подпрограмма устанавливает группу процессов / потоков, которые будут запускать рабочую функцию.
3. Основная процедура запускает все процессы.
4. Основная процедура читает корпус, добавляя документы в очередь.
5. Рабочая функция каждого процесса зацикливается, считывая документ из очереди, извлекая из него информацию, используя processdocument, и записывая информацию в новую запись в таблице в базе данных.
6. Рабочий цикл прерывается, когда очередь пуста, и основной подпрограммой установлен соответствующий флаг (если больше нет документов для добавления в очередь).

Вопрос

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

Ничего особенно сложного не происходит: каждый процесс получает уникальное значение, которое присваивается записи из объекта multiprocessing.Value, защищенного замком. Я просто не уверен, следует ли передать то, что я должен передать рабочей функции (кроме очереди), если что-нибудь. Я передаю экземпляр sqlalchemy.Engine, который я создал в основной процедуре? Экземпляр метаданных? Создаю ли я новый движок для каждого процесса? Есть ли какой-то другой канонический способ сделать это? Что-то особенное, что мне нужно иметь в виду?

Дополнительные комментарии

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

Заранее спасибо за помощь!

Answer 1

MetaData и его коллекция объектов Table следует рассматривать как фиксированную, неизменную структуру вашего приложения, в отличие от ваших функций и определений классов.Как известно, при разветвлении дочернего процесса все структуры уровня модуля вашего приложения остаются за границами процесса, а определения таблиц обычно относятся к этой категории.

Однако Engine относится к пулу соединений DBAPI.которые обычно являются соединениями TCP / IP и иногда файловыми дескрипторами.Сами соединения DBAPI, как правило, не переносимы через границу подпроцесса, поэтому вы можете либо создать новый движок для каждого подпроцесса, либо использовать движок без пула, что означает, что вы используете NullPool.

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

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