Hadoop MapReduce Deltag og tæller med eksempel

Hvad er Join i MapReduce?

MapReduce Join-operationen bruges til at kombinere to store datasæt. Denne proces involverer dog at skrive en masse kode for at udføre selve join-operationen. Joining af to datasæt starter med at sammenligne størrelsen af ​​hvert datasæt. Hvis det ene datasæt er mindre sammenlignet med det andet datasæt, distribueres det mindre datasæt til alle datanoder i klyngen.

Når man først er med KortReducer distribueres, bruger enten Mapper eller Reducer det mindre datasæt til at udføre et opslag efter matchende poster fra det store datasæt og kombinerer derefter disse poster for at danne outputposter.

Typer af tilslutning

Afhængigt af hvor den faktiske join udføres, klassificeres joins i Hadoop i to typer.

1. Kortsidet join — Når join'et udføres af mapperen, kaldes det et map-side join. I denne type udføres join'et, før dataene rent faktisk forbruges af map-funktionen. Det er obligatorisk, at inputtet til hvert map er i form af en partition og er i sorteret rækkefølge. Der skal også være et lige antal partitioner, og det skal sorteres efter join-nøglen.
2. Reducer-side join — Når join'en udføres af reducer'en, kaldes det en reduce-side join. Der er ingen nødvendighed i denne join at have et datasæt i en struktureret form (eller partitioneret). Her udsender map-side behandling join-nøglen og de tilsvarende tupler fra begge tabeller. Som en effekt af denne behandling falder alle tupler med den samme join-nøgle ind i den samme reducer, som derefter joiner posterne med den samme join-nøgle.

Et overordnet procesflow af joinforbindelser i Hadoop er afbildet i diagrammet nedenfor.

Når de to varianter er klare, gennemgår næste afsnit en reduce-side join på to små afdelingsfiler.

Sådan forbinder du to datasæt: MapReduce Eksempel

Der er to datasæt i to forskellige filer (vist nedenfor). Nøglen Dept_ID er fælles i begge filer. Målet er at bruge MapReduce Join til at kombinere disse filer.

Input: Inputdatasættet er en txt-fil, DeptName.txt & DeptStrength.txt

Download inputfiler herfra

Sørg for, at du har Hadoop installeret. Før du starter med den faktiske MapReduce Join-eksempelproces, skal du ændre brugeren til 'hduser' (id'et blev brugt under Hadoop-konfigurationen. Du kan skifte til det bruger-id, der blev brugt under din Hadoop-konfiguration).

su - hduser_

Prompten ændrer sig til Hadoop-kontoen, som vist nedenfor.

Trin 1) Kopier zip-filen til den ønskede placering

Trin 2) Udkomprimer zip-filen

sudo tar -xvf MapReduceJoin.tar.gz

Eksmandentracted-filnavne ruller forbi, mens tar udpakker arkivet.

Trin 3) Gå til mappen MapReduceJoin/

cd MapReduceJoin/

Trin 4) Start Hadoop

$HADOOP_HOME/sbin/start-dfs.sh

$HADOOP_HOME/sbin/start-yarn.sh

Begge scripts udskriver de dæmoner, de åbner.

Trin 5) DeptStrength.txt og DeptName.txt er inputfilerne, der bruges til dette MapReduce Join-eksempelprogram.

Disse filer skal kopieres til HDFS ved hjælp af nedenstående kommando-

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

Trin 6) Kør programmet ved at bruge nedenstående kommando-

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

Kommandoen afspilles først, og jobbet rapporterer derefter sin status på konsollen.

Trin 7) Efter udførelse gemmes outputfilen (med navnet 'part-00000') i mappen /output_mapreducejoin på HDFS.

Resultater kan ses ved hjælp af kommandolinjegrænsefladen

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

Resultater kan også ses via en webgrænseflade som-

Vælg nu 'Gennemse filsystemet' og naviger op til /output_mapreducejoin

Åben del-r-00000

Resultater vises

BEMÆRK VENLIGST: Bemærk venligst, at før du kører dette program til næste gang, skal du slette output-mappen /output_mapreducejoin

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

