KaartVerminderen

M ap R educe is ontstaan ​​uit de sinds de jaren tachtig bekende principes van distributed computing .

M ap R educe is een programmeermodel ter ondersteuning van parallel computing op grote gegevensverzamelingen in groepen computers en commodity computing . De naam van het framework is geïnspireerd op de namen van twee belangrijke methoden , macro's of functies in functioneel programmeren: Map en Reduce . M ap R educe is wereldwijd geadopteerd, omdat er een OpenSource- implementatie is genaamd Hadoop . De ontwikkeling ervan werd aanvankelijk geleid door Yahoo en wordt momenteel uitgevoerd door het Apache-project . Sinds de jaren 2010 zijn er verschillende initiatieven vergelijkbaar met Hadoop in zowel de industrie als de academische wereld. Implementaties van Map Reduce-bibliotheken zijn geschreven in verschillende programmeertalen zoals C ++ , Java en Python .

M ap R educe wordt gebruikt bij de praktische resolutie van sommige algoritmen die kunnen worden geparalleliseerd. Map Reduce is echter niet de oplossing voor elk probleem, net zoals elk probleem niet efficiënt kan worden opgelost door Map Reduce . Problemen met grote datasets, die petabytes groot kunnen worden, worden over het algemeen aangepakt. Om deze reden draait dit framework meestal op een gedistribueerd bestandssysteem (HDFS).

Geschiedenis

Vroege Google-implementaties moesten grote matrixvermenigvuldigingsbewerkingen uitvoeren om PageRank te berekenen , dat wil zeggen de rangschikking van pagina's in een zoekopdracht. Op deze manier werd Map Reduce populair als rekenmethode voor lineaire algebra . De zorg voor het omgaan met grote gegevensverzamelingen leidde tot de creatie van algoritmen en kaders die terabytes aan informatie konden verwerken. Een van de eerste toepassingen waarmee Map Reduce kon worden geprogrammeerd, werd aanvankelijk geïmplementeerd in Hadoop , oorspronkelijk ontworpen door Doug Cutting , die het naar de speelgoedolifant van zijn zoon noemde. Het werd oorspronkelijk ontwikkeld om de distributie van het Nutch - zoekmachineproject te ondersteunen . ]

Overzicht

M ap R educe is een raamwerk voor het parallel verwerken van problemen over grote datasets met behulp van een groot aantal computers (knooppunten), gezamenlijk aangeduid als een cluster (als alle knooppunten zich op hetzelfde lokale netwerk bevinden en vergelijkbare hardware gebruiken) of een netwerk (als knooppunten worden gedeeld over geografisch en administratief gedistribueerde systemen en meer heterogene hardware gebruiken). Verwerking kan plaatsvinden op gegevens die zijn opgeslagen in een bestandssysteem (ongestructureerd) of in een database (gestructureerd). M ap R educe kan profiteren van de locatie van de gegevens door deze te verwerken in de buurt van waar ze zijn opgeslagen om de communicatie-overhead tot een minimum te beperken.

Een Map Reduce - framework (of systeem) bestaat doorgaans uit drie bewerkingen (of stappen):

1. Map: elk slave-knooppunt (werknemer) past de functie maptoe op lokale gegevens en schrijft de uitvoer naar tijdelijke opslag. Een masternode zorgt ervoor dat slechts één kopie van de redundante invoergegevens wordt verwerkt.
2. Shuffle: Werknemers herdistribueren gegevens op basis van uitvoersleutels (geproduceerd door de functie map), zodat alle gegevens die bij een sleutel horen, zich in hetzelfde werkknooppunt bevinden.
3. Verminderen: de werkknooppunten verwerken nu elke groep uitvoergegevens, per sleutel, parallel.

M ap R educe maakt gedistribueerde verwerking van kaarten mogelijk en reduceert bewerkingen. Kaarten kunnen parallel worden uitgevoerd, zolang elke kaartbewerking onafhankelijk is van de andere; in de praktijk wordt dit beperkt door het aantal onafhankelijke gegevensbronnen en/of het aantal CPU's in de buurt van elke bron. Evenzo kan een set "verloopstukken" de verkleiningsfase uitvoeren, zolang alle uitgangen van de kaartbewerking die dezelfde toets delen tegelijkertijd aan dezelfde verkleiner worden aangeboden, of de verkleiningsfunctie associatief is . Hoewel dit proces vaak inefficiënt lijkt in vergelijking met algoritmen die meer sequentieel zijn (omdat meerdere instanties van het reductieproces moeten worden uitgevoerd), kan Map Reduce worden toegepast op aanzienlijk grotere datasets dan een enkele "basis" server aankan. ". Een grote serverfarm kan Map Reduce gebruiken om een ​​petabyte aan gegevens in slechts een paar uur te sorteren. Parallellisme biedt ook de mogelijkheid om te herstellen van een gedeeltelijke storing van servers of opslag tijdens bedrijf: als een mapper of reducer uitvalt, kan de taak opnieuw worden gepland, ervan uitgaande dat invoergegevens nog beschikbaar zijn.

