безопасное хранение многопоточных веб-скребковых данных

Question

Я использую селен для создания веб-соскабливания, а затем пытается сохранить данные в CSV-файле. Я использую очередь с рабочими, чтобы ускорить работу скребков.

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

Есть ли безопасный способ заставить нескольких работников писать в CSV (или какой-либо другой тип файла) одновременно?

Вот мой рабочий и очереди:

def worker(): 
while True: 
    params = q.get() 
    crawl(*params) 
    q.task_done() 

q = Queue() 

for i in range(7): 
    t = Thread(target=worker) 
    t.daemon = True 
    t.start() 

каждый раз, когда «ползти» называется, работник записывает строку в файл CSV, а затем закрывает CSV:

data_fd = open('data.csv','a') 
data_fd.write(line) 
data_fd.close() 

Однако, иногда происходит следующее, когда работник пытается записать в CSV до того, как другой работник закончил запись в CSV:

item_1, item_2, item_3, item_4 
item_1, item_2, item_3, 
item_1, item_2, item_3, item_4 
item_4 

Answer 1

Вы можете использовать два разных подхода ...

1) Вы «заблокировали» файл как-то - не обязательно на уровне ОС (вы можете просто получить монитор, который вы приобрели, прежде чем записывать в файл и выпускать в конце, так что в любой момент времени только один поток удерживает монитор и записывает в файл).

2) Вы можете разделить отдельный отдельный поток, посвященный записи данных в csv. Ваши искатели проскальзывают данные и добавят их в очередь. Поток csv продолжает считывать очередь и записывать данные на диск.

Оба подхода имеют свои достоинства - первый из них, вероятно, менее эффективен, но с ним легче справиться. Второй подход потребует некоторой мысли о том, насколько большой может увеличиться очередь (в случае, если запись csv будет медленнее, чем очистка), а также убедитесь, что очередь вышла из системы до того, как вы покинули приложение :)