Zusammenführen von Ausgabedateien nach der Reduzierungsphase

In mapreduce schreibt jede Reduzierungsaufgabe ihre Ausgabe in eine Datei mit dem Namen part-r-nnnnn, wobei nnnnn eine Partitions-ID ist, die der Reduktionsaufgabe zugeordnet ist. Führt Map / Reduce diese Dateien zusammen? Wenn ja, wie?

Anstatt das Zusammenführen von Dateien selbst durchzuführen, können Sie das gesamte Zusammenführen der reduzierten Ausgabedateien delegieren, indem Sie Folgendes aufrufen:

hadoop fs -getmerge /output/dir/on/hdfs/ /desired/local/output/file.txt

Hinweis Hiermit werden die HDFS-Dateien lokal kombiniert. Stellen Sie vor dem Ausführen sicher, dass genügend Speicherplatz vorhanden ist

Nein, diese Dateien werden von Hadoop nicht zusammengeführt. Die Anzahl der Dateien, die Sie erhalten, entspricht der Anzahl der Reduzierungsaufgaben.

Wenn Sie dies als Eingabe für einen nächsten Job benötigen, müssen Sie sich keine Gedanken über separate Dateien machen. Geben Sie einfach das gesamte Verzeichnis als Eingabe für den nächsten Job an.

Wenn Sie die Daten außerhalb des Clusters benötigen, füge ich sie normalerweise am empfangenden Ende zusammen, wenn ich die Daten aus dem Cluster ziehe.

Dh so etwas:

hadoop fs -cat /some/where/on/hdfs/job-output/part-r-* > TheCombinedResultOfTheJob.txt

Mit dieser Funktion können Sie Dateien in HDFS zusammenführen

public boolean getMergeInHdfs(String src, String dest) throws IllegalArgumentException, IOException {
    FileSystem fs = FileSystem.get(config);
    Path srcPath = new Path(src);
    Path dstPath = new Path(dest);

    // Check if the path already exists
    if (!(fs.exists(srcPath))) {
        logger.info("Path " + src + " does not exists!");
        return false;
    }

    if (!(fs.exists(dstPath))) {
        logger.info("Path " + dest + " does not exists!");
        return false;
    }
    return FileUtil.copyMerge(fs, srcPath, fs, dstPath, false, config, null);
}

Verwenden Sie den folgenden Befehl, um nur Textdateien und HDFS als Quelle und Ziel zu verwenden:

hadoop fs -cat /input_hdfs_dir/* | hadoop fs -put - /output_hdfs_file

Dadurch werden alle Dateien in verkettet input_hdfs_dirund die Ausgabe wird unter an HDFS zurückgeschrieben output_hdfs_file. Denken Sie daran, dass alle Daten auf das lokale System zurückgebracht und dann erneut auf hdfs hochgeladen werden, obwohl keine temporären Dateien erstellt werden und dies im laufenden Betrieb mit UNIX pe geschieht.

Dies funktioniert auch nicht mit Nicht-Textdateien wie Avro, ORC usw.

Bei Binärdateien können Sie Folgendes tun (wenn Sie Hive-Tabellen in den Verzeichnissen zugeordnet haben):

insert overwrite table tbl select * from tbl

Abhängig von Ihrer Konfiguration können dadurch auch mehr als nur Dateien erstellt werden. Um eine einzelne Datei zu erstellen, setzen Sie entweder die Anzahl der Reduzierungen explizit auf 1 mapreduce.job.reduces=1oder setzen Sie die Eigenschaft hive auf hive.merge.mapredfiles=true.

Die part-r-nnnnn-Dateien werden nach der dazwischen mit 'r' bezeichneten Reduzierungsphase generiert. Tatsache ist nun, wenn Sie einen Reduzierer haben, haben Sie eine Ausgabedatei wie part-r-00000. Wenn die Anzahl der Reduzierungen 2 beträgt, haben Sie Teil-r-00000 und Teil-r-00001 und so weiter. Wenn die Ausgabedatei zu groß ist, um in den Maschinenspeicher zu passen, da das Hadoop-Framework für die Ausführung auf Commodity-Maschinen konzipiert wurde , wird die Datei aufgeteilt. Gemäß MRv1 haben Sie ein Limit von 20 Reduzierern, um an Ihrer Logik zu arbeiten. Möglicherweise haben Sie mehr, aber das gleiche muss in den Konfigurationsdateien mapred-site.xml angepasst werden . Über Ihre Frage sprechen; Sie können entweder getmerge verwenden oder die Anzahl der Reduzierungen auf 1 setzen, indem Sie die folgende Anweisung in den Treibercode einbetten

job.setNumReduceTasks(1);

Hoffe das beantwortet deine Frage.

Sie können eine zusätzliche Map / Reduce-Aufgabe ausführen, bei der Map und Reduce die Daten nicht ändern und der Partitionierer alle Daten einem einzelnen Reduzierer zuweist.

Neben meiner vorherigen Antwort habe ich noch eine Antwort für Sie, die ich vor einigen Minuten versucht habe. Sie können CustomOutputFormat verwenden, das wie der unten angegebene Code aussieht

public class VictorOutputFormat extends FileOutputFormat<StudentKey,PassValue> {

    @Override
    public RecordWriter<StudentKey,PassValue> getRecordWriter(
            TaskAttemptContext tac) throws IOException, InterruptedException {
        //step 1: GET THE CURRENT PATH
        Path currPath=FileOutputFormat.getOutputPath(tac);

        //Create the full path
        Path fullPath=new Path(currPath,"Aniruddha.txt");

        //create the file in the file system
        FileSystem fs=currPath.getFileSystem(tac.getConfiguration());
        FSDataOutputStream fileOut=fs.create(fullPath,tac);
        return new VictorRecordWriter(fileOut);
    }

}

Schauen Sie sich einfach die vierte Zeile der letzten an. Ich habe meinen eigenen Namen als Ausgabedateinamen verwendet und das Programm mit 15 Reduzierern getestet. Trotzdem bleibt die Datei gleich. Es ist also möglich, eine einzelne Ausgabedatei anstelle von zwei oder mehr zu erhalten, wobei jedoch noch nicht klar ist, dass die Größe der Ausgabedatei die Größe des Primärspeichers nicht überschreiten darf, dh die Ausgabedatei muss in den Speicher der Commodity-Maschine passen, da dies sonst der Fall sein könnte Ein Problem mit der Aufteilung der Ausgabedatei. Vielen Dank!!

Warum nicht ein Pig-Skript wie dieses zum Zusammenführen von Partitionsdateien verwenden:

stuff = load "/path/to/dir/*"

store stuff into "/path/to/mergedir"

Wenn die Dateien einen Header haben, können Sie ihn folgendermaßen entfernen:

hadoop fs -cat /path/to/hdfs/job-output/part-* | grep -v "header" > output.csv

Fügen Sie dann den Header manuell für output.csv hinzu

. Führt Map / Reduce diese Dateien zusammen?

Nein, es wird nicht zusammengeführt.

Sie können IdentityReducer verwenden , um Ihr Ziel zu erreichen.

Führt keine Reduzierung durch und schreibt alle Eingabewerte direkt in die Ausgabe.

public void reduce(K key,
                   Iterator<V> values,
                   OutputCollector<K,V> output,
                   Reporter reporter)
            throws IOException

Schreibt alle Schlüssel und Werte direkt in die Ausgabe.

Schauen Sie sich verwandte SE-Beiträge an:

hadoop: Unterschied zwischen 0 Reduzierer und Identitätsreduzierer?