MapaReduce

Map R educe wywodzi się z zasad znanych od lat osiemdziesiątych przetwarzania rozproszonego .

Map R educe to model programowania do obsługi obliczeń równoległych na dużych zbiorach danych w grupach komputerów i przetwarzania towarów . Nazwa frameworka jest inspirowana nazwami dwóch ważnych metod , makr lub funkcji w programowaniu funkcjonalnym: Map i Reduce . Map Reduce został przyjęty na całym świecie, ponieważ istnieje implementacja OpenSource o nazwie Hadoop . Jego rozwój był początkowo kierowany przez Yahoo , a obecnie jest realizowany przez projekt Apache . Od 2010 roku pojawiło się kilka inicjatyw podobnych do Hadoop zarówno w przemyśle, jak i na uczelni. Implementacje bibliotek Map Reduce zostały napisane w różnych językach programowania , takich jak C ++ , Java i Python .

Map Reduce jest używany w praktycznej rozdzielczości niektórych algorytmów, które można zrównoleglać . Jednak Map Reduce nie jest rozwiązaniem dla każdego problemu, podobnie jak każdy problem nie może być skutecznie rozwiązany przez Map Reduce . Problemy z dużymi zestawami danych, osiągającymi rozmiar w petabajtach , są ogólnie rozwiązywane. Z tego powodu ten framework zwykle działa w rozproszonym systemie plików (HDFS).

Historia

Wczesne implementacje Google wymagały wykonywania dużych operacji mnożenia macierzy w celu obliczenia PageRank , czyli rankingu stron w wyszukiwaniu. W ten sposób Map Reduce stała się popularna jako metoda obliczeniowa algebry liniowej . Troska o radzenie sobie z dużymi zbiorami danych doprowadziła do stworzenia algorytmów i struktur zdolnych do przetwarzania terabajtów informacji. Jedna z pierwszych aplikacji zdolnych do programowania Map Reduce została początkowo zaimplementowana w Hadoop , początkowo zaprojektowana przez Douga Cutting który na cześć zabawkowego słonia swojego syna. Został pierwotnie opracowany w celu wspierania dystrybucji projektu wyszukiwarki Nutch . ]

Przegląd

Map Reduce to platforma do równoległego przetwarzania problemów w dużych zbiorach danych przy użyciu dużej liczby komputerów (węzłów), określanych zbiorczo jako klaster (jeśli wszystkie węzły znajdują się w tej samej sieci lokalnej i używają podobnego sprzętu) lub sieci (jeśli węzły są współużytkowane przez systemy rozproszone geograficznie i administracyjnie oraz używają bardziej heterogenicznego sprzętu). Przetwarzanie może odbywać się na danych przechowywanych w systemie plików (nieustrukturyzowany) lub w bazie danych (ustrukturyzowany). Map Reduce może wykorzystać lokalizację danych, przetwarzając je blisko miejsca ich przechowywania, aby zminimalizować obciążenie komunikacyjne.

Struktura (lub system) Map Reduce składa się zazwyczaj z trzech operacji (lub kroków):

1. Mapa: Każdy węzeł podrzędny (pracownik) stosuje funkcję mapdo danych lokalnych i zapisuje dane wyjściowe do pamięci tymczasowej. Węzeł główny zapewnia przetwarzanie tylko jednej kopii nadmiarowych danych wejściowych.
2. Shuffle: Pracownicy redystrybuują dane na podstawie kluczy wyjściowych (wytworzonych przez funkcję map), tak aby wszystkie dane należące do klucza znajdowały się w tym samym węźle roboczym.
3. Zmniejszenie: Węzły robocze przetwarzają teraz równolegle każdą grupę danych wyjściowych według klucza.

M ap R educe umożliwia rozproszone przetwarzanie map i zmniejsza liczbę operacji. Mapy można wykonywać równolegle, o ile każda operacja mapowania jest niezależna od pozostałych; w praktyce jest to ograniczone liczbą niezależnych źródeł danych i/lub liczbą procesorów w pobliżu każdego źródła. Podobnie zestaw „reduktorów” może wykonać fazę redukcji, o ile wszystkie wyjścia operacji mapy, które mają ten sam klucz, są prezentowane w tym samym czasie w tym samym reduktorze lub funkcja zmniejszania jest asocjacyjna . Chociaż ten proces często wydaje się nieefektywny w porównaniu z algorytmami, które są bardziej sekwencyjne (ponieważ musi być wykonanych wiele wystąpień procesu Reduce), Map Reduce można zastosować do znacznie większych zestawów danych niż może obsłużyć pojedynczy „podstawowy” serwer. Duża farma serwerów może użyć funkcji Map Reduce , aby posortować petabajt danych w ciągu zaledwie kilku godzin. Równoległość oferuje również możliwość odzyskania sprawności po częściowej awarii serwerów lub pamięci masowej podczas działania: jeśli mapper lub reduktor ulegnie awarii, zadanie można przełożyć na inny termin, zakładając, że dane wejściowe są nadal dostępne.

