Scal pliki wyjściowe po fazie redukcji

Nie, Te pliki nie są łączone przez Hadoop. Liczba otrzymywanych plików jest taka sama jak liczba zadań redukcji.

Jeśli potrzebujesz tego jako danych wejściowych do następnego zadania, nie martw się o osobne pliki. Po prostu określ cały katalog jako dane wejściowe dla następnego zadania.

Jeśli potrzebujesz danych poza klastrem, Zwykle scalam je na końcu odbiorczym podczas ściągania danych z klastra.

Czyli coś takiego:

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

To funkcja, której możesz użyć do scalania plików w HDFS

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);
}

Tylko dla plików tekstowych i HDFS jako źródła i miejsca docelowego, użyj poniższego polecenia:

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

To połączy wszystkie pliki w input_hdfs_dir i zapisze wyjście z powrotem do HDFS w output_hdfs_file. Należy pamiętać, że wszystkie dane zostaną przeniesione z powrotem do systemu lokalnego, a następnie ponownie przesłane do systemu hdfs, chociaż nie są tworzone żadne pliki tymczasowe i dzieje się to w locie za pomocą Unix pe.

Również nie będzie to działać z plikami nietekstowymi, takimi jak Avro, ORC itp.

Dla pliki binarne, możesz zrobić coś takiego (jeśli masz tabele Hive odwzorowane w katalogach):

insert overwrite table tbl select * from tbl

W zależności od konfiguracji, może to również tworzyć więcej niż pliki. Aby utworzyć pojedynczy plik, Ustaw liczbę reduktorów na 1 jawnie używając mapreduce.job.reduces=1 lub ustaw właściwość hive jako hive.merge.mapredfiles=true.

Można uruchomić dodatkowe zadanie map/reduce, gdzie map i reduce nie zmieniają danych, a partycjoner przypisuje wszystkie dane do jednego reduktora.

Pliki part-r-NNNN są generowane po fazie redukcji oznaczonej przez " R " pomiędzy. Faktem jest, że jeśli masz uruchomiony jeden reduktor, będziesz miał plik wyjściowy, taki jak part-r-00000. Jeśli liczba reduktorów wynosi 2, będziesz miał część-r-00000 i część-r-00001 i tak dalej. Jeśli plik wyjściowy jest zbyt duży, aby zmieścić się w pamięci maszyny, ponieważ Framework hadoop został zaprojektowany do pracy na Commodity Machines , to plik zostanie podzielony. Jak na MRv1, ty masz limit 20 reduktorów do pracy nad swoją logiką. Możesz mieć więcej, ale to samo musi być dostosowane w plikach konfiguracyjnych mapred-site.xml . Mówiąc o swoim pytaniu; możesz użyć getmerge lub możesz ustawić liczbę reduktorów na 1, osadzając następującą instrukcję w kodzie sterownika

job.setNumReduceTasks(1);

Oprócz mojej poprzedniej odpowiedzi mam dla Ciebie jeszcze jedną odpowiedź, którą próbowałem kilka minut temu. Możesz użyć CustomOutputFormat który wygląda jak kod podany poniżej

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);
    }

}

Dlaczego nie użyć skryptu świni takiego jak ten do scalania plików partycji:

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

store stuff into "/path/to/mergedir"

Jeśli pliki mają nagłówek, możesz się go pozbyć, wykonując to:

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

Następnie dodaj nagłówek ręcznie dla wyjścia.csv

. Czy map/reduce łączy te pliki?

Możesz użyć IdentityReducer , aby osiągnąć swój cel.

Nie wykonuje redukcji, zapisując wszystkie wartości wejściowe bezpośrednio na wyjście.

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

Zapisuje wszystkie klucze i wartości bezpośrednio do wyjścia.

Zobacz podobne posty SE:

Hadoop: różnica między reduktorem 0 a reduktorem tożsamości?