Ist handgeschriebener Code erforderlich oder kann Hive diese Arbeit erledigen?

Für die meisten Reporting-Aufgaben nein. Ein HiveQL-Join lässt sich in wenigen Zeilen auf dasselbe Shuffle-and-Merge-Muster kompilieren. Manuell geschriebene Jobs lohnen sich nur dann, wenn die Merge-Logik nicht in eine SQL-Klausel passt.

Hadoop MapReduce Join & Counter mit Beispiel

⚡ Intelligente Zusammenfassung

MapReduce-Joins kombinieren zwei große Datensätze über einen gemeinsamen Schlüssel, entweder innerhalb des Mappers oder innerhalb des Reducers, während MapReduce-Zähler Statistiken über den Job sammeln, damit fehlerhafte Datensätze gemessen statt erraten werden können.

🔘 Grundlagen für Mitglieder: Der kleinere der beiden Datensätze wird an jeden Datenknoten verteilt und als Nachschlagetabelle verwendet.
☑️ Kartenseitiger Anschluss: Erfordert, dass jede Eingabe partitioniert, gleichmäßig aufgeteilt und nach dem Join-Schlüssel sortiert wird, bevor die Map-Funktion ausgeführt wird.
✅ Reduzierseiten-Join: Eine Partitionierung ist nicht erforderlich, da jedes Tupel mit demselben Join-Schlüssel im selben Reducer landet.
🧪 Ausgearbeitetes Beispiel: Die Dateien DeptName.txt und DeptStrength.txt werden in HDFS kopiert und über eine JAR-Datei anhand der Dept_ID zusammengeführt.
️ Zählertypen: Fünf integrierte Zählergruppen werden mit jedem Job ausgeliefert, und benutzerdefinierte Zähler werden als deklariert. Java Aufzählung.
⚠️ Verwendung am Zähler: Durch das Erhöhen eines Zählers bei jedem fehlenden oder ungültigen Datensatz werden Datenqualitätsprobleme in eine Zahl im Jobbericht umgewandelt.

Was ist ein Join in MapReduce?

Die MapReduce-Join-Operation dient dazu, zwei große Datensätze zu kombinieren. Dieser Prozess erfordert jedoch umfangreiches Programmieren, um die eigentliche Join-Operation durchzuführen. Der Join zweier Datensätze beginnt mit dem Vergleich ihrer Größe. Ist ein Datensatz kleiner als der andere, wird der kleinere Datensatz auf alle Datenknoten im Cluster verteilt.

Sobald man beitritt MapReduce Bei verteilten Daten verwendet entweder der Mapper oder der Reducer den kleineren Datensatz, um im großen Datensatz nach passenden Datensätzen zu suchen und diese dann zu Ausgabedatensätzen zu kombinieren.

Arten von Joins

Je nachdem, wo der eigentliche Join durchgeführt wird, werden Joins in Hadoop in zwei Arten unterteilt.

Kartenseitiger Anschluss Wenn der Join vom Mapper durchgeführt wird, spricht man von einem Map-seitigen Join. In diesem Fall erfolgt der Join, bevor die Daten von der Map-Funktion verarbeitet werden. Die Eingabe für jede Map muss als Partition vorliegen und sortiert sein. Außerdem muss die Anzahl der Partitionen gleich sein und diese müssen nach dem Join-Schlüssel sortiert sein.
Reduzierseitenverbindung Wird der Join vom Reducer durchgeführt, spricht man von einem Reduce-Side-Join. Für diesen Join ist es nicht erforderlich, dass der Datensatz strukturiert (oder partitioniert) vorliegt. Die Map-Side-Verarbeitung erzeugt den Join-Schlüssel und die entsprechenden Tupel beider Tabellen. Dadurch gelangen alle Tupel mit demselben Join-Schlüssel in denselben Reducer, der dann die Datensätze mit demselben Join-Schlüssel verknüpft.

Der Gesamtprozessablauf von Joins in Hadoop ist im folgenden Diagramm dargestellt.

Prozessablaufdiagramm zum Vergleich eines Map-Side-Joins mit einem Reduce-Side-Join in Hadoop — Arten von Joins in Hadoop MapReduce

Nachdem die beiden Varianten geklärt sind, wird im nächsten Abschnitt ein Reduce-Side-Join für zwei kleine Abteilungsdateien erläutert.

So verbinden Sie zwei DataSets: MapReduce-Beispiel

Es liegen zwei Datensätze in zwei verschiedenen Dateien vor (siehe unten). Der Schlüssel „Dept_ID“ ist in beiden Dateien vorhanden. Ziel ist es, diese Dateien mithilfe von MapReduce Join zu kombinieren.

Erste Eingabedatei mit Auflistung der Abteilungs-IDs und Abteilungsnamen — 1 Datei

Zweite Eingabedatei mit Auflistung der Abteilungs-IDs und der Abteilungsstärkewerte — 2 Datei

Eingang: Der Eingabedatensatz besteht aus einer Textdatei, DeptName.txt und DeptStrength.txt.

Laden Sie hier Eingabedateien herunter