Widok logiczny

Nie wszystkie procesy mogą być obsługiwane z systemu Map Reduce . W szczególności adresowalne są tylko te, które można podzielić na operacje map() i Reduce() , co jest ważne przy wyborze tego frameworka do rozwiązania problemu. Funkcje Map i Reduce są zdefiniowane w odniesieniu do danych ustrukturyzowanych w krotki typu (klucz, wartość).

Funkcja Map()

Map pobiera jedną z tych par danych z typem w jednej domenie danych i zwraca listę par w innej domenie:

Mapa(k 1 ,v 1 ) -> lista(k 2 ,v 2 )

• Funkcja map() : obsługuje mapowanie i jest stosowana równolegle dla każdego elementu we wprowadzanych danych . Daje to listę par (k2,v2) dla każdego wywołania. Następnie platforma Map Reduce zbiera wszystkie pary z tym samym kluczem ze wszystkich list i grupuje je, tworząc grupę dla każdego z różnych wygenerowanych kluczy. Z architektonicznego punktu widzenia węzeł nadrzędny pobiera dane wejściowe, dzieli je na małe fragmenty lub problemy o mniejszej tożsamości i dystrybuuje je do tzw. węzłów roboczych . Węzeł roboczy może się ponownie podzielić, dając początek strukturze drzewiastej . Węzeł roboczy przetwarza problem i przekazuje odpowiedź do węzła głównego.

Funkcja Reduce()

Funkcja zmniejszania jest stosowana równolegle dla każdej grupy, tworząc zbiór wartości dla każdej domeny:

Zmniejsz(k 2 , lista (v 2 )) -> lista(v 3 )

• Funkcja Reduce () : Każde wywołanie funkcji Reduce zwykle generuje wartość v3 lub puste wywołanie, chociaż wywołanie może zwrócić więcej niż jedną wartość. Zwrot wszystkich tych połączeń jest zbierany jako lista pożądanych wyników.

W związku z tym struktura MapReduce przekształca listę par (klucz, wartość) w listę wartości. To zachowanie różni się od join programowania funkcyjnego "map and Reduce" , które akceptuje dowolną listę wartości i zwraca pojedynczą wartość, która łączy wszystkie wartości zwrócone przez map.

Architektura MapReduce

Funkcja map() jest wykonywana w sposób rozproszony na wielu komputerach. Dane wejściowe, zwykle pochodzące z dużego pliku (pliku), są podzielone na zestaw M partycji wejściowych o wielkości od 16 do 64 megabajtów. Te partycje mogą być przetwarzane na różnych komputerach. W wywołaniu MapReduce zwykle występuje kilka operacji :

• Wpisy są następnie dzielone na M partycji o przybliżonej wielkości od 16 do 64 megabajtów . Program MapReduce zaczyna tworzyć wystąpienia na różnych komputerach w klastrze. Z reguły liczba instancji jest konfigurowana w aplikacjach.

• Jeden z egzemplarzy programu jest wyjątkowy i pełni rolę „mastera”. Reszta kopii nazywana jest „pracownikami” i otrzymuje od mistrza przydział swoich zadań. Uważa się, że istnieje wiele zadań M map() i R Reduce(). „Mistrz” jest odpowiedzialny za zbieranie „pracowników” w stanie spoczynku (tzn. bez przypisanego zadania) i przypisze do niego określone zadanie map() lub Reduce(). Pracownik może mieć tylko trzy stany: bezczynny, pracujący, kompletny.

• Pracownik, któremu przydzielono określone zadanie map() , przyjmie jako dane wejściowe odpowiednią partycję. Przeanalizuje pary (klucz, wartość), aby utworzyć nową parę wyjściową, jak określono w jego programowaniu. Pary klucz-wartość generowane przez funkcję map() są buforowane w pamięci.

• Okresowo buforowane pary klucz-wartość są zapisywane na dysku lokalnym w regionach R. Regiony tych par klucz-wartość są przekazywane do mastera, który jest odpowiedzialny za przekierowywanie procesów roboczych, które mają zadania reduction().

