Algorytm Parallel top ten dla danych rozproszonych

Question

Algorytm Parallel top ten dla danych rozproszonych

To jest pytanie z wywiadu. Załóżmy, że jest kilka komputerów i każdy komputer przechowuje bardzo duży plik dziennika odwiedzanych adresów URL. Znajdź top ten najczęściej odwiedzanych adresów URL.

Na przykład: załóżmy, że są tylko 3 Komputery i potrzebujemy dwóch najlepszych najczęściej odwiedzanych adresów URL.

Computer A: url1, url2, url1, url3
Computer B: url4, url2, url1, url1
Computer C: url3, url4, url1, url3

url1 appears 5 times in all logs
url2 2
url3 3
url4 2 

So the answer is url1, url3

Pliki dziennika są zbyt duże, aby zmieścić się w pamięci RAM i skopiować je przez sieć. Jak rozumiem, ważne jest również, aby obliczenia były równoległe i wykorzystywały wszystkie dane Komputery.

Jak rozwiążesz to?

19

language-agnostic algorithm parallel-processing distributed-computing

Author: Michael, 2013-03-25

Source

5 answers

Biorąc pod uwagę skalę plików dziennika i ogólny charakter pytania, jest to dość trudny problem do rozwiązania. Nie sądzę, że istnieje jeden najlepszy algorytm dla wszystkich sytuacji. To zależy od charakteru zawartości plików dziennika. Weźmy na przykład przypadek narożny, że wszystkie adresy URL są unikalne we wszystkich plikach dziennika. W takim przypadku, w zasadzie każde rozwiązanie zajmie dużo czasu, aby wyciągnąć ten wniosek (jeśli nawet zajdzie tak daleko...), a na twoje pytanie nie ma nawet odpowiedzi bo nie ma pierwszej dziesiątki.

Nie mam wodoszczelnego algorytmu, który mogę przedstawić, ale chciałbym zbadać rozwiązanie, które wykorzystuje histogramy wartości skrótu adresów URL w przeciwieństwie do samych adresów URL. Te histogramy mogą być obliczane za pomocą odczytu jednego pliku, dzięki czemu może radzić sobie z plikami dziennika o dowolnym rozmiarze. W pseudo-kodzie wybrałbym coś takiego:

Użyj funkcji skrótu z ograniczoną przestrzenią docelową (powiedzmy 10 000, zauważ, że kolidujące wartości skrótu są oczekiwane), aby obliczyć wartość hash każdej pozycji w pliku dziennika i policzyć, ile razy każda z wartości has występuje. Przekazanie wynikowego histogramu serwerowi (chociaż prawdopodobnie możliwe jest również uniknięcie centralnego serwera przez multicasting wyniku do każdego innego węzła-ale zostanę przy bardziej oczywistym podejściu do serwera)
serwer powinien połączyć histogramy i przekazać wynik z powrotem. W zależności od dystrybucji adresów URL, może być liczba widocznych już szczytów, zawierających odwiedzane adresy URL.
każdy z węzłów powinien skupić się na szczytach histogramu. Powinien ponownie przejść przez plik dziennika, użyć dodatkowej funkcji skrótu (ponownie z ograniczoną przestrzenią docelową), aby obliczyć nowy hash-histogram dla tych adresów URL, które mają swoją pierwszą wartość skrótu w jednym ze szczytów (liczba szczytów, na których należy się skupić, byłaby parametrem do dostrojenia w algorytmie, w zależności od rozkładu adresów URL), i Oblicz drugi histogram z nowymi wartościami skrótu. Wynik powinien zostać przekazany do serwera.
serwer powinien ponownie połączyć wyniki i przeanalizować nowy histogram z oryginalnym histogramem. W zależności od wyraźnie widocznych szczytów, może być w stanie wyciągnąć wnioski na temat dwóch wartości skrótu z pierwszej dziesiątki adresów URL. Lub może to poinstruować maszyny, aby obliczyły więcej wartości skrótu za pomocą drugiej funkcji skrótu, a prawdopodobnie po tym przejść przez trzeci przebieg hash-obliczenia z jeszcze jedną funkcją hash. Musi to być kontynuowane, dopóki nie można wyciągnąć wniosku ze zbiorczej grupy histogramów, jakie są wartości hash szczytowych adresów URL, a następnie węzły mogą zidentyfikować różne adresy URL z tego.

Zauważ, że ten mechanizm będzie wymagał strojenia i optymalizacji w odniesieniu do kilku aspektów algorytmu i funkcji skrótu. Będzie również wymagać orkiestracji przez serwer co do tego, które obliczenia należy wykonać w dowolnym momencie. Informatyka prawdopodobnie będzie również musiał ustawić pewne granice w celu zawarcia, gdy nie można wyciągnąć żadnych wniosków, innymi słowy, gdy "widmo" wartości skrótu URL jest zbyt płaski, aby warto było kontynuować obliczenia.