Logische

Niet alle processen kunnen worden aangesproken vanuit het M ap R educe framework . In het bijzonder zijn alleen de bewerkingen die kunnen worden onderverdeeld in de bewerkingen map() en reduce() adresseerbaar, en dit is belangrijk bij het kiezen van dit raamwerk om een ​​probleem op te lossen. De functies Map en Reduce zijn beide gedefinieerd met betrekking tot gegevens die zijn gestructureerd in tupels van het type (sleutel, waarde).

Kaart() functie

Kaart neemt een van deze gegevensparen met een type in één gegevensdomein en retourneert een lijst met paren in een ander domein:

Kaart(k 1 ,v 1 ) -> lijst(k 2 ,v 2 )

• De functie map() : verwerkt de toewijzing en wordt parallel toegepast voor elk item in de gegevensinvoer . Dit levert een lijst op van (k2,v2) paren voor elke aanroep. Daarna verzamelt het Map Reduce-framework alle paren met dezelfde sleutel uit alle lijsten en groepeert ze, waardoor een groep wordt gemaakt voor elk van de verschillende gegenereerde sleutels. Vanuit architectonisch oogpunt neemt het hoofdknooppunt de invoer, verdeelt het in kleine stukjes of problemen met een kleinere identiteit en distribueert deze naar de zogenaamde werkknooppunten . Een werkknooppunt kan weer onderverdelen, waardoor een boomstructuur ontstaat . Het werkknooppunt verwerkt het probleem en geeft het antwoord door aan het hoofdknooppunt.

Reduce() functie

De reduceerfunctie wordt parallel toegepast voor elke groep, waardoor een verzameling waarden voor elk domein wordt geproduceerd:

Verkleinen(k 2 , lijst (v 2 )) -> lijst(v 3 )

• De functie reduce() : Elke aanroep van Reduce produceert doorgaans een v3-waarde of een lege aanroep, hoewel een aanroep meer dan één waarde kan retourneren. De terugkeer van al die oproepen wordt verzameld als de gewenste resultatenlijst.

Daarom transformeert het MapReduce-framework een lijst met (sleutel, waarde) paren in een lijst met waarden. Dit gedrag verschilt van de "map and reduce" join van functionele programmering , die een willekeurige lijst met waarden accepteert en een enkele waarde retourneert die alle waarden combineert die door de kaart worden geretourneerd.

MapReduce-architectuur

De functie map() wordt op een gedistribueerde manier over meerdere machines uitgevoerd. De invoergegevens, die gewoonlijk uit een groot bestand (bestand) komen, zijn verdeeld in een reeks M invoerpartities van doorgaans 16 tot 64 megabyte. Deze partities kunnen op verschillende machines worden verwerkt. Verschillende bewerkingen vinden meestal plaats in een MapReduce- aanroep :

• De ingangen worden vervolgens verdeeld in M-partities van ongeveer 16 tot 64 megabytes . Het MapReduce-programma begint zichzelf te instantiëren op de verschillende machines in het cluster. In de regel wordt het aantal instances in de applicaties geconfigureerd.

• Een van de exemplaren van het programma is bijzonder en vervult de rol van "meester". De rest van de kopieën worden "arbeiders" genoemd en krijgen de toewijzing van hun taken van de meester. Er wordt aangenomen dat er een aantal M map()- taken en R reduce() zijn. De "master" is verantwoordelijk voor het verzamelen van "werknemers" in rust (dwz zonder toegewezen taak) en zal er een specifieke map() of reduce() taak aan toewijzen. Een werknemer kan slechts drie toestanden hebben: inactief, werkend, voltooid.

• Een werknemer die aan een specifieke map()- taak is toegewezen, zal de corresponderende partitie als invoer gebruiken. Het zal de paren (sleutel, waarde) ontleden om een ​​nieuw uitgangspaar te creëren, zoals gespecificeerd in de programmering. De sleutel-waardeparen die door de functie map() worden geproduceerd, worden in het geheugen gebufferd.

