Hadoop MapReduce Gå med och räkna med exempel

Vad är Join i MapReduce?

MapReduce Join-operationen används för att kombinera två stora datamängder. Denna process innebär dock att man skriver mycket kod för att utföra själva kopplingsoperationen. Att koppla ihop två datamängder börjar med att jämföra storleken på varje datamängd. Om en datamängd är mindre jämfört med den andra datamängden distribueras den mindre datamängden till varje datanod i klustret.

När man väl har gått med MapReduce distribueras använder antingen Mapper eller Reducer den mindre datamängden för att utföra en sökning efter matchande poster från den stora datamängden och kombinerar sedan dessa poster för att bilda utdataposter.

Typer av gå med

Beroende på var den faktiska kopplingen utförs klassificeras kopplingar i Hadoop i två typer.

1. Kartsidans koppling — När kopplingen utförs av mappern kallas det en map-side join. I den här typen utförs kopplingen innan data faktiskt förbrukas av map-funktionen. Det är obligatoriskt att indata till varje map är i form av en partition och är i sorterad ordning. Det måste också finnas lika många partitioner och de måste sorteras efter join-nyckeln.
2. Reduce-side join — När kopplingen utförs av reduceraren kallas det en reduce-side-koppling. Det finns inget behov av att ha en dataset i strukturerad (eller partitionerad) form i denna koppling. Här genererar map-side-bearbetningen kopplingsnyckeln och motsvarande tupler för båda tabellerna. Som en effekt av denna bearbetning hamnar alla tupler med samma kopplingsnyckel i samma reducerare, som sedan kopplar samman posterna med samma kopplingsnyckel.

Ett övergripande processflöde av kopplingar i Hadoop visas i diagrammet nedan.

Med de två varianterna avklarade går nästa avsnitt igenom en reduce-side join på två små avdelningsfiler.

Hur man går med i två datauppsättningar: MapReduce Exempel

Det finns två datamängder i två olika filer (visas nedan). Nyckeln Dept_ID är gemensam i båda filerna. Målet är att använda MapReduce Join för att kombinera dessa filer.

Ingång: Indatamängden är en txt-fil, DeptName.txt och DeptStrength.txt

Ladda ner indatafiler härifrån

Se till att du har Hadoop installerat. Innan du börjar med MapReduce Join-exemplet i den faktiska processen, ändra användaren till 'hduser' (id som användes under Hadoop-konfigurationen, du kan byta till det användar-ID som användes under din Hadoop-konfiguration).

su - hduser_

Prompten ändras till Hadoop-kontot, som visas nedan.

Steg 1) Kopiera zip-filen till den plats du väljer

Steg 2) Packa upp zip-filen

sudo tar -xvf MapReduceJoin.tar.gz

Exettracted-filnamn rullar förbi medan tar packar upp arkivet.

Steg 3) Gå till katalogen MapReduceJoin/

cd MapReduceJoin/

Steg 4) Starta Hadoop

$HADOOP_HOME/sbin/start-dfs.sh

$HADOOP_HOME/sbin/start-yarn.sh

Båda skripten skriver ut de daemoner de öppnar.

Steg 5) DeptStrength.txt och DeptName.txt är indatafilerna som används för detta MapReduce Join-exempelprogram.

Dessa filer måste kopieras till HDFS med hjälp av kommandot nedan-

$HADOOP_HOME/bin/hdfs dfs -copyFromLocal DeptStrength.txt DeptName.txt /

Steg 6) Kör programmet med kommandot nedan-

$HADOOP_HOME/bin/hadoop jar MapReduceJoin.jar MapReduceJoin/JoinDriver/DeptStrength.txt /DeptName.txt /output_mapreducejoin

Kommandot upprepas först, och jobbet rapporterar sedan sina förlopp i konsolen.

Steg 7) Efter körning lagras utdatafilen (med namnet 'part-00000') i katalogen /output_mapreducejoin på HDFS.

Resultaten kan ses med hjälp av kommandoradsgränssnittet

$HADOOP_HOME/bin/hdfs dfs -cat /output_mapreducejoin/part-00000

Resultaten kan också ses via ett webbgränssnitt som-

Välj nu "Bläddra i filsystemet" och navigera upp till /output_mapreducejoin

Öppna del-r-00000

Resultaten visas

OBS: Observera att innan du kör det här programmet nästa gång måste du ta bort utdatakatalogen /output_mapreducejoin

$HADOOP_HOME/bin/hdfs dfs -rm -r /output_mapreducejoin

Alternativet är att använda ett annat namn för utdatakatalogen.

Kopplingar visar hur informationen ser ut. Räknare, som behandlas härnäst, visar hur jobbet som producerade den betedde sig.