Stellen Sie sicher, dass Sie haben Hadoop installiert. Bevor Sie mit dem eigentlichen Prozess des MapReduce Join-Beispiels beginnen, ändern Sie den Benutzer in 'hduser' (die während der Hadoop-Konfiguration verwendete Benutzer-ID; Sie können zur Benutzer-ID wechseln, die Sie während Ihrer Hadoop-Konfiguration verwendet haben).

su - hduser_

Die Eingabeaufforderung ändert sich zum Hadoop-Konto, wie unten dargestellt.

Schritt 1) Kopieren Sie die ZIP-Datei an den Speicherort Ihrer Wahl

Schritt 2) Dekomprimieren Sie die Zip-Datei

sudo tar -xvf MapReduceJoin.tar.gz

Die ExtracDie Dateinamen werden beim Entpacken des Archivs durch tar eingeblendet.

Schritt 3) Gehen Sie zum Verzeichnis MapReduceJoin/

cd MapReduceJoin/

Schritt 4) Starten Sie Hadoop

$HADOOP_HOME/sbin/start-dfs.sh

$HADOOP_HOME/sbin/start-yarn.sh

Beide Skripte geben die von ihnen gestarteten Daemons aus.

Schritt 5) DeptStrength.txt und DeptName.txt sind die Eingabedateien, die für dieses MapReduce Join-Beispielprogramm verwendet werden.

Diese Dateien müssen kopiert werden nach HDFS mit dem folgenden Befehl:

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

Schritt 6) Führen Sie das Programm mit dem folgenden Befehl aus:

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

Der Befehl wird zuerst wiederholt, und anschließend wird der Fortschritt des Auftrags in der Konsole gemeldet.

Schritt 7) Nach der Ausführung wird die Ausgabedatei (mit dem Namen „part-00000“) im Verzeichnis /output_mapreducejoin auf HDFS gespeichert.

Die Ergebnisse können über die Befehlszeilenschnittstelle angezeigt werden

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

Die Ergebnisse können auch über eine Weboberfläche eingesehen werden.

Wählen Sie nun „Dateisystem durchsuchen“ und navigieren Sie zu /output_mapreducejoin.

Öffnen Sie Teil-r-00000

Ergebnisse werden angezeigt

Anmerkungen: Bitte beachten Sie, dass Sie das Ausgabeverzeichnis /output_mapreducejoin löschen müssen, bevor Sie dieses Programm das nächste Mal ausführen

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

Alternativ können Sie einen anderen Namen für das Ausgabeverzeichnis verwenden.

Joins zeigen Ihnen, wie die Daten aussehen. Zähler, die im nächsten Abschnitt behandelt werden, zeigen Ihnen, wie sich der Prozess verhalten hat, der diese Daten erzeugt hat.

Was ist Counter in MapReduce?

Ein Zähler in MapReduce ist ein Mechanismus zum Sammeln und Auswerten statistischer Informationen über MapReduce-Jobs und -Ereignisse. Zähler speichern die track verschiedene Jobstatistiken in MapReduce, wie die Anzahl der durchgeführten Operationen und deren Fortschritt. Zähler werden zur Problemdiagnose in MapReduce verwendet.

Hadoop-Zähler ähneln dem Einfügen einer Protokollnachricht in den Code für eine Karte oder Reduzierung. Diese Informationen könnten für die Diagnose eines Problems bei der MapReduce-Auftragsverarbeitung nützlich sein.

Typischerweise werden diese Zähler in Hadoop in einem Programm (Map oder Reduce) definiert und während der Ausführung inkrementiert, wenn ein bestimmtes Ereignis oder eine bestimmte Bedingung (spezifisch für diesen Zähler) eintritt. Ein sehr gutes Anwendungsbeispiel für Hadoop-Zähler ist … track gültige und ungültige Datensätze aus einem Eingabedatensatz.

Arten von MapReduce-Zählern

Grundsätzlich gibt es 2 Arten von MapReduce-Zählern.

Hadoop-integrierte Zähler: Es gibt einige integrierte Hadoop-Zähler, die pro Job vorhanden sind. Nachfolgend finden Sie integrierte Zählergruppen.
- MapReduce-Aufgabenzähler — Sammelt während der Ausführungszeit aufgabenspezifische Informationen (z. B. Anzahl der Eingabedatensätze).
- Dateisystemzähler — Sammelt Informationen wie die Anzahl der von einem Prozess gelesenen oder geschriebenen Bytes.
- FileInputFormat-Zähler — Sammelt Informationen über eine Anzahl von Bytes, die über FileInputFormat gelesen werden.
- FileOutputFormat-Zähler — Sammelt Informationen über eine Anzahl von Bytes, die mit FileOutputFormat geschrieben wurden.
- Stellenanzeigen — Diese Zähler erfassen auftragsweite Statistiken, wie zum Beispiel die Anzahl der für einen Auftrag gestarteten Aufgaben.
Benutzerdefinierte Zähler: Zusätzlich zu den integrierten Zählern kann ein Benutzer mithilfe ähnlicher Funktionen von Programmiersprachen eigene Zähler definieren. Zum Beispiel in Java, wird ein 'Enum' verwendet, um benutzerdefinierte Zähler zu definieren.