• Periodiek worden gebufferde sleutel-waardeparen naar de lokale schijf geschreven, verspreid over R-regio's. De regio's van deze sleutel-waardeparen worden doorgegeven aan de master, die verantwoordelijk is voor het omleiden van werknemers die reduce()-taken hebben.

• Wanneer een werker van het type reduce door de "master" op de hoogte wordt gebracht van de locatie van een partitie, gebruikt deze externe oproepen om de informatie te lezen die is opgeslagen op de harde schijven van de verschillende werkers van het type map(). Wanneer een worker van het type reduce() alle tussenliggende gegevens leest, ordent hij de sleutels op zo'n manier dat de gevonden gegevens met dezelfde sleutel bij elkaar worden gegroepeerd. De volgorde is noodzakelijk omdat, als algemene regel, veel verschillende map()-functietoetsen in dezelfde reduce()-functie kunnen gaan. In gevallen waar de hoeveelheid tussenliggende gegevens erg groot is, wordt vaak een externe sortering gebruikt.

• De worker van het type reduce() herhaalt de reeks tussenliggende gesorteerde waarden en doet dit voor elke gevonden unieke sleutel. Het neemt de sleutel en de bijbehorende reeks waarden en geeft deze door aan de functie reduce(). De uitvoer van reduce() wordt toegevoegd aan het MapReduce-uitvoerbestand.

• Wanneer alle map() en reduce() taken zijn voltooid, roept de "master" het gebruikersprogramma op. Op dit punt geeft de MapReduce-aanroep de controle terug aan de code van een gebruiker.

Taken worden geacht te zijn beëindigd wanneer dit besturingselement aan de gebruiker is teruggegeven. De outputs worden gedistribueerd in een compleet bestand, of bij gebrek daaraan worden ze gedistribueerd in R-bestanden. Deze R-bestanden kunnen de invoer zijn van een andere MapReduce of kunnen worden verwerkt door een ander programma dat deze gegevens nodig heeft.

Combiner (lokale aggregators)

In een clusteromgeving is een van de beperkingen te vinden in het transport van grote bestanden tussen computers vanwege hun beperkte bandbreedte . In het MapReduce-framework schrijft de functie map() naar een lokale tekenbuffer, zoals een harde schijf . De informatie die lokaal wordt geschreven, wordt geaggregeerd en geordend door een aggregatorfunctie die verantwoordelijk is voor het uitvoeren van deze bewerking. De geordende waarden zijn van de vorm [k, [v ₁ , v ₂ , v ₃ , ..., v _n ]]. Op deze manier ontvangt de functie reduce() een lijst met waarden die zijn gekoppeld aan een enkele sleutel van de combinator. Aangezien de latentie van het netwerk van computers en hun schijven gewoonlijk groter is dan bij elke andere bewerking, zal elke vermindering van de hoeveelheid tussentijdse gegevens de efficiëntie van de algoritmen verhogen. In MapReduce zorgt elke lokale aggregatie van de tussenresultaten voor een echte verbetering van de globale efficiëntie.

Het is om deze reden dat veel officiële MapReduce-distributies vaak lokale aggregatiebewerkingen bevatten, door functies te gebruiken die gegevens lokaal kunnen aggregeren. De verplaatsing van grote bestanden vermijden of zoveel mogelijk beperken. Ofwel toegevoegd aan map()-functies, of aan lokale aggregators.

Fouttolerantie

Het MapReduce-mechanisme is fouttolerant wanneer een van de werknemers wordt blootgesteld aan een storing. Net als MapReduce is het ontworpen voor processen waarbij grote hoeveelheden gegevens worden gebruikt met honderden of duizenden computers. Hoewel de faalkans klein is, is het heel goed mogelijk dat een (of meerdere) van de werknemers wordt uitgeschakeld, juist door het falen van de machine die het ondersteunde. De "master" pingt elke werknemer periodiek om de status ervan te controleren.

Als er na een bepaalde wachttijd geen reactie is, interpreteert de master dat de werknemer is gedeactiveerd. Elke map()-taak die door de werknemer is voltooid, keert onmiddellijk terug naar de wachtstatus en kan dus in aanmerking komen voor toewijzing aan andere werknemers. Op dezelfde manier wordt elke map() (of reduce)-functie die tijdens de fout aan de gang is, gereset naar inactief en kan worden gekozen voor hertoewijzing.

