объединить выходные файлы после фазы сокращения

Question

В mapreduce каждая задача сокращения записывает свои выходные данные в файл с именем part-r-nnnnn , где nnnnn - идентификатор раздела, связанный с задачей сокращения. сопоставляет / уменьшает слияние этих файлов?Если да, то как?

Answer 1

Вместо того, чтобы выполнять слияние файлов самостоятельно, вы можете делегировать полное слияние выходных файлов при уменьшении, вызвав:

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

Примечание Объединяет файлы HDFS локально. Убедитесь, что у вас достаточно места на диске перед запуском

Answer 2

Нет, эти файлы не объединяются Hadoop.Количество файлов, которые вы получаете, совпадает с количеством задач сокращения.

Если вам нужно это как ввод для следующей работы, не беспокойтесь о наличии отдельных файлов.Просто укажите весь каталог в качестве входных данных для следующего задания.

Если вам нужны данные вне кластера, то я обычно объединяю их на принимающей стороне, когда извлекаю данные из кластера.Что-то вроде этого:

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

Answer 3

Только для текстовых файлов и HDFS как источника и назначения, используйте следующую команду:

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

Это объединит все файлы в input_hdfs_dir и запишетвывод обратно в HDFS на output_hdfs_file.Помните, что все данные будут возвращены в локальную систему, а затем снова загружены в hdfs, хотя временные файлы не создаются, и это происходит на лету с использованием UNIX pe.

Кроме того, это выиграло 't работать с нетекстовыми файлами, такими как Avro, ORC и т. д.

Для двоичных файлов вы можете сделать что-то вроде этого (если у вас есть таблицы Hive, сопоставленные с каталогами):

insert overwrite table tbl select * from tbl

В зависимости от вашей конфигурации, это может также создать больше, чем файлы.Чтобы создать отдельный файл, либо установите число редукторов равным 1, явно используя mapreduce.job.reduces=1, либо установите свойство улья как hive.merge.mapredfiles=true.

Answer 4

Это функция, которую вы можете использовать для объединения файлов в 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);
}

Answer 5

Файлы part-r-nnnnn генерируются после фазы сокращения, обозначенной как 'r' между ними.Теперь дело в том, что если у вас работает один редуктор, у вас будет выходной файл наподобие part-r-00000.Если число редукторов равно 2, то у вас будет part-r-00000 и part-r-00001 и так далее.Посмотрите, если выходной файл слишком велик, чтобы уместиться в память машины, так как инфраструктура hadoop была разработана для работы на Commodity Machines , тогда файл разбивается.Согласно MRv1, у вас есть ограничение в 20 редукторов для работы над вашей логикой.У вас может быть больше, но то же самое необходимо настроить в файлах конфигурации mapred-site.xml .Говорить о вашем вопросе;вы можете использовать getmerge или установить число редукторов в 1, добавив следующее выражение в код драйвера

job.setNumReduceTasks(1);

Надеюсь, что это ответ на ваш вопрос.

Answer 6

Вы можете запустить дополнительную задачу сопоставления / уменьшения, в которой сопоставление и уменьшение не изменяют данные, а разделитель назначает все данные одному редуктору.

Answer 7

Помимо моего предыдущего ответа у меня есть еще один ответ для вас, который я пробовал несколько минут назад. Вы можете использовать CustomOutputFormat , который выглядит как приведенный ниже код

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

}

Просто взгляните на четвертую строку из последней. Я использовал свое собственное имя в качестве имени выходного файла и протестировал программу с 15 редукторами. Тем не менее, файл остается прежним. Таким образом, получение одного файла вместо двух или более возможно, но при этом очень ясно, что размер выходного файла не должен превышать размер первичной памяти, т. Е. Выходной файл должен помещаться в память обычной машины, в противном случае проблема с разделением выходного файла. Спасибо !!

Answer 8

. Сопоставляет ли карта / уменьшение эти файлы?

Нет. Это не сливается.

Вы можете использовать IdentityReducer для достижения своей цели.

Не выполняет сокращения, записывая все входные значения непосредственно на выход.

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

Записывает все ключи и значения непосредственно в вывод.

Посмотрите на соответствующие сообщения SE:

hadoop: разница между 0 редуктором и идентификатором редуктора?

Answer 9

Если файлы имеют заголовок, от него можно избавиться, выполнив следующее:

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

затем добавьте заголовок вручную для output.csv

Answer 10

Почему бы не использовать подобный скрипт для объединения файлов разделов:

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

store stuff into "/path/to/mergedir"