💡 Versionshinweis: Die Stellenanzeigen wurden von der Jobzentrale geführt.TracUnter MRv1 gehört diese Rolle zum MapReduce ApplicationMaster. Daher bleiben die Zählernamen erhalten, aber die Komponente, die sie meldet, hat sich geändert.

Ein Auftrag kann keine unbegrenzte Anzahl von Zählern deklarieren. mapreduce.job.counters.max Die Einstellung begrenzt die Gesamtzahl pro Auftrag standardmäßig auf 120, und ein Auftrag, der mehr deklariert, schlägt mit einer Fehlermeldung fehl. LimitExceededExceptionZähler sind also eher für eine Handvoll aggregierter Signale als für die Zählung einzelner Tasten gedacht.

Beispiel für Zähler

Eine Beispiel-MapClass mit Zählern zur Ermittlung fehlender und ungültiger Werte. Die in diesem Tutorial verwendete Eingabedatendatei ist eine CSV-Datei mit dem Namen 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"));
        }
    }
}

Der obige Codeausschnitt zeigt eine Beispielimplementierung von Zählern in Hadoop MapReduce.

Dabei steht: Verkaufstheken ist ein Zähler, der mit ' definiert istenum'. Es dient dazu, fehlende und ungültige Eingabedatensätze zu zählen.

Im Code-Snippet, wenn 'LandWenn ein Feld die Länge Null hat, fehlt sein Wert und der entsprechende Zähler SalesCounters.MISSING wird inkrementiert.

Als nächstes, wenn 'VertriebWenn ein Feld mit einem einfachen „ beginnt, wird der Datensatz als ungültig betrachtet. Dies wird durch Erhöhen des Zählers SalesCounters.INVALID angezeigt.

💡 API-Hinweis: Der obige Ausschnitt verwendet das Original. org.apache.hadoop.mapred API, wo MapReduceBase, hat das Mapper Schnittstelle, OutputCollector und Reporter erscheinen separat. Der aktuelle Code ist gegen org.apache.hadoop.mapreduce, wo ein einzelnes Context ersetzt den Sammler und den Melder, und ein Zähler wird erhöht um context.getCounter(SalesCounters.MISSING).increment(1)Das Thekenkonzept ist in beiden Fällen identisch.

Häufig gestellte Fragen

Wählen Sie die Mapper-Variante, wenn eine Seite klein genug ist, um sie auf jedem Knoten im Speicher zu halten, da hierbei das Mischen vollständig entfällt. Wählen Sie die Reducer-Variante, wenn beide Seiten groß oder unsortiert sind, und akzeptieren Sie die zusätzlichen Netzwerkkosten.

Die Modelle lernen aus der Historie vergangener Jobs, um Laufzeiten vorherzusagen, Splitgrößen und Reducer-Anzahlen zu empfehlen und Abweichungen von den Zählerwerten zu erkennen. Sie kennzeichnen außerdem Jobs, deren Zähler für übersprungene Datensätze oder fehlgeschlagene Aufgaben außerhalb des normalen Bereichs für diese Pipeline liegen.

Copilot erzeugt plausible Mapper- und Reducer-Grundgerüste, vermischt aber das alte mapred-Paket mit dem neueren mapreduce-Paket in einer Klasse, was zu einem Kompilierungsfehler führt. Korrigieren Sie die Importe und Methodensignaturen, bevor Sie der Logik vertrauen.

Dieser Mechanismus sorgt dafür, dass die kleinere Datei vor dem Start der Aufgaben an jeden Knoten gesendet wird. Jeder Mapper lädt diese Kopie dann in eine Hash-Map und sucht lokal nach Übereinstimmungen, wodurch ein Join auf Mapper-Seite möglich wird.

Sie werden in der Konsolenzusammenfassung ausgegeben, wenn der Job abgeschlossen ist, sind auf den Webseiten des Jobverlaufs und des Ressourcenmanagers einsehbar und können programmatisch aus dem Jobobjekt gelesen werden, sodass ein Treiber sie überprüfen und einen fehlerhaften Lauf abbrechen kann.

Ein einzelnes Context-Objekt. Es übernimmt die Arbeit, die OutputCollector und Reporter zuvor zwischen ihnen aufgeteilt haben, sodass Ausgaben geschrieben und Zähler über denselben Handle inkrementiert werden, der an die map-Methode übergeben wird.

Für die meisten Reportagearbeiten: Nein. HiveQL join Lässt sich in wenigen Zeilen auf dasselbe Shuffle-and-Merge-Muster reduzieren. Manuell geschriebene Jobs lohnen sich, wenn die Merge-Logik nicht in eine SQL-Klausel passt.

Hadoop schreibt nicht in ein bereits existierendes Ausgabeverzeichnis, um zu verhindern, dass fertige Ergebnisse überschrieben werden. Löschen Sie das Verzeichnis daher zunächst rekursiv oder geben Sie beim nächsten Durchlauf einen anderen Ausgabepfad an.