Map-Reduce не может доставить ожидаемые разделенные файлы - PullRequest
Новая
       75

Map-Reduce не может доставить ожидаемые разделенные файлы

0 голосов
/ 14 сентября 2018

В работе Map-Reduce я использую пять разных файлов, где в моем наборе данных содержатся значения в двух категориях P и I. После того, как I определенные значения найдены, я передаю их в файл I-part-r-00000 и, соответственно, для P. Я использую класс MultipleOutputformat в редукторе для достижения этой цели.

Класс My Mapper содержит: 

public class parserMapper extends Mapper<LongWritable, Text, Text, Text> {
   public void map(LongWritable key, Text value, Context context)
   throws IOException, InterruptedException {

   String IPFLAG = "";
   String[] element_data= value.toString.split(","); 

    if(element_data[0].toString().trim().equalsIgnoreCase("005010X222A1")){
        IPFLAG = "P"; 
     }

    else {
       IPFLAG = "I";
     }

   if (IPFLAG == "P") {
     context.write(new Text(IPFLAG), new Text(theData));
     } 

   else if (IPFLAG == "I") {
   context.write(new Text(IPFLAG), new Text(theData));
     }

   else{
   System.out.println("No category found");
     } 


  }

  public void run(Context context) throws IOException, InterruptedException {
        setup(context);
        while (context.nextKeyValue()) {
            map(context.getCurrentKey(), context.getCurrentValue(), context);
        }
        cleanup(context);
    }


}

Класс редуктора включает в себя: 

public class parserReducer extends Reducer<Text, Text, Text, Text> {

    private MultipleOutputs multipleOutputs;

    @Override
    protected void setup(Context context) throws IOException, InterruptedException {
        multipleOutputs = new MultipleOutputs(context);
    }

    @Override
    protected void cleanup(Context context) throws IOException, InterruptedException {
        multipleOutputs.close();
    }

    @Override
    public void reduce(Text key, Iterable<Text> values, Context context) throws IOException, InterruptedException {

        Object c = null;

        try{
            if (!(key.toString().isEmpty())) {

                for (Text value : values) {

                    multipleOutputs.write(c, value, key.toString());
                }

            }
        }
        catch(Exception e){ System.out.println("Caught Exception: " + e.getMessage());}
    }


}

и код драйвера включает в себя => 

public class parserDriver {

 public static void main(String[] args) throws Exception {
    Configuration conf = new Configuration();
    conf.set("textinputformat.record.delimiter", "~"+"\n"+"ISA*");
    Job job = new Job(conf);
        job.setJobName("PARSER");
        job.setJarByClass(parserDriver.class);
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(Text.class);
        job.setMapperClass(parserMapper.class);
        job.setReducerClass(parserReducer.class);
//      job.setOutputFormatClass(TextOutputFormat.class);
        LazyOutputFormat.setOutputFormatClass(job, TextOutputFormat.class);
    //  job.setOutputFormatClass(LazyOutputFormat.class);

/*       MultipleOutputs.addNamedOutput(job, "P", TextOutputFormat.class, Text.class, Text.class);
         MultipleOutputs.addNamedOutput(job, "I", TextOutputFormat.class, Text.class, Text.class);
*/
        // Pass as option -D mapred.reduce.tasks=<number>
        job.setNumReduceTasks(3);       

        /* This line is to accept the input recursively */
        //FileInputFormat.setInputDirRecursive(job, true);

        FileInputFormat.addInputPath(job, "/Users/Mohit/input");
        FileOutputFormat.setOutputPath(job, "/Users/Mohit/output");

        /*
         * Delete output file path if already exists
         */
        FileSystem fs = FileSystem.get(conf);

        if (fs.exists(outputFilePath)) {
            fs.delete(outputFilePath, true);
        }

        return job.waitForCompletion(true) ? 0: 1;
    }
}

Через все это я пытаюсь добиться двух разделов для одного файла

file1 -> P-part-r00000, I-part-r00001

file2 -> P-part-r00002, I-part-r00003

. но я получаю два раздела против всех файлов, которые вводятся в качестве входных данных для этой работы.

file1, file2, file3, file4, file5 -> P-part-r00000, I-part-r00001

Не уверен, что мне здесь не хватает, если кто-нибудь может помочь, пожалуйста?

1 Ответ

0 голосов
/ 14 сентября 2018

1) В вашем драйвере добавьте следующие строки в именование файлов: 

   job.setOutputFormatClass(TextOutputFormat.class);
   MultipleOutputs.addNamedOutput(job, "I", TextOutputFormat.class,
          Text.class, Text.class);
   MultipleOutputs.addNamedOutput(job, "P", TextOutputFormat.class,
          Text.class, Text.class);

2) Измените ваш редуктор так, чтобы каждое значение передавалось в файл с определенным именем: 

@Override
public void reduce(Text key, Iterable<Text> values, Context context) throws IOException, InterruptedException {
    try{
        if (!(key.toString().isEmpty())) {

            for (Text value : values) {

                multipleOutputs.write(key.toString(), key, value);
            }

        }
    }
    catch(Exception e){ System.out.println("Caught Exception: " + e.getMessage());}
}

3) измените число редукторов на 2, чтобы получить ровно 2 файла.

...