• Kiedy proces roboczy typu Reduce jest powiadamiany przez „master” z lokalizacją partycji, używa wywołań zdalnych do odczytania informacji przechowywanych na dyskach twardych różnych procesów roboczych typu map(). Kiedy proces roboczy typu Reduce() odczytuje wszystkie dane pośrednie, porządkuje klucze w taki sposób, że znalezione dane z tym samym kluczem są grupowane razem. Porządkowanie jest konieczne , ponieważ z reguły wiele różnych klawiszy funkcyjnych map() może być przypisanych do tej samej funkcji reduction(). W przypadkach, gdy ilość danych pośrednich jest bardzo duża, często stosuje się sortowanie zewnętrzne.

• Proces roboczy typu Reduce() iteruje po zbiorze pośrednich posortowanych wartości i robi to dla każdego znalezionego unikalnego klucza. Pobiera klucz i skojarzony z nim zbiór wartości i przekazuje je do funkcji Reduce(). Dane wyjściowe Reduce() są dodawane do pliku wyjściowego MapReduce.

• Kiedy wszystkie zadania map() i Reduce() zakończą się, „master” wywołuje program użytkownika. W tym momencie wywołanie MapReduce zwraca kontrolę do kodu użytkownika.

Zadania są uważane za zakończone, gdy ta kontrolka zostanie zwrócona użytkownikowi. Dane wyjściowe są dystrybuowane w kompletnym pliku lub w przypadku braku są dystrybuowane w plikach R. Te pliki R mogą być danymi wejściowymi innego MapReduce lub mogą być przetwarzane przez dowolny inny program, który potrzebuje tych danych.

Combiner (agregatorzy lokalni)

W środowisku klastrowym jednym z ograniczeń jest transport dużych plików między komputerami ze względu na ich ograniczoną przepustowość . W strukturze MapReduce funkcja map() zapisuje w lokalnym buforze znaków, takim jak dysk twardy . Informacje zapisywane lokalnie są agregowane i porządkowane przez funkcję agregatora odpowiedzialną za wykonanie tej operacji. Uporządkowane wartości mają postać [k, [v ₁ , v ₂ , v ₃ , ..., v _n ]]. W ten sposób funkcja Reduce() otrzymuje z kombinatora listę wartości skojarzonych z pojedynczym kluczem. Ponieważ opóźnienie sieci komputerów i ich dysków jest zwykle większe niż jakakolwiek inna operacja, każde zmniejszenie ilości danych pośrednich zwiększy wydajność algorytmów. W MapReduce każda lokalna agregacja wyników pośrednich powoduje rzeczywistą poprawę globalnej wydajności.

Z tego powodu wiele oficjalnych dystrybucji MapReduce często zawiera lokalne operacje agregacji przy użyciu funkcji zdolnych do lokalnej agregacji danych. Unikanie lub ograniczanie w miarę możliwości przenoszenia dużych plików. Albo dodane do funkcji map(), albo do lokalnych agregatorów.

Tolerancja błędu

Mechanizm MapReduce jest odporny na błędy , gdy jeden z pracowników ulega awarii. Podobnie jak MapReduce, został zaprojektowany dla procesów obejmujących duże rozmiary danych przy użyciu setek lub tysięcy komputerów. Nawet jeśli prawdopodobieństwo awarii jest niskie, bardzo możliwe, że jeden (lub kilku) pracowników zostanie dezaktywowanych właśnie z powodu awarii maszyny, która go obsługiwała. „Master” okresowo wysyła pingi do każdego pracownika, aby sprawdzić jego status.

Jeśli nie ma odpowiedzi po określonym czasie oczekiwania, master interpretuje, że pracownik jest dezaktywowany. Każde zadanie map(), które zostało ukończone przez pracownika, natychmiast powraca do stanu oczekiwania, a zatem może kwalifikować się do przypisania innym pracownikom. Podobnie każda funkcja map() (lub Reduce) działająca podczas awarii jest resetowana do stanu bezczynności i może zostać wybrana do ponownego przydziału.

Ukończone zadania map() są częściowo ponownie wykonywane w przypadku niepowodzenia, ponieważ ich dane wyjściowe są przechowywane na lokalnych dyskach maszyny, która uległa awarii, i dlatego jest uważana za niedostępną. Całe zadania Reduce() nie muszą być ponownie wykonywane, ponieważ ich dane wyjściowe są przechowywane w systemie globalnym. Gdy zadanie map() jest wykonywane przez pracownika A, a następnie przez pracownika B (głównie z powodu awarii), w tym przypadku wszystkie zadania reduction() są powiadamiane o usunięciu danych od pracownika A i zaakceptowaniu danych od pracownika A. pracownik B. W ten sposób wykonanie MapReduce jest odporne .