To podejście powinno działać dobrze, jeśli istnieje wyraźna dystrybucja w adresach URL. Podejrzewam, że praktycznie rzecz biorąc, pytanie ma sens tylko w tym przypadku.

2

Author: Reinier Torenbeek,
Warning: date(): Invalid date.timezone value 'Europe/Kyiv', we selected the timezone 'UTC' for now. in /var/www/agent_stack/data/www/doraprojects.net/template/agent.layouts/content.php on line 54
2013-03-25 16:34:00

Zakładając, że poniższe warunki są prawdziwe:

potrzebujesz najlepszych adresów URL N hostów m.
nie można przechowywać plików w pamięci RAM
istnieje węzeł główny

Przyjąłbym podejście poniżej:

Każdy węzeł odczytuje część pliku (tj. MAX URL, gdzie MAX może być, powiedzmy, 1000 URL) i utrzymuje tablicę arr [MAX]={url, hits}.

Gdy węzeł odczytuje maks. adresy URL z pliku, wysyła listę do węzła głównego i uruchamia ponownie odczyt, aż maks. adresy URL / align = "left" /

Gdy węzeł osiągnie EOF, wysyła pozostałą listę adresów URL i flagę EOF do węzła głównego.

Gdy węzeł główny otrzymuje listę adresów URL, porównuje ją z ostatnią listą adresów URL i generuje nową, zaktualizowaną.

Gdy węzeł główny otrzymuje flagę EOF z każdego węzła i kończy czytanie własnego pliku, szukamy adresów URL z ostatniej wersji jego listy.

~~lub~~

Inny podejście, które uwolniłoby mistrza od wykonywania całej pracy może być:

Każdy węzeł odczytuje swój plik i przechowuje tablicę taką samą jak powyżej, czytając do EOF.

Gdy EOF, węzeł wyśle pierwszy adres url listy i liczbę trafień do master.

Gdy master zebrał pierwszy adres url i liczbę trafień dla każdego węzła, generuje listę. Jeśli węzeł główny ma mniej niż n adresów URL, poprosi węzły o wysłanie drugiego i tak dalej. Dopóki pan nie adresy URL posortowane.

1

Author: ophintor,
Warning: date(): Invalid date.timezone value 'Europe/Kyiv', we selected the timezone 'UTC' for now. in /var/www/agent_stack/data/www/doraprojects.net/template/agent.layouts/content.php on line 54
2013-03-25 14:01:16

Wstępne przetwarzanie: każdy system komputerowy przetwarza kompletny plik dziennika i przygotowuje unikalną listę adresów URL z liczeniem na nich.

Pobieranie najlepszych adresów URL:

oblicz liczbę adresów URL w każdym systemie komputerowym
Proces zestawiania w systemie centralnym (wirtualnym)
- wysyłanie adresów URL z licznikiem do centralnej jednostki przetwarzającej jeden po drugim w kolejności DESC (tj. od górnej większości)
- w systemie centralnym zestawiaj przychodzące szczegóły URL
- powtarzaj aż suma wszystkich zliczeń z przychodzących Adresy URL są mniejsze niż liczba dziesiątych adresów URL na liście głównej. ważny krok, aby być absolutnie pewnym

PS: będziesz miał dziesięć najlepszych adresów URL w różnych systemach niekoniecznie w tej kolejności. Aby uzyskać rzeczywistą kolejność można odwrotne zestawienie. Dla danego adresu URL w pierwszej dziesiątce uzyskaj Indywidualne liczenie z dist-computers i formuj ostateczną kolejność.

1

Author: SparKot,
Warning: date(): Invalid date.timezone value 'Europe/Kyiv', we selected the timezone 'UTC' for now. in /var/www/agent_stack/data/www/doraprojects.net/template/agent.layouts/content.php on line 54
2013-03-25 14:01:32

Poniższy opis to pomysł na rozwiązanie. nie jest to pseudokod.
Załóżmy, że masz zbiór systemów.
1.dla każdego A: Collections (systems)
1.1) Uruchom demona na każdym komputerze, który sprawdza w pliku dziennika zmiany.
1.2) gdy zmiana zostanie zauważona, wakeup AnalyzerThreadA
1.3) Jeśli AnalyzerThreadA znajdzie adres URL używając regex, zaktualizuj localHashMapA za pomocą count++.
(key = URL, value = count).
2) Push TOPTEN wpisy z localHashMapA do ComputerA gdzie będzie uruchomiony Demon Analytizeall.

