Proste wyjaśnienie MapReduce?

Powiązane z moim pytaniem na CouchDB .

Czy ktoś może wyjaśnić MapReduce w terminach, które mogą zrozumieć drętwy?

Zejście do podstaw Map i Reduce.

Mapa to funkcja, która „przekształca” elementy w jakiejś liście w inny rodzaj i umieszcza je z powrotem na tej samej liście.

przypuśćmy, że mam listę liczb: [1, 2, 3] i chcę podwoić każdą liczbę, w tym przypadku funkcją „podwojenia każdej liczby” jest funkcja x = x * 2. I bez mapowań mógłbym pisać powiedzmy prostą pętlę

A = [1, 2, 3]
foreach (item in A) A[item] = A[item] * 2

i miałbym A = [2, 4, 6], ale zamiast pisać pętle, gdybym miał funkcję mapy, mógłbym napisać

A = [1, 2, 3].Map(x => x * 2)

x => x * 2 jest funkcją do wykonania względem elementów w [1,2,3]. Dzieje się tak, że program pobiera każdy element, wykonuje (x => x * 2) względem niego, ustawiając x równe każdej pozycji i tworzy listę wyników.

1 : 1 => 1 * 2 : 2  
2 : 2 => 2 * 2 : 4  
3 : 3 => 3 * 2 : 6  

więc po wykonaniu funkcji mapowania z (x => x * 2) miałbyś [2, 4, 6].

Reduce to funkcja, która „zbiera” pozycje na listach i wykonuje obliczenia na wszystkich z nich, redukując je w ten sposób do jednej wartości.

Znajdowanie sumy lub znajdowanie średnich to przykłady funkcji redukcji. Na przykład, jeśli masz listę liczb, powiedz [7, 8, 9] i chcesz je zsumować, napiszesz taką pętlę

A = [7, 8, 9]
sum = 0
foreach (item in A) sum = sum + A[item]

Ale jeśli masz dostęp do funkcji redukcji, możesz napisać to w ten sposób

A = [7, 8, 9]
sum = A.reduce( 0, (x, y) => x + y )

Teraz jest trochę zagmatwane, dlaczego przekazano 2 argumenty (0 i funkcja z x i y). Aby funkcja redukująca była użyteczna, musi być w stanie wziąć 2 elementy, coś obliczyć i „zredukować” te 2 elementy do jednej pojedynczej wartości, dzięki czemu program mógłby zredukować każdą parę, aż otrzymamy jedną wartość.

wykonanie następowałoby:

result = 0
7 : result = result + 7 = 0 + 7 = 7
8 : result = result + 8 = 7 + 8 = 15
9 : result = result + 9 = 15 + 9 = 24

Ale nie chcesz zawsze zaczynać od zer, więc pierwszy argument służy do określenia wartości początkowej, a konkretnie wartości w pierwszym result =wierszu.

powiedz, że chcesz zsumować 2 listy, może to wyglądać tak:

A = [7, 8, 9]
B = [1, 2, 3]
sum = 0
sum = A.reduce( sum, (x, y) => x + y )
sum = B.reduce( sum, (x, y) => x + y )

lub wersję, którą najprawdopodobniej znajdziesz w prawdziwym świecie:

A = [7, 8, 9]
B = [1, 2, 3]

sum_func = (x, y) => x + y
sum = A.reduce( B.reduce( 0, sum_func ), sum_func )

Jest to dobra rzecz w oprogramowaniu DB, ponieważ dzięki obsłudze Map \ Reduce możesz pracować z bazą danych bez konieczności znajomości sposobu przechowywania danych w DB, aby z niej korzystać, do tego służy silnik DB.

Musisz tylko być w stanie "powiedzieć" silnikowi, czego chcesz, dostarczając mu funkcję Map lub Reduce, a wtedy silnik DB może znaleźć drogę dookoła danych, zastosować twoją funkcję i uzyskać wyniki chcą wszystkiego bez Twojej wiedzy, w jaki sposób zapętla się wszystkie rekordy.

