Jaki jest najlepszy sposób na dzielenie dużych plików w Pythonie na potrzeby wieloprocesowości?

Question

Prowadzę wiele "zawstydzająco równoległych" projektów, które chciałabym zrównoważyć z modułem multiprocessing. Jednak często wymagają one czytania dużych plików (większych niż 2 GB), przetwarzania ich wiersz po wierszu, wykonywania podstawowych obliczeń, a następnie zapisywania wyników. Jaki jest najlepszy sposób na podzielenie pliku i przetworzenie go za pomocą modułu wieloprocesorowego Pythona? Czy należy używać Queue lub JoinableQueue w multiprocessing? A może sam moduł Queue? Lub, czy powinienem mapować plik iterowalny nad pulą procesów przy użyciu multiprocessing? Eksperymentowałem z tymi podejściami, ale koszty ogólne są ogromne w dystrybucji danych po linii. Oparłem się na lekkim projekcie filtrów rurowych, używając cat file | process1 --out-file out1 --num-processes 2 | process2 --out-file out2, który przekazuje pewien procent danych wejściowych pierwszego procesu bezpośrednio do drugiego wejścia (patrz this post), ale chciałbym mieć rozwiązanie zawarte w całości w Pythonie.

Co zaskakujące, dokumentacja w języku Python nie sugeruje kanonicznego sposobu wykonania tej czynności (pomimo obszernej sekcji dotyczącej wytycznych programowania w dokumentacji multiprocessing).

Dzięki, Vince

Informacje dodatkowe: Czas przetwarzania jednej linii zmienia. Niektóre problemy są szybkie i ledwo nie są związane z I/O, niektóre są związane z CPU. Zależne od CPU, niezależne zadania uzyskają pozycję zrównoleglania, tak że nawet nieefektywne sposoby przypisywania danych do funkcji przetwarzania będą nadal korzystne pod względem czasu zegara ściennego.

Dobrym przykładem jest skrypt, który wyodrębnia pola z linii, sprawdza różne flagi bitowe i zapisuje linie z pewnymi flagami do nowego pliku w całkowicie nowym formacie. To wydaje się być problemem związanym z wejściem/wyjściem, ale kiedy uruchomiłem go z moją tanią, współbieżną wersją z rurkami, było to około 20% szybsze. Po uruchomieniu go z pulą i mapą lub kolejką w trybie multiprocessing jest zawsze o ponad 100% wolniejszy.

Answer 1

Jedna z najlepszych architektur jest już częścią systemu operacyjnego Linux. Nie są wymagane żadne specjalne biblioteki.

Chcesz zaprojektować "fan-out".

1. Program "główny" tworzy wiele podprocesów połączonych rurami.

2. Główny program odczytuje plik, zapisując linie do rur, wykonując minimalne filtrowanie wymagane do obsłużenia linii odpowiednimi podprocesami.

Każdy podproces powinien być prawdopodobnie potokiem różnych procesów, które odczytują i zapisują ze standardowego wejścia.

Nie potrzebujesz struktury danych kolejki, to jest dokładnie to, czym jest potok w pamięci - kolejka bajtów między dwoma współbieżnymi procesami.

Answer 2

To zależy w dużym stopniu od formatu pliku.

Czy ma sens podzielenie go w dowolnym miejscu? Czy musisz podzielić go na nową linię? Czy musisz się upewnić, że podzielisz go na końcu definicji obiektu?

Zamiast dzielić plik, należy użyć wielu czytników tego samego pliku, używając os.lseek, aby przeskoczyć do odpowiedniej części pliku.

Aktualizacja: Dodano plakat, że chce podzielić na nowe linie. Następnie proponuję następujące:

Załóżmy, że masz 4 procesy. Wtedy prostym rozwiązaniem jest os.lseek do 0%, 25%, 50% i 75% pliku, a następnie odczytaj bajty, aż trafisz na pierwszą nową linię. To jest twój punkt wyjścia dla każdego procesu. Nie musisz dzielić pliku, aby to zrobić, po prostu szukaj odpowiedniej lokalizacji w dużym pliku w każdym procesie i zacznij czytać stamtąd.

Answer 3

Nie wspomnisz, w jaki sposób przetwarzasz linie; prawdopodobnie najważniejsza informacja.

Czy każda linia jest niezależna? Czy obliczenia zależą od jednego wiersza przed następnym? Czy muszą być przetwarzane w blokach? Jak długo trwa przetwarzanie dla każdej linii? Czy istnieje etap przetwarzania, który musi zawierać "wszystkie" dane na końcu? A może wyniki pośrednie zostaną wyrzucone i utrzymywana będzie tylko całkowita suma? Czy plik może być początkowo podzielony przez podzielenie rozmiaru pliku przez liczbę wątków? A może rośnie w miarę przetwarzania?

Jeśli linie są niezależne, a plik nie rośnie, jedyną potrzebną koordynacją jest wydzielenie "początkowych adresów" i "długości" każdemu z pracowników; mogą niezależnie otwierać i szukać w aktach, a następnie po prostu koordynować ich wyniki; być może przez czekanie, aż wyniki N powrócą do kolejki.

Jeśli linie nie są niezależne, odpowiedź będzie bardzo zależała od struktury pliku.

Answer 4

Wiem, że konkretnie zapytałeś o Python, ale zachęcam Cię do spojrzenia na Hadoop (http://hadoop.apache.org/): implementuje algorytm Map and Reduce, który został specjalnie zaprojektowany, aby rozwiązać ten problem.

Powodzenia

Answer 5

Jeżeli czas pracy jest długi, zamiast czytać każdy proces jego następną linię przez Queue mają procesy czytać partii linii. W ten sposób narzut jest amortyzowany w kilku liniach (na przykład w tysiącach lub więcej).

Answer 6

Jedną z strategii jest przypisanie każdemu pracownikowi przesunięcia, więc jeśli masz osiem procesów roboczych, które przypiszesz, to numery od 0 do 7. Pracownik numer 0 czyta pierwszy rekord, następnie pomija 7 i przechodzi do przetwarzania ósmego rekordu itp. , numer 1 pracownika odczytuje drugi rekord, a następnie pomija 7 i przetwarza 9. rekord ...

Istnieje wiele korzyści tego schematu. Nie ma znaczenia, jak duży jest plik, a praca jest zawsze dzielona równomiernie, procesy na tym samym komputerze będą przetwarzane z mniej więcej taką samą szybkością i będą używać tych samych obszarów bufora, aby nie ponosić nadmiernego obciążenia we/wy. Dopóki plik nie zostanie zaktualizowany, możesz ponownie uruchomić pojedyncze wątki w celu odzyskania po niepowodzeniach.

Answer 7

Fredrik Lundh to Some Notes on Tim Bray's Wide Finder Benchmark to interesująca lektura, o bardzo podobnym przypadku użycia, z wieloma dobrymi radami. Różni inni autorzy również zaimplementowali to samo, niektórzy są powiązani z tego artykułu, ale możesz spróbować google'ować do "python wide finder" lub czegoś, aby znaleźć coś więcej. (istniało również rozwiązanie oparte na module multiprocessing, ale to już nie jest dostępne)