Powyższy krok będzie ostatnim krokiem w każdym systemie, który wypchnie wpisy topTen do głównego systemu, na przykład: computerA.

3) AnalyzeAll uruchomiony w computerA rozwiąże duplikaty i zaktualizuje liczbę w masterHashMap adresów URL.

4) Wydrukuj topTen z masterHashMap.

0

Author: a3.14_Infinity,
Warning: date(): Invalid date.timezone value 'Europe/Kyiv', we selected the timezone 'UTC' for now. in /var/www/agent_stack/data/www/doraprojects.net/template/agent.layouts/content.php on line 54
2013-03-25 12:31:28

score 16 · Accepted Answer

Jest to dość standardowy problem, dla którego istnieje dobrze znane rozwiązanie. Po prostu sortujesz pliki dziennika na każdym komputerze według adresu URL, a następnie łączysz je przez kolejkę priorytetową o rozmiarze k (Liczba żądanych pozycji) na komputerze "master". Technika ta istnieje od 1960 roku i jest nadal używana (choć nieco zmodyfikowana) w formie MapReduce.

Na każdym komputerze wyodrębnij adres URL i liczbę z pliku dziennika i posortuj według adresu URL. Ponieważ log pliki są większe niż zmieści się w pamięci, musisz wykonać scalenie na dysku. Wiąże się to z odczytaniem fragmentu pliku dziennika, sortowaniem według adresu URL, zapisem fragmentu na dysk. Odczyt kolejnego fragmentu, sortowanie, zapis na dysk itp. W pewnym momencie masz kawałki pliku dziennika m, każdy posortowany. Następnie można wykonać m-way merge. Ale zamiast zapisywać elementy na dysk, prezentujesz je, w porządku posortowanym (posortowanym według adresu URL, czyli), do "master".

Każda maszyna sortuje swój własny dziennik.

"mistrz" komputer łączy dane z oddzielnych komputerów i dokonuje wyboru top K. W rzeczywistości są to dwa problemy, ale można je połączyć w jeden.

Master tworzy dwie kolejki priorytetowe: jedną dla połączenia i jedną dla topowego wyboru K. Pierwszy ma rozmiar N, gdzie N to liczba komputerów, z których łączy się dane. Drugi ma rozmiar K: liczbę elementów, które chcesz wybrać. Używam do tego sterty min, ponieważ jest to łatwe i w miarę szybkie.

Aby ustawić kolejkę scalania, zainicjalizuj kolejkę i pobieraj pierwszą pozycję z każdego z komputerów "worker". W poniższym pseudo-kodzie "get lowest item from merge queue" oznacza pobranie głównego item z kolejki merge, a następnie pobranie następnego item z dowolnego komputera, który go zaprezentował. Jeśli więc Kolejka zawiera [1, 2, 3], A pozycje pochodziły z komputerów B, C, A (w tej kolejności), to wzięcie najniższego elementu oznaczałoby pobranie następnego elementu z komputera B i dodanie go do kolejki priorytetów.

The następnie mistrz wykonuje następujące czynności:

working = get lowest item from merge queue
while (items left to merge)
{
    temp = get lowest item from merge queue
    while (temp.url == working.url)
    {
        working.count += temp.count
        temp = get lowest item from merge queue
    }
    // Now have merged counts for one url.
    if (topK.Count < desired_count)
    {
        // topK queue doesn't have enough items yet.
        // so add this one.
        topK.Add(working);
    }
    else if (topK.Peek().count < working.count)
    {
        // the count for this url is larger
        // than the smallest item on the heap
        // replace smallest on the heap with this one
        topK.RemoveRoot()
        topK.Add(working)
    }
    working = temp;
}
// Here you need to check the last item:
if (topK.Peek().count < working.count)
{
    // the count for this url is larger
    // than the smallest item on the heap
    // replace smallest on the heap with this one
    topK.RemoveRoot()
    topK.Add(working)
}

W tym momencie Kolejka topK ma elementy K z najwyższą liczbą.

Więc każdy komputer musi wykonać sortowanie scalające, które jest O (n log n), gdzie n jest liczbą pozycji w dzienniku tego komputera. Merge na master to O (n), gdzie n jest sumą wszystkich elementów z poszczególnych komputerów. Wybór top K pozycji jest O (n log k), gdzie n jest liczbą unikalnych adresów URL.

Rodzaje wykonywane są równolegle, z kurs, z każdym komputerem przygotowując własną posortowaną listę. Ale część "scalania" jest wykonywana w tym samym czasie, gdy komputer główny się łączy, więc istnieje pewna koordynacja i wszystkie maszyny są zaangażowane na tym etapie.