Istnieją indeksy, klucze, sprzężenia i widoki oraz wiele elementów, które może pomieścić jedna baza danych, więc chroniąc Cię przed faktycznym przechowywaniem danych, Twój kod jest łatwiejszy do napisania i utrzymania.

To samo dotyczy programowania równoległego, jeśli tylko określisz, co chcesz zrobić z danymi, zamiast faktycznie implementować kod pętli, to podstawowa infrastruktura może „zrównoleglać” i wykonywać funkcje w równoległej pętli za Ciebie.

MapReduce to metoda umożliwiająca równoległe przetwarzanie ogromnych sum danych bez konieczności pisania przez programistę dowolnego kodu innego niż program mapujący i redukowania funkcji.

Funkcja mapy pobiera dane i generuje wynik, który jest przechowywany w barierze. Ta funkcja może działać równolegle z dużą liczbą tego samego zadania mapy . Zbiór danych można następnie zredukować do wartości skalarnej.

Więc jeśli myślisz o tym jak o instrukcji SQL

SELECT SUM(salary)
FROM employees
WHERE salary > 1000
GROUP by deptname

Możemy użyć mapy, aby uzyskać podzbiór pracowników z pensją> 1000, których mapa emituje do bariery w grupach wielkości grupy.

Zmniejsz sumę każdej z tych grup. Dając ci swój zestaw wyników.

Ten właśnie wyrwane z moich uniwersyteckich notatek studiów papieru google

1. Weź kilka danych
2. Wykonaj jakąś transformację, która przekształci każdy punkt odniesienia w inny rodzaj odniesienia
3. Połącz te nowe dane w jeszcze prostsze dane

Krok 2 to mapa. Krok 3 to Zmniejsz.

Na przykład,

1. Uzyskaj czas między dwoma impulsami na parze mierników ciśnienia na drodze
2. Zamapuj te czasy na prędkości w oparciu o odległość w metrach
3. Zmniejsz te prędkości do średniej prędkości

Powodem, dla którego MapReduce jest podzielony między Map i Reduce, jest to, że różne części można łatwo wykonać równolegle. (Zwłaszcza jeśli Reduce ma pewne właściwości matematyczne).

Złożony, ale dobry opis MapReduce, zobacz: Model programowania MapReduce Google - Revisited (PDF) .

MAP i REDUCE to stare funkcje Lispa z czasów, gdy człowiek zabił ostatnie dinozaury.

Wyobraź sobie, że masz listę miast z informacjami o nazwie, liczbie osób w nich mieszkających i wielkości miasta:

(defparameter *cities*
  '((a :people 100000 :size 200)
    (b :people 200000 :size 300)
    (c :people 150000 :size 210)))

Teraz możesz chcieć znaleźć miasto o największej gęstości zaludnienia.

Najpierw tworzymy listę nazw miast i gęstości zaludnienia za pomocą MAP:

(map 'list
     (lambda (city)
         (list (first city)
               (/ (getf (rest city) :people)
                  (getf (rest city) :size))))
     *cities*)

=>   ((A 500) (B 2000/3) (C 5000/7))

Za pomocą REDUCE możemy teraz znaleźć miasto o największej gęstości zaludnienia.