Voltooide map()-taken worden opnieuw uitgevoerd bij een storing, gedeeltelijk omdat hun uitvoer is opgeslagen op de lokale schijven van de machine die is uitgevallen en daarom als ontoegankelijk wordt beschouwd. Gehele reduce()-taken hoeven niet opnieuw te worden uitgevoerd omdat hun uitvoer is opgeslagen in het globale systeem. Wanneer de map()-taak wordt uitgevoerd door werknemer A en vervolgens door werknemer B (voornamelijk als gevolg van een crash), krijgen in dit geval alle reduce()-taken een melding om gegevens van werknemer A te verwijderen en gegevens van werknemer A. werknemer B te accepteren. Op deze manier is de uitvoering van MapReduce veerkrachtig .

Voorbeelden

In de beschrijving van de voorbeelden van het gebruik van Map Reduce , is het alleen nodig om in detail te beschrijven hoe de bewerkingen map() en reduce() in elk geval worden geïmplementeerd. De literatuur toont herhaalde voorbeelden van woordentellingen in een document, matrixbewerkingen en relationele databasequerybewerkingen.

aantal woorden

Dit Map Reduce- voorbeeld is een proces om de voorkomens van elk woord in een set documenten te tellen :

 kaart ( String naam , String document ) :   
  // sleutel: documentnaam 
// waarde: documentinhoud voor elk woord w in document :  
       
    EmitIntermediate ( w , 1 ); 

De functie map() splitst in dit geval een document in woorden (dat wil zeggen, tokeniseert het ) met behulp van een eenvoudige lexicale analysator en geeft een reeks tupels van de vorm ( key , value ) af waarbij de sleutel het woord is en de waarde is "1". Dat wil zeggen, uit het document "Little House on the Prairie" zou de kaartfunctie teruggeven: ("la", "1"), ("house", "1"), ("of", "1" ), ("de", "1"), ("weide", "1").

 
 reduce ( String woord , Iterator gedeeltelijke Tellingen ) :   
  // woord: een woord 
// partiëleCounts: an [[Iterator (ontwerppatroon)|gedeeltelijke lijst]] om geaggregeerde tellingen uit te voeren int result = 0 ;  
     
  voor elke v in gedeeltelijkeCounts :    
    resultaat += ParseInt ( v );  
  Uitzenden ( resultaat );

Hier wordt elk document in woorden verdeeld en wordt elk woord geteld met de beginwaarde "1" door de kaartfunctie, waarbij het woord als het belangrijkste resultaat wordt gebruikt. Het framework verzamelt alle paren met dezelfde sleutel en wordt in dezelfde Reduce-aanroep ingevoerd, dus deze functie heeft alleen de som van al zijn invoerwaarden nodig om het totale aantal keren dat dat woord voorkomt te vinden. In het bovenstaande voorbeeld ("la", "1") verschijnt twee keer omdat de sleutel "la" twee keer voorkomt, de rest van de sleutels komt maar één keer voor.

Vermenigvuldiging van een matrix met een vector

Lineaire algebra - voorbeelden voor matrixbewerkingen zijn het meest geschikt vanwege de geschiktheid van het raamwerk in deze gevallen. Stel dat we een vierkante matrix M hebben van grootte nxn. We noemen het element in rij i en kolom j m _ij . Stel we hebben een vector v zodat we op positie j het element v _j hebben . Op deze manier wordt het resultaat van de vermenigvuldiging tussen de matrix M en de vector v een vector x met lengte n, zodanig dat het element x _i zodanig is dat:

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

Deze bewerking wordt zonder enig probleem uitgevoerd voor arrays van enkele duizenden elementen, hetgeen enkele miljoenen kost. Het probleem van de berekening komt wanneer het de bedoeling is om het met honderden miljarden te doen . Om deze reden wordt bij de toepassing van M tot Reduce aangenomen dat n in de orde van 10 ¹² is . De functie map() neemt in dit geval een rij i van de matrix en de hele vector v om paren te vormen: (i, m _ij v _j ). Dat wil zeggen, van de vorm (1, m ₁₁ v ₁ ), (1, m ₁₂ v ₂ ), (1, m ₁₃ v ₃ ) ... (1, m _1j v _j ).

 kaart ( Vector rowMatrix , Vector vector ) :   
  // sleutel: i -> vectorindex 
// waarde: product van m<sub>ij</sub> door v<sub>j</sub>. voor elke positie i in vector :  
       
    EmitIntermediate ( i , waarde ); 

De functie reduce() hoeft in dit geval alleen de paren met dezelfde sleutel i te verzamelen en op te tellen.

 
 reduce ( String woord , Iterator gedeeltelijke Tellingen ) :   
  // woord: een woord 
