（Hadoop）：reduce方法在執行mapreduce作業時未被執行/調用

我在執行一個我的mapreduce作業時遇到問題。作爲我的map reduce任務的一部分，我使用了包含多個映射方法和單個reducer方法的mapreduce連接。（Hadoop）：reduce方法在執行mapreduce作業時未被執行/調用

我的兩個map方法都得到執行，但我的reducer沒有從我的驅動程序類執行/調用。

因此，最終輸出只包含在我的地圖階段收集的數據。

我在減少階段使用錯誤的輸入和輸出值嗎？地圖和縮小階段之間是否有任何輸入和輸出不匹配？

在這方面幫助我。

這裏是我的代碼..

public class CompareInputTest extends Configured implements Tool { 

public static class FirstFileInputMapperTest extends Mapper<LongWritable,Text,Text,Text>{ 


    private Text word = new Text(); 
    private String keyData,data,sourceTag = "S1$"; 

    public void map(LongWritable key,Text value,Context context) throws IOException, InterruptedException{ 

     String[] values = value.toString().split(";"); 
     keyData = values[1]; 
     data = values[2]; 

     context.write(new Text(keyData), new Text(data+sourceTag)); 


    } 
} 

public static class SecondFileInputMapperTest extends Mapper<LongWritable,Text,Text,Text>{ 
    private Text word = new Text(); 
    private String keyData,data,sourceTag = "S2$"; 
    public void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException{ 

     String[] values = value.toString().split(";"); 
     keyData = values[1]; 
     data = values[2]; 


     context.write(new Text(keyData), new Text(data+sourceTag)); 

    } 

       } 

public static class CounterReducerTest extends Reducer 
{ 
    private String status1, status2; 

    public void reduce(Text key, Iterable<Text> values, Context context) 
     throws IOException, InterruptedException { 
     System.out.println("in reducer"); 

     for(Text value:values) 
      { 
      String splitVals[] = currValue.split("$"); 
     System.out.println("in reducer"); 
     /* 
     * identifying the record source that corresponds to a commonkey and 
     * parses the values accordingly 
     */ 
     if (splitVals[0].equals("S1")) { 
     status1 = splitVals[1] != null ? splitVals[1].trim(): "status1"; 
     } else if (splitVals[0].equals("S2")) { 
      // getting the file2 and using the same to obtain the Message 
      status2 = splitVals[2] != null ? splitVals[2].trim(): "status2"; 
     } 
      } 

     context.write(key, new Text(status1+"$$$")); 
    } 






public static void main(String[] args) throws Exception { 


    int res = ToolRunner.run(new Configuration(), new CompareInputTest(), 
      args); 
System.exit(res); 

    }

}

public int run(String[] args) throws Exception { 
    Configuration conf = new Configuration(); 
    Job job = new Job(conf, "count"); 
    job.setJarByClass(CompareInputTest.class); 
    MultipleInputs.addInputPath(job,new Path(args[0]),TextInputFormat.class,FirstFileInputMapperTest.class); 
    MultipleInputs.addInputPath(job,new Path(args[1]),TextInputFormat.class,SecondFileInputMapperTest.class); 
    job.setReducerClass(CounterReducerTest.class); 
    //job.setNumReduceTasks(1); 
    job.setMapOutputKeyClass(Text.class); 
    job.setMapOutputValueClass(Text.class); 
    job.setOutputKeyClass(Text.class); 
    job.setOutputValueClass(Text.class); 




    FileOutputFormat.setOutputPath(job, new Path(args[2])); 



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

}

}

來源

2014-10-12 user3854185

的Hadoop的哪個版本？ – 2014-10-12 04:16:20

只是檢查減速機類的原型。

extends Reducer<KEY, VALUE, KEY,VALUE>

在你的情況下，由於減速變得作爲輸入，併發出作爲輸出的文本，從

public static class CounterReducerTest extends Reducer

的定義修改爲

public static class CounterReducerTest extends Reducer<Text,Text,Text,Text>

來源

2014-10-12 04:51:36

（Hadoop）：reduce方法在執行mapreduce作業時未被執行/調用

回答

相關問題