在羣集上運行Hadoop中的Reducer異常

Question

我有一個在獨立模式下運行時運行完美的map reduce程序，但是當我在學校的Hadoop羣集上運行它時，異常發生在Reducer中。我不知道是什麼例外。我知道這一點，因爲當我在減速器中保持試驗/捕獲時，該作業通過但空輸出。當我不保留try/catch時，作業失敗。由於它是一個學校集羣，我無法訪問任何作業跟蹤器或其他文件。我所能找到的只是通過編程方式。有沒有一種方法可以在運行時找到hadoop上發生的異常？

以下是我的代碼片段

public static class RowMPreMap extends MapReduceBase implements 
      Mapper<LongWritable, Text, Text, Text> { 

    private Text keyText = new Text(); 
    private Text valText = new Text(); 

    public void map(LongWritable key, Text value, 
      OutputCollector<Text, Text> output, Reporter reporter) 
      throws IOException { 

     // Input: (lineNo, lineContent) 

     // Split each line using seperator based on the dataset. 
     String line[] = null; 

     line = value.toString().split(Settings.INPUT_SEPERATOR); 

     keyText.set(line[0]); 
     valText.set(line[1] + "," + line[2]); 

     // Output: (userid, "movieid,rating") 
     output.collect(keyText, valText); 
    } 
} 

public static class RowMPreReduce extends MapReduceBase implements 
     Reducer<Text, Text, Text, Text> { 

    private Text valText = new Text(); 

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

     // Input: (userid, List<movieid, rating>) 

     float sum = 0.0F; 
     int totalRatingCount = 0; 

     ArrayList<String> movieID = new ArrayList<String>(); 
     ArrayList<Float> rating = new ArrayList<Float>(); 

     while (values.hasNext()) { 
      String[] movieRatingPair = values.next().toString().split(","); 
      movieID.add(movieRatingPair[0]); 
      Float parseRating = Float.parseFloat(movieRatingPair[1]); 
      rating.add(parseRating); 

      sum += parseRating; 
      totalRatingCount++; 
     } 

     float average = ((float) sum)/totalRatingCount; 

     for (int i = 0; i < movieID.size(); i++) { 
      valText.set("M " + key.toString() + " " + movieID.get(i) + " " 
        + (rating.get(i) - average)); 
      output.collect(null, valText); 
     } 

     // Output: (null, <M userid, movieid, normalizedrating>) 
    } 
} 

例外發生在上述減速。下面是配置

public void normalizeM() throws IOException, InterruptedException { 
    JobConf conf1 = new JobConf(UVDriver.class); 
    conf1.setMapperClass(RowMPreMap.class); 
    conf1.setReducerClass(RowMPreReduce.class); 
    conf1.setJarByClass(UVDriver.class); 

    conf1.setMapOutputKeyClass(Text.class); 
    conf1.setMapOutputValueClass(Text.class); 

    conf1.setOutputKeyClass(Text.class); 
    conf1.setOutputValueClass(Text.class); 

    conf1.setKeepFailedTaskFiles(true); 

    conf1.setInputFormat(TextInputFormat.class); 
    conf1.setOutputFormat(TextOutputFormat.class); 

    FileInputFormat.addInputPath(conf1, new Path(Settings.INPUT_PATH)); 
    FileOutputFormat.setOutputPath(conf1, new Path(Settings.TEMP_PATH + "/" 
      + Settings.NORMALIZE_DATA_PATH_TEMP)); 

    JobConf conf2 = new JobConf(UVDriver.class); 
    conf2.setMapperClass(ColMPreMap.class); 
    conf2.setReducerClass(ColMPreReduce.class); 
    conf2.setJarByClass(UVDriver.class); 

    conf2.setMapOutputKeyClass(Text.class); 
    conf2.setMapOutputValueClass(Text.class); 

    conf2.setOutputKeyClass(Text.class); 
    conf2.setOutputValueClass(Text.class); 

    FileInputFormat.addInputPath(conf2, new Path(Settings.TEMP_PATH + "/" 
      + Settings.NORMALIZE_DATA_PATH_TEMP)); 
    FileOutputFormat.setOutputPath(conf2, new Path(Settings.TEMP_PATH + "/" 
      + Settings.NORMALIZE_DATA_PATH)); 

    Job job1 = new Job(conf1); 
    Job job2 = new Job(conf2); 

    JobControl jobControl = new JobControl("jobControl"); 
    jobControl.addJob(job1); 
    jobControl.addJob(job2); 
    job2.addDependingJob(job1); 
    handleRun(jobControl); 

} 

Answer 1

即使你沒有訪問到服務器，你可以得到計數器工作：

Counters counters = job.getCounters(); 

和傾倒的計數器集到本地控制檯。除其他外，這些計數器將顯示輸入和寫入映射器和縮減器的記錄數的計數。值爲零的計數器表示工作流程中的問題位置。您可以測量您自己的計數器以幫助調試/監控流量。

Answer 2

我在reducer中發現異常並將堆棧跟蹤寫入文件系統中的文件。我知道這是做這件事最骯髒的方式，但我現在沒有選擇。以下是代碼，如果它將來有助於任何人。把代碼放在catch塊中。

   String valueString = ""; 
       while (values.hasNext()) { 
        valueString += values.next().toString(); 
       } 

       StringWriter sw = new StringWriter(); 
       e.printStackTrace(new PrintWriter(sw)); 
       String exceptionAsString = sw.toString(); 

       Path pt = new Path("errorfile"); 
       FileSystem fs = FileSystem.get(new Configuration()); 
       BufferedWriter br = new BufferedWriter(new OutputStreamWriter(fs.create(pt,true))); 
       br.write(exceptionAsString + "\nkey: " + key.toString() + "\nvalues: " + valueString); 
       br.close(); 

歡迎使用乾淨的方式輸入。

在旁邊的筆記，最終我發現它是一個NumberFormatException。計數器不會幫助我識別這一點。後來我意識到在獨立和羣集中分割輸入的格式是以不同的方式發生的，我還沒有找到原因。