Alternativet er at bruge et andet navn til output-mappen.

Joins fortæller dig, hvordan dataene ser ud. Tællere, som vi gennemgår nedenfor, fortæller dig, hvordan det job, der producerede dem, opførte sig.

Hvad er tæller i MapReduce?

En tæller i MapReduce er en mekanisme, der bruges til at indsamle og måle statistiske oplysninger om MapReduce-job og -hændelser. Tællere holder styr på track af forskellige jobstatistikker i MapReduce, såsom antallet af forekomne operationer og operationens status. Tællere bruges til problemdiagnose i MapReduce.

Hadoop-tællere ligner at sætte en logmeddelelse i koden til et kort eller reducere. Disse oplysninger kan være nyttige til diagnosticering af et problem i MapReduce-jobbehandling.

Typisk er disse tællere i Hadoop defineret i et program (kortlægges eller reduceres) og øges under udførelsen, når en bestemt hændelse eller betingelse (specifik for den pågældende tæller) opstår. En meget god anvendelse af Hadoop-tællere er at track gyldige og ugyldige poster fra et inputdatasæt.

Typer af MapReduce-tællere

Der er grundlæggende 2 typer MapReduce-tællere

1. Indbyggede Hadoop-tællere: Der er nogle indbyggede Hadoop-tællere, som findes pr. job. Nedenfor er indbyggede tællergrupper-
   • KortReducer opgavetællere — Indsamler opgavespecifikke oplysninger (f.eks. antal inputposter) i løbet af dens udførelsestid.
   • Filsystemtællere — Indsamler oplysninger som antallet af bytes, der er læst eller skrevet af en opgave.
   • FileInputFormat-tællere — Indsamler information om et antal bytes læst via FileInputFormat.
   • FileOutputFormat-tællere — Indsamler information om et antal bytes skrevet via FileOutputFormat.
   • Jobtællere — Disse tællere registrerer statistikker for hele jobbet, f.eks. antallet af opgaver, der er startet for et job.
2. Brugerdefinerede tællere: Ud over indbyggede tællere kan en bruger definere sine egne tællere ved hjælp af lignende funktioner, der tilbydes af programmeringssprog. For eksempel i Java, en 'enum' bruges til at definere brugerdefinerede tællere.

💡 Versionsnote: Jobtællere blev vedligeholdt af JobTracker under MRv1. På YARN tilhører den rolle MapReduce ApplicationMaster, så tællernavnene bevares, men komponenten, der rapporterer dem, er ændret.

Et job kan ikke deklarere et ubegrænset antal tællere. mapreduce.job.counters.max Indstillingen begrænser det samlede antal pr. job til 120 som standard, og et job, der erklærer flere, mislykkes med en LimitExceededException, så tællere er beregnet til en håndfuld aggregerede signaler i stedet for optællinger pr. nøgle.

Eksempel på tællere

Et eksempel på en MapClass med tællere til at tælle antallet af manglende og ugyldige værdier. Inputdatafil brugt i denne vejledning. Vores inputdatasæt er 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"));
        }
    }
}

Ovenstående kodestykke viser et eksempel på implementering af tællere i Hadoop MapReduce.

Her, Salgstællere er en tæller defineret ved hjælp af 'enum'. Den bruges til at tælle MANGLENDE og UGYLDIGE inputposter.

I kodestykket, hvis 'land'-feltet har en længde på nul, mangler dets værdi, og derfor øges den tilsvarende tæller SalesCounters.MISSING.

Dernæst, hvis 'salg'-feltet starter med et “, betragtes posten som UGYLDIG. Dette angives ved at øge tælleren SalesCounters.INVALID.

💡 API-bemærkning: uddraget ovenfor bruger originalen org.apache.hadoop.mapred API, hvor MapReduceBase, Mapper grænseflade, OutputCollector og Reporter vises separat. Den nuværende kode er skrevet imod org.apache.hadoop.mapreduce, hvor en enkelt Context erstatter opsamleren og reporteren, og en tæller øges med context.getCounter(SalesCounters.MISSING).increment(1)Tællerkonceptet er identisk i begge.

Ofte Stillede Spørgsmål