// partiëleCounts: an [[Iterator (ontwerppatroon)|gedeeltelijke lijst]] om geaggregeerde tellingen uit te voeren int result = 0 ;  
     
  voor elke v in gedeeltelijkeCounts :    
    resultaat += ParseInt ( v );  
  Uitzenden ( resultaat );

Gegevensstroom

Het MapReduce-framework is een groot gedistribueerd sorteeralgoritme. De belangrijkste modules die de applicatie definieert zijn:

• een invoerlezer
• Een kaartfunctie
• een partitiefunctie
• Een vergelijkingsfunctie
• Een functie Verminderen
• Een uitgaande schrijver

Invoerlezer

De invoerlezer verdeelt de invoer in 'slices' van de juiste grootte (meestal tussen 64 MB en 128 MB) en het framework wijst een slice toe aan elke kaartfunctie . De invoerlezer leest de gegevens uit stabiele opslag (meestal een Distributed File System ) en genereert sleutel/waarde-paren.

Een veelvoorkomend voorbeeld leest een map vol met tekstbestanden en retourneert elke regel als een record.

Kaartfunctie

De functie Map neemt een reeks sleutel/waarde-paren, verwerkt deze en genereert nul of meer output-sleutel/waarde-paren. De typen input- en outputkaarten kunnen (en zijn) vaak van elkaar verschillen.

Als de toepassing woorden telt, zal de kaartfunctie de regels opsplitsen in woorden en een uitvoersleutel/waarde-paar genereren voor elk woord. Elk uitvoerpaar bevat het woord als de sleutel en het aantal instanties ervan op de regel als de waarde.

Partitie functie

Elke uitgang van de kaartfunctie wordt toegewezen aan een verloopstuk dat de partitiefunctie gebruikt om chunking te genereren. De partitiefunctie ontvangt de sleutel en het aantal verloopstukken en retourneert de index van het gewenste verloopstuk .

Het standaardgedrag is om de hash van de sleutel te krijgen en de hash te gebruiken modulo het aantal reducers . Het is belangrijk om een ​​partitiefunctie te kiezen die een ongeveer gelijkmatige verdeling van gegevens per shard genereert om de balans te behouden, anders kan de MapReduce-bewerking vertragen door te wachten tot langzame reducers (reductoren die zijn toegewezen aan meer gegevens dan in de shard zijn opgenomen) zijn voltooid.

Tussen de toewijzings- en reductiestadia worden de gegevens geschud (parallel gesorteerd / verwisseld tussen knooppunten) om de gegevens te verplaatsen van de shard waar ze zijn geproduceerd naar de shard waar ze worden verkleind. Shuffelen kan in sommige gevallen langer duren dan de verwerking, afhankelijk van de bandbreedte, CPU-snelheden, geproduceerde gegevens en de tijd die wordt verbruikt tussen het in kaart brengen en het verminderen van de verwerking.

Vergelijkingsfunctie

De invoer voor elke reductie wordt verkregen van de machine waarop de kaart is uitgevoerd en besteld met behulp van de vergelijkingsfunctie

Reductiefunctie

Het framework roept de Shrink -functie van de applicatie eenmaal aan voor elke unieke sleutel in de gesorteerde lijst. De Reduce kan de waarden doorlopen die bij die sleutel horen en nul of meer outputs produceren.

In het voorbeeld van het aantal woorden neemt de functie Verminderen de invoerwaarden, telt deze bij elkaar op en genereert een enkele uitvoer voor het woord en de uiteindelijke som.

Sluit

De Output Writer schrijft de uitvoer van de Shrink -functie naar opslagtabellen, meestal een gedistribueerd bestandssysteem.

Gebruikt

Over het algemeen wordt Map Reduce gebruikt bij gelijktijdige computerproblemen waarbij grote datasets betrokken zijn die moeten worden verwerkt door een groot aantal computers (knooppunten), die gezamenlijk clusters worden genoemd (als alle knooppunten zich in hetzelfde lokale netwerk bevinden en dezelfde hardware gebruiken), of naar grids (als de nodes op een gedistribueerde manier worden gedeeld over grote geografische of administratieve gebieden, en die over het algemeen meer heterogene hardware hebben). Parallelle verwerking kan plaatsvinden met behulp van gegevens die zijn opgeslagen in een bestandssysteem (ongestructureerd) of in een database (gestructureerd). Om deze reden wordt het gebruikt in toepassingen met grootschalige gegevens, zoals parallelle toepassingen, webindexering, datamining en wetenschappelijke simulatie.

Zie ook

MapReduce implementaties

• Apache Hadoop
• Bankbasis
• couchdb
• Infinispan
• MongoDB
• Riak