Python: параллельная пакетная запись Neo4J - PullRequest
Новая
       73

Python: параллельная пакетная запись Neo4J

0 голосов
/ 01 августа 2020

Я хочу отображать отношения слов в Neo4j 4.0 . Я написал парсер в Python 3.8 , который считывает 3.000 текстовых файлов и затем сохраняет информацию в Neo4J с помощью Neo4j Python Driver 4.1 . Весь процесс занимает 0,5–2 секунды на текстовый файл. В настоящее время я пытаюсь ускорить эти записи, выполняя их параллельно. После прочтения документации:

сеансы могут принимать только одну транзакцию за раз. Для параллельного выполнения следует использовать несколько сеансов. https://neo4j.com/docs/driver-manual/current/cypher-workflow/#driver -сессии

Я решил объединить обработку каждого файла / текста в одну транзакцию за один сеанс. Итак, для каждого файла, который я создаю, а затем закрываю новый сеанс: 

driver = GraphDatabase.driver('uri', auth=('user', 'password'), encrypted=False)

def feed_word_list(tx, word_list, title, headline):
    ...
    #example batch
    batch = {'batch':[{'from':'start','to':'end','from_tf':3,'to_tf':4}]}
    tx.run( 'UNWIND $batch as tuples '
            'MERGE (word1: Word {title: tuples.from}) '
            'MERGE (word2: Word {title: tuples.to}) '
            'MERGE (text1: Text {title: $title, headline: $headline}) '
            'MERGE (text1) -[:CONTAINS {term_frequency: tuples.from_tf}]-> (word1) '
            'MERGE (text1) -[:CONTAINS {term_frequency: tuples.to_tf}]-> (word2) '
            'MERGE (word1) -[:IS_NEIGHBORING]-> (word2) '
            , batch, title=title, headline=headline)

def parse_write(f):
    text = parse(f)
    with driver.session() as session:
        session.write_transaction(feed_word_list, text.text, text.identifier, text.headline)

#path is a directory of files
def scrape(path):
    #files is a list of files, which can processed independently 
    files = create_files_list(path)
    with Pool(2) as p:
        p.map(process_file, files)
    driver.close()

Это приводит к следующему исключению / ошибке: 

Transaction failed and will be retried in 1.0903621073189202s (LockClient[220] can't wait on resource RWLock[NODE(11398), hash=1487896506] since => LockClient[220] <-[:HELD_BY]- RWLock[NODE(10846), hash=1926906302] <-[:WAITING_FORsource RWLock[NODE(11398), hash=1487896506] since => LockClient[220] <-[:HELD_BY]- RWLock[NODE(10846), hash=1926906302] <-[:WAITING_FOR]- LockClient[222] <-[:HELD_BY]- RWLock[NODE(11398), hash=1487896506])
FIXME: should always disconnect before connect

Я предполагаю, что параллельных подключений слишком много. которые блокируют друг друга. Но насколько я понял: with Pool(2) as p: есть всего два параллельных исполнения. Итак, мой вопрос: как возможна параллельная запись с python и Neo4j?

...