Przykłady

W opisie przykładów użycia Map Reduce wystarczy szczegółowo opisać, jak w każdym przypadku realizowane są operacje map() i Reduce () . Literatura przedstawia powtarzające się przykłady zliczania słów w dokumencie, operacje na macierzach i operacje zapytań w relacyjnej bazie danych.

Liczba słów

Ten przykład Map Reduce to proces liczenia wystąpień każdego słowa w zestawie dokumentów :

 mapa ( String name , String document ) :   
  // klucz: nazwa dokumentu 
// wartość: treść dokumentu dla każdego słowa w w dokumencie :  
       
    EmitPośredni ( w , 1 ); 

W tym przypadku funkcja map() dzieli dokument na słowa (czyli tokenizuje go ) za pomocą prostego analizatora leksykalnego i wypisuje serię krotek w postaci ( klucz , wartość ), gdzie kluczem jest słowo, a wartość jest „1”. Czyli np. z dokumentu "Mały domek na prerii" funkcja mapy zwróciłaby: ("la", "1"), ("dom", "1"), ("z", "1" ), („the”, „1”), („łąka”, „1”).

 
 zmniejszyć ( Słowo typu String , PartialCounts Iteratora ) :   
  // słowo: słowo 
// częściowe zliczanie: [[Iterator (wzorzec projektowy)|lista częściowa]] do zliczania agregacji int wynik = 0 ;  
     
  dla każdego v w częściowych liczbach :    
    wynik += ParseInt ( v );  
  Emituj ( wynik );

Tutaj każdy dokument jest podzielony na słowa , a każde słowo jest liczone z początkową wartością „1” przez funkcję Map, używając słowa jako kluczowego wyniku. Struktura gromadzi wszystkie pary z tym samym kluczem i przekazuje je do tego samego wywołania Redukuj, więc ta funkcja potrzebuje tylko sumy wszystkich wartości wejściowych, aby znaleźć wszystkie wystąpienia tego słowa. W powyższym przykładzie ("la", "1") pojawia się dwa razy, ponieważ klawisz "la" ma dwa wystąpienia, reszta klawiszy pojawia się tylko raz.

Mnożenie macierzy przez wektor

Przykłady algebry liniowej dla operacji na macierzach są najbardziej odpowiednie ze względu na przydatność szkieletu w tych przypadkach. Załóżmy, że mamy macierz kwadratową M o rozmiarze nxn. Element znajdujący się w wierszu i oraz kolumnie j nazywamy m _ij . Załóżmy, że mamy wektor v taki, że na pozycji j mamy element v _j . W ten sposób wynikiem mnożenia między macierzą M a wektorem v będzie wektor x o długości n w taki sposób, że element x _i będzie taki, że:

$x_{i} = \sum_{j=1}^n m_{ij} v_j$

Operacja ta jest wykonywana bez problemu dla tablic składających się z kilku tysięcy elementów, kosztujących kilka milionów. Problem z jego obliczeniem pojawia się, gdy ma być wykonany z setkami miliardów . Z tego powodu zakłada się w zastosowaniu M do Redukcji , że n jest rzędu 10 ¹² . Funkcja map() w tym przypadku pobiera wiersz i macierzy i cały wektor v , tworząc pary: (i, m _ij v _j ). To znaczy postaci (1, m ₁₁ v ₁ ), (1, m ₁₂ v ₂ ), (1, m ₁₃ v ₃ ) ... (1, m _1j v _j ).

 mapa ( Vector rowMatrix , Vector vector ) :   
  // klucz: i -> indeks wektora 
// wartość: iloczyn m<sub>ij</sub> przez v<sub>j</sub>. dla każdej pozycji i w wektorze :  
       
    EmitIntermediate ( i , wartość ); 

Funkcja Reduce() w tym przypadku musi jedynie zebrać pary, które mają ten sam klucz i, i dodać je.

 
 zmniejszyć ( Słowo typu String , PartialCounts Iteratora ) :   
  // słowo: słowo 
// częściowe zliczanie: [[Iterator (wzorzec projektowy)|lista częściowa]] do zliczania agregacji int wynik = 0 ;  
     
  dla każdego v w częściowych liczbach :    
    wynik += ParseInt ( v );  
  Emituj ( wynik );