(reduce (lambda (a b)
          (if (> (second a) (second b))
             a
             b))
        '((A 500) (B 2000/3) (C 5000/7)))

 =>   (C 5000/7)

Łącząc obie części otrzymujemy następujący kod:

(reduce (lambda (a b)
          (if (> (second a) (second b))
             a
             b))
        (map 'list
             (lambda (city)
                (list (first city)
                   (/ (getf (rest city) :people)
                      (getf (rest city) :size))))
             *cities*))

Przedstawmy funkcje:

(defun density (city)
   (list (first city)
         (/ (getf (rest city) :people)
            (getf (rest city) :size))))

(defun max-density (a b)
   (if (> (second a) (second b))
          a
          b))

Następnie możemy napisać nasz kod REDUCE MAPY jako:

(reduce 'max-density
        (map 'list 'density *cities*))

 =>   (C 5000/7)

Wzywa MAPi REDUCE(ocena jest na lewą stronę), więc nazywa się mapą redukuj .

Weźmy przykład z artykułu Google . Celem MapReduce jest możliwość efektywnego wykorzystania obciążenia jednostek przetwarzających pracujących równolegle dla pewnego rodzaju algorytmów. Przykład jest następujący: chcesz wyodrębnić wszystkie słowa i ich liczbę w zestawie dokumentów.

Typowa realizacja:

for each document
    for each word in the document
        get the counter associated to the word for the document
        increment that counter 
    end for
end for

Implementacja MapReduce:

Map phase (input: document key, document)
for each word in the document
    emit an event with the word as the key and the value "1"
end for

Reduce phase (input: key (a word), an iterator going through the emitted values)
for each value in the iterator
    sum up the value in a counter
end for

Co więcej, będziesz mieć program główny, który podzieli zestaw dokumentów na „podziały”, które będą obsługiwane równolegle w fazie mapy. Emitowane wartości są zapisywane przez pracownika w buforze specyficznym dla pracownika. Następnie program główny deleguje innych pracowników do wykonania fazy Redukcji, gdy tylko zostanie powiadomiony, że bufor jest gotowy do obsługi.

Każdy wynik roboczy (będący pracownikiem Map lub Reduce) jest w rzeczywistości plikiem przechowywanym w rozproszonym systemie plików (GFS dla Google) lub w rozproszonej bazie danych CouchDB.

Naprawdę łatwe , szybkie i „dla opornych” wprowadzenie do MapReduce jest dostępne pod adresem : http://www.marcolotz.com/?p=67

Publikowanie niektórych treści:

Po pierwsze, dlaczego pierwotnie stworzono MapReduce?

Zasadniczo firma Google potrzebowała rozwiązania umożliwiającego łatwe zrównoleglenie dużych zadań obliczeniowych, które umożliwią dystrybucję danych na wielu maszynach połączonych w sieci. Poza tym musiał obsługiwać awarię maszyny w przejrzysty sposób i zarządzać problemami z równoważeniem obciążenia.

Jakie są prawdziwe mocne strony MapReduce?

Można powiedzieć, że magia MapReduce opiera się na zastosowaniu funkcji Map i Reduce. Muszę przyznać kolego, że zdecydowanie się z tym nie zgadzam. Główną cechą, która uczyniła MapReduce tak popularną, jest możliwość automatycznej równoległości i dystrybucji w połączeniu z prostym interfejsem. Te czynniki w połączeniu z przejrzystą obsługą większości błędów sprawiły, że ten framework stał się tak popularny.

Trochę więcej głębi na papierze:

MapReduce zostało pierwotnie wymienione w artykule Google (Dean i Ghemawat, 2004 - link tutaj) jako rozwiązanie do wykonywania obliczeń w Big Data przy użyciu podejścia równoległego i klastrów typu towar-komputer. W przeciwieństwie do Hadoopa, który jest napisany w Javie, framework Google jest napisany w C ++. W dokumencie opisano, jak struktura równoległa zachowywałaby się przy użyciu funkcji Map i Reduce z programowania funkcjonalnego na dużych zbiorach danych.

W tym rozwiązaniu byłyby dwa główne kroki - zwane Mapowaniem i Zmniejszaniem - z opcjonalnym krokiem między pierwszym a drugim - zwane Połącz. Krok Map byłby uruchamiany jako pierwszy, wykonywał obliczenia w wejściowej parze klucz-wartość i generował nową wyjściową klucz-wartość. Należy pamiętać, że format wejściowych par klucz-wartość nie musi koniecznie odpowiadać parze formatu wyjściowego. Krok Zmniejsz zgromadziłby wszystkie wartości tego samego klucza, wykonując na nim inne obliczenia. W rezultacie ten ostatni krok wyprowadziłby pary klucz-wartość. Jedną z najbardziej trywialnych aplikacji MapReduce jest implementacja liczenia słów.

Pseudokod tej aplikacji jest podany poniżej:

map(String key, String value):

// key: document name
// value: document contents
for each word w in value:
EmitIntermediate(w, “1”);

reduce(String key, Iterator values):

// key: a word
// values: a list of counts
int result = 0;
for each v in values:
    result += ParseInt(v);
Emit(AsString(result));

Jak można zauważyć, mapa odczytuje wszystkie słowa w rekordzie (w tym przypadku rekord może być linią) i emituje słowo jako klucz oraz liczbę 1 jako wartość. Później redukcja grupuje wszystkie wartości tego samego klucza. Podajmy przykład: wyobraź sobie, że słowo „dom” występuje w rekordzie trzy razy. Wejście reduktora to [house, [1,1,1]]. W reduktorze zsumuje wszystkie wartości dla domu kluczy i da na wyjściu następującą wartość klucza: [house, [3]].

Oto obraz tego, jak wyglądałoby to we frameworku MapReduce:

Jako kilka innych klasycznych przykładów aplikacji MapReduce można powiedzieć:

• Liczba częstotliwości dostępu do adresu URL

• Odwrotny wykres linków internetowych

• Rozproszony Grep

• Wektor terminów na hosta

Aby uniknąć zbyt dużego ruchu w sieci, w artykule opisano, w jaki sposób struktura powinna próbować zachować lokalność danych. Oznacza to, że zawsze powinien starać się upewnić, że maszyna, na której są uruchomione zadania mapowania, ma dane w swojej pamięci / lokalnej pamięci, unikając pobierania ich z sieci. Mając na celu zredukowanie sieci poprzez wprowadzenie elementu odwzorowującego, używany jest opcjonalny krok sumatora, opisany wcześniej. Combiner wykonuje obliczenia na wyjściu mapperów na danej maszynie przed wysłaniem ich do reduktorów - które mogą znajdować się na innej maszynie.

Dokument opisuje również, jak elementy frameworka powinny zachowywać się w przypadku błędów. Te elementy w artykule nazywane są pracownikami i mistrzami. W implementacjach open source zostaną podzielone na bardziej szczegółowe elementy. Ponieważ Google tylko opisał podejście w artykule i nie udostępnił własnego oprogramowania, stworzono wiele frameworków open source w celu wdrożenia modelu. Jako przykłady można powiedzieć Hadoop lub ograniczoną funkcję MapReduce w MongoDB.

Środowisko wykonawcze powinno zadbać o szczegóły programistów niebędących ekspertami, takie jak partycjonowanie danych wejściowych, planowanie wykonywania programu na dużym zestawie maszyn, obsługa awarii maszyn (oczywiście w przejrzysty sposób) i zarządzanie komunikacją między maszynami . Doświadczony użytkownik może dostroić te parametry, ponieważ dane wejściowe zostaną podzielone między procesy robocze.

Kluczowe idee:

• Tolerancja błędów:Musi z wdziękiem tolerować awarie maszyn. W tym celu kapitan okresowo pinguje pracowników. Jeżeli kapitan nie otrzyma odpowiedzi od danego pracownika w określonym czasie, kapitan określi pracę jako nieudaną u tego pracownika. W takim przypadku wszystkie zadania mapy ukończone przez wadliwego pracownika są odrzucane i przekazywane innemu dostępnemu robotnikowi. Podobnie dzieje się, gdy pracownik nadal przetwarzał mapę lub zadanie redukcji. Należy zauważyć, że jeśli pracownik wykonał już swoją część redukującą, wszystkie obliczenia zostały już zakończone przed niepowodzeniem i nie trzeba ich resetować. Jako główny punkt awarii, jeśli master zawiedzie, całe zadanie kończy się niepowodzeniem. Z tego powodu można zdefiniować okresowe punkty kontrolne dla administratora w celu zapisania jego struktury danych.

• Lokalność: aby uniknąć ruchu sieciowego, struktura próbuje upewnić się, że wszystkie dane wejściowe są lokalnie dostępne dla maszyn, które mają wykonywać na nich obliczenia. W pierwotnym opisie używa systemu plików Google (GFS) ze współczynnikiem replikacji ustawionym na 3 i rozmiarami bloków 64 MB. Oznacza to, że ten sam blok o wielkości 64 MB (który tworzy plik w systemie plików) będzie miał identyczne kopie na trzech różnych komputerach. Mistrz wie, gdzie są bloki i próbuje zaplanować zadania map w tej maszynie. Jeśli to się nie powiedzie, master próbuje przydzielić maszynę w pobliżu repliki danych wejściowych zadań (tj. Maszynę roboczą w tej samej szafie co maszyna danych).

• Szczegółowość zadań: Zakładając, że każda faza mapy jest podzielona na M części, a każda faza Redukcji jest podzielona na R części, idealnie byłoby, gdyby M i R były dużo większe niż liczba maszyn pracujących. Wynika to z faktu, że pracownik wykonujący wiele różnych zadań poprawia dynamiczne równoważenie obciążenia. Poza tym zwiększa szybkość odzyskiwania w przypadku awarii pracownika (ponieważ wiele wykonanych przez niego zadań mapy można rozłożyć na wszystkie inne maszyny).

• Zadania tworzenia kopii zapasowych: Czasami pracownik mapy lub reduktora może zachowywać się znacznie wolniej niż inni w klastrze. Może to utrzymać całkowity czas przetwarzania i sprawić, że będzie on równy czasowi przetwarzania tej pojedynczej powolnej maszyny. W oryginalnym artykule opisano alternatywę zwaną Zadaniami kopii zapasowej, które są planowane przez mistrza, gdy operacja MapReduce jest bliska ukończenia. Są to zadania, które są planowane przez mistrza zadań w toku. W ten sposób operacja MapReduce kończy się po zakończeniu podstawowej lub kopii zapasowej.

• Liczniki: Czasami ktoś może chcieć policzyć zdarzenia. Z tego powodu liczy się miejsce utworzenia. Wartości liczników dla każdego pracownika są okresowo propagowane do kapitana. Master następnie agreguje (tak. Wygląda na to, że agregatory Pregel pochodzą z tego miejsca) wartości liczników pomyślnie zakończonej mapy, redukują zadanie i zwracają je do kodu użytkownika po zakończeniu operacji MapReduce. W statusie nadrzędnym dostępna jest również aktualna wartość licznika, więc człowiek obserwujący proces może śledzić jego zachowanie.

Cóż, myślę, że biorąc pod uwagę wszystkie powyższe koncepcje, Hadoop będzie dla ciebie bułką z masłem. Jeśli masz jakiekolwiek pytania dotyczące oryginalnego artykułu MapReduce lub cokolwiek z nim związanego, daj mi znać.

Nie chcę brzmieć banalnie, ale to mi bardzo pomogło i jest całkiem proste:

cat input | map | reduce > output

Jeśli znasz Python, poniżej znajduje się najprostsze możliwe wyjaśnienie MapReduce:

In [2]: data = [1, 2, 3, 4, 5, 6]
In [3]: mapped_result = map(lambda x: x*2, data)

In [4]: mapped_result
Out[4]: [2, 4, 6, 8, 10, 12]

In [10]: final_result = reduce(lambda x, y: x+y, mapped_result)

In [11]: final_result
Out[11]: 42

Zobacz, jak każdy segment surowych danych był przetwarzany indywidualnie, w tym przypadku pomnożony przez 2 ( część mapy MapReduce). Na podstawie tego mapped_resultdoszliśmy do wniosku, że wynikiem będzie 42( zmniejszona część MapReduce).

Ważnym wnioskiem z tego przykładu jest fakt, że każdy fragment przetwarzania nie zależy od innego fragmentu. Na przykład, jeśli thread_1mapy [1, 2, 3]i thread_2mapy [4, 5, 6], ostateczny wynik obu wątków nadal byłby, [2, 4, 6, 8, 10, 12]ale zmniejszyliśmy o połowę czas przetwarzania. To samo można powiedzieć o operacji redukuj i stanowi ona istotę działania MapReduce w obliczeniach równoległych.