Vad är Counter i MapReduce?

En räknare i MapReduce är en mekanism som används för att samla in och mäta statistisk information om MapReduce-jobb och händelser. Räknare håller track av olika jobbstatistik i MapReduce, såsom antalet utförda operationer och operationens förlopp. Räknare används för problemdiagnostik i MapReduce.

Hadoop-räknare liknar att lägga ett loggmeddelande i koden för en karta eller reducera. Denna information kan vara användbar för diagnos av ett problem i MapReduce-jobbbearbetning.

Vanligtvis definieras dessa räknare i Hadoop i ett program (mappa eller reducera) och ökas under körning när en viss händelse eller ett villkor (specifikt för den räknaren) inträffar. En mycket bra tillämpning av Hadoop-räknare är att track giltiga och ogiltiga poster från en indatauppsättning.

Typer av MapReduce-räknare

Det finns i princip två typer av MapReduce-räknare.

1. Inbyggda räknare i Hadoop: Det finns några inbyggda Hadoop-räknare som finns per jobb. Nedan finns inbyggda räknargrupper-
   • MapReduce Task Counters — Samlar in uppgiftsspecifik information (t.ex. antal indataposter) under sin körningstid.
   • Filsystemräknare — Samlar in information som antal byte som lästs eller skrivits av en uppgift.
   • FileInputFormat-räknare — Samlar in information om ett antal byte som lästs via FileInputFormat.
   • FileOutputFormat-räknare — Samlar in information om ett antal byte skrivna via FileOutputFormat.
   • Jobbräknare — Dessa räknare registrerar jobbomfattande statistik, till exempel antalet uppgifter som startats för ett jobb.
2. Användardefinierade räknare: Förutom inbyggda räknare kan en användare definiera sina egna räknare med hjälp av liknande funktioner som programmeringsspråk erbjuder. Till exempel, i Java, en 'enum' används för att definiera användardefinierade räknare.

💡 Versionsnotering: jobbräknare underhålls av JobTracker under MRv1. På YARN tillhör den rollen MapReduce ApplicationMaster, så räknarnamnen finns kvar men komponenten som rapporterar dem har ändrats.

Ett jobb kan inte deklarera ett obegränsat antal räknare. mapreduce.job.counters.max inställningen begränsar totalsumman per jobb till 120 som standard, och ett jobb som deklarerar fler misslyckas med en LimitExceededException, så räknare är avsedda för en handfull aggregerade signaler snarare än räkningar per nyckel.

Exempel på räknare

Ett exempel på en MapClass med räknare för att räkna antalet saknade och ogiltiga värden. Indatafil som används i den här handledningen. Vår indatauppsättning är en CSV-fil, SalesJan2009.csv.

public static class MapClass
            extends MapReduceBase
            implements Mapper<LongWritable, Text, Text, Text>
{
    static enum SalesCounters { MISSING, INVALID };
    public void map ( LongWritable key, Text value,
                 OutputCollector<Text, Text> output,
                 Reporter reporter) throws IOException
    {
        
        //Input string is split using ',' and stored in 'fields' array
        String fields[] = value.toString().split(",", -20);
        //Value at 4th index is country. It is stored in 'country' variable
        String country = fields[4];
        
        //Value at 8th index is sales data. It is stored in 'sales' variable
        String sales = fields[8];
      
        if (country.length() == 0) {
            reporter.incrCounter(SalesCounters.MISSING, 1);
        } else if (sales.startsWith("\"")) {
            reporter.incrCounter(SalesCounters.INVALID, 1);
        } else {
            output.collect(new Text(country), new Text(sales + ",1"));
        }
    }
}

Kodavsnittet ovan visar ett exempel på implementering av räknare i Hadoop MapReduce.

Här, Försäljningsräknare är en räknare definierad med hjälp av 'uppräkning'. Den används för att räkna SAKNADE och OGILTIGDA indataposter.

I kodavsnittet, om 'land'-fältet har längden noll så saknas dess värde och därmed ökas motsvarande räknare SalesCounters.MISSING.

Nästa, om 'försäljning'-fältet börjar med ett “, då anses posten vara OGILTIG. Detta indikeras genom att räknaren SalesCounters.INVALID ökar.

💡 API-anmärkning: utdraget ovan använder originalet org.apache.hadoop.mapred API, där MapReduceBase, den Mapper gränssnitt, OutputCollector och Reporter visas separat. Nuvarande kod är skriven mot org.apache.hadoop.mapreduce, där en enda Context ersätter samlaren och rapportören, och en räknare ökas med context.getCounter(SalesCounters.MISSING).increment(1)Räknarekonceptet är identiskt i båda.

Vanliga frågor