Przepływ danych

Struktura MapReduce to duży algorytm sortowania rozproszonego. Główne moduły, które definiuje aplikacja to:

• czytnik wejściowy
• Funkcja mapy
• funkcja partycji
• Funkcja porównawcza
• Funkcja redukcji
• Pisarz wychodzący

Czytnik wejściowy

Czytnik danych wejściowych dzieli dane wejściowe na „kawałki” o odpowiedniej wielkości (zwykle od 64 MB do 128 MB), a struktura przypisuje wycinek do każdej funkcji Map . Czytnik danych wejściowych odczytuje dane z magazynu stabilnego (zazwyczaj Distributed File System ) i generuje pary klucz/wartość.

Typowy przykład odczyta katalog pełen plików tekstowych i zwróci każdy wiersz jako rekord.

Funkcja mapy

Funkcja Map pobiera serię par klucz/wartość, przetwarza je i generuje zero lub więcej wyjściowych par klucz/wartość. Rodzaje map wejściowych i wyjściowych mogą być (i często różnią się) od siebie.

Jeśli aplikacja liczy słowa, funkcja Map podzieli wiersze na słowa i wygeneruje wyjściową parę klucz/wartość dla każdego słowa. Każda para wyjściowa będzie zawierać słowo jako klucz i liczbę jego wystąpień w wierszu jako wartość.

Funkcja partycji

Każde wyjście funkcji Map jest przypisane do reduktora przy użyciu funkcji partycji w celu wygenerowania fragmentacji. Funkcja partycji otrzymuje klucz i liczbę reduktorów i zwraca indeks żądanego reduktora .

Domyślnym zachowaniem jest pobranie skrótu klucza i użycie skrótu modulo liczba reduktorów . Ważne jest, aby wybrać funkcję partycji, która generuje w przybliżeniu równomierną dystrybucję danych na fragment, aby zachować równowagę, w przeciwnym razie operacja MapReduce może spowolnić, czekając na zakończenie powolnych reduktorów (reduktorów przypisanych do większej liczby danych niż jest zawartych we fragmencie).

Między etapami mapowania i redukcji dane są tasowane (sortowane równolegle / zamieniane między węzłami) w celu przeniesienia danych z fragmentu, w którym zostały utworzone, do fragmentu, w którym zostaną zredukowane. Tasowanie może w niektórych przypadkach zająć więcej czasu niż przetwarzanie, w zależności od przepustowości, szybkości procesora, generowanych danych i czasu zużytego między mapowaniem a redukcją przetwarzania.

Funkcja porównania

Dane wejściowe dla każdej redukcji są uzyskiwane z maszyny, na której mapa została wykonana i zamówiona za pomocą funkcji porównania

Funkcja redukcji

Struktura wywołuje funkcję Shrink aplikacji raz dla każdego unikalnego klucza na posortowanej liście. Zmniejszenie może iterować przez wartości skojarzone z tym kluczem i generować zero lub więcej danych wyjściowych.

W przykładzie liczby słów funkcja Reduce pobiera wartości wejściowe, dodaje je do siebie i generuje pojedyncze dane wyjściowe dla słowa i sumy końcowej.

Zakończ pisarza

Moduł zapisujący dane wyjściowe zapisuje dane wyjściowe funkcji Shrink do tabel pamięci, zwykle rozproszonego systemu plików.

Używa

Zasadniczo funkcja Map Reduce jest używana w przypadku współbieżnych problemów obliczeniowych związanych z dużymi zestawami danych , które muszą być przetwarzane przez dużą liczbę komputerów (węzłów), które są zbiorczo określane jako klastry (jeśli wszystkie węzły znajdują się w tej samej sieci lokalnej i przy użyciu tego samego sprzętu) lub do sieci (jeśli węzły są współdzielone w sposób rozproszony na dużych obszarach geograficznych lub administracyjnych, i które zazwyczaj mają bardziej heterogeniczny sprzęt). Przetwarzanie równoległe może odbywać się przy użyciu danych przechowywanych w systemie plików (nieustrukturyzowanym) lub w bazie danych (ustrukturyzowany). Z tego powodu jest on używany w aplikacjach, które zawierają dane na dużą skalę, takich jak aplikacje równoległe, indeksowanie stron internetowych, eksploracja danych i symulacje naukowe.

Zobacz także

Implementacje MapReduce

• Apache Hadoop
• Podstawa kanapy
• kanapa
• Nieskończoność
• MongoDB
• Riak