Я работаю над проектом, который требует быстрого воспроизведения большого количества информации в Neo4J через py2neo, и я столкнулся с проблемой при попытке использовать многопроцессорную библиотеку Python для ускорения процесса.
В общих чертах, то, что мы делаем, состоит из двух частей, где обе части происходят в унисон:
Часть 1:
- Создание отношений между парой узлов (A-> B)
- Создание другого узла, который имеет одно отношение к каждой из предыдущей пары (C -> A, C -> B)
- Pu sh все 3 узла и их связи в Neo4j,
- Pu sh идентификатор узла C к топике кафки c
Часть 2:
- Создание N "потоков" с помощью многопроцессорной обработки python, все из которых открывают своего собственного потребителя kafka
- потребители читают topi c и выполняют код, который может обновлять отношения / узлы или создайте дополнительные.
Возникающая проблема возникает во второй части, особенно при обновлении или создании связей или узлов. py2neo выдает py2neo.database.TransientError, что в соответствии с документацией:
База данных не может обработать запрос прямо сейчас, повторная попытка позже может привести к успешному результату.
Я попытался добавить проверку безопасности, которая будет повторять транзакцию X раз, и повторно отправлять sh ее в kafka topi c, чтобы другой поток мог попытаться выполнить ее после неудачных попыток, но конечные результаты все еще неверны .
Я ужасно неосведомлен, когда дело доходит до многопоточности / многопроцессорности, поэтому я чувствую, что мои первоначальные исследовательские усилия были совершенно неосновными, и я потратил много циклов, пытаясь это исправить - - Любое направление или понимание очень ценится.
Может предоставить дополнительную информацию, если это необходимо - не уверен, что мне не хватает.
Спасибо