Hadoop編寫調試MapReduce程序詳解

    編程學習，最好的方法仍是本身動手，因此這裏簡單介紹在Hadoop上編寫調試一個MapReduce程序。

    先說一下個人開發環境，個人操做系統是Centos6.0，Hadoop版本是0.20.2，開發環境是eclipse。在Hadoop的0.20.0版本之後，都包含一個新的Java MapReduce API，這個API有時也稱爲上下文對象，新的API在類型上不兼容之前的API。關於新舊API的區別，這裏先不作介紹了，只是在編程的時候必定要注意下，網上好多MapReduce的程序是基於舊的API編寫的，若是本身安裝的是新版的Hadoop，就會在調試時出現錯誤，初學者不明白這一點或許會走彎路。

    1問題描述：查找最高氣溫。就是從氣候日誌數據中查找當年的最高氣溫，假設每一條記錄的格式以下：「China2013inbeijing023isok0"，其中2013是年份，023是溫度記錄，ok表示數據是無缺的（爲了簡單易懂，省略其餘的信息），咱們的任務是從大量的數據記錄中找出北京2013年的最高氣溫。這樣的數據很適合用MapReduce來處理。

    2問題分析，這個問題很簡單，這裏用這麼簡單的數據只是爲了說明在Hadoop上編寫調試MapReduce程序的過程。對於一條數據這須要提取出來年份和溫度，而後找出最大溫度就好了。這了相似於分治法，每個Map過程就是分得過程，Reduce就是合的過程。

    3 編碼：

    3.1Map函數：

//載入一些必要的包
import java.io.IOException;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapred.OutputCollector;
import org.apache.hadoop.mapred.Reporter;
import org.apache.hadoop.mapreduce.Mapper;
//新版APIMap過程須要繼承org.apache.hadoop.mapreduce包Mapper類，並重寫其map方法
public class MaxtemMapper extends 
Mapper<LongWritable,Text,Text,IntWritable>{
	private static final int MISSING=9999;
	public void map(LongWritable key,Text value,Context context)
	throws IOException,InterruptedException{
		String line=value.toString();//把Text類的對象轉化爲String類來處理
		String year=line.substring(5,9);//該String的第5-9就是年份
		int airtemperature;
              //有的數據中溫度前面帶」+「，須要處理下，帶」+「的話第19到21爲溫度，不帶的的話第18到21
              //爲溫度
              if(line.charAt(18)=='+'){
			airtemperature=Integer.parseInt(line.substring(19, 21));
		}
		else {
			airtemperature=Integer.parseInt(line.substring(18,21));
		}
		System.out.println("year:"+year+"tem:"+airtemperature);
              判斷數據是不是正確的
              String quality=line.substring(23, 25);
		if(airtemperature!=MISSING && quality.matches("ok")){
		    context.write(new Text(year), new IntWritable(airtemperature));
		}
	}
}

    3.2Reduce函數：

import java.io.IOException;
import java.util.Iterator;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;

public class MaxtemReducer extends 
Reducer<Text,IntWritable,Text,IntWritable>{
	public void reduce(Text key,Iterator<IntWritable> values,
			Context context)
	throws IOException,InterruptedException{
		int maxValue=Integer.MIN_VALUE;
		while(values.hasNext()){
			maxValue=Math.max(maxValue,values.next().get());
		}
		context.write( key, new IntWritable(maxValue));
	}
	
}

3.3執行MapReduce過程

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;


public class MaxTemperature{
	
	//the main function
	public static void main(String [] args)throws Exception{
		Configuration conf=new Configuration();
		
		if(args.length!=2){
			System.out.println("Usage: Maxtemperature  
                <input path> <output path>");
			System.exit(-1);
		}
		Job job=new Job(conf,"MaxTemperature");
		job.setJarByClass(MaxTemperature.class);
		
		FileInputFormat.addInputPath(job, new
                Path(args[0]));
		FileOutputFormat.setOutputPath(job, new 
                Path(args[1]));
		
		job.setMapperClass(MaxtemMapper.class);
		job.setReducerClass(MaxtemReducer.class);
		job.setOutputKeyClass(Text.class);
		job.setOutputValueClass(IntWritable.class);
		job.waitForCompletion(true);
		//output the details of the work
		System.out.println("name:"+job.getJobName());
		System.out.println("isSuccessful?:"+  
                 (job.isSuccessful()?"yes":"no"));
  		
	}
}

    4進行單元測試：

    編寫完程序後須要進行單元測試，分別對Map函數和Reduce函數進行測試，功能正確後，就能夠在小規模集羣上進行測試，測試成功後就能夠在Hadoop集羣上進行運行。這裏先不介紹單元測試的知識，後面再專門介紹如何在Hadoop中使用MRUnit進行單元測試。MRUnit是由Couldera公司開發的專門針對 Hadoop中 編寫MapReduce單元測試的框架，基本原理是JUnit4和 EasyMock。MR就是Map和Reduce的縮寫。MRUnit框架很是精簡，其核心的單元測試依賴於JUnit。並且MRUnit實現了一套 Mock對象來控制OutputCollector的操做，從而能夠攔截OutputCollector的輸出，和咱們的指望結果進行比較，達到自動斷言 的目的。

    有了MRUnit，對MR程序作重構的時候，只要明確輸入和輸出，就能夠寫出單元測試，而且在放到羣集校驗前進行試驗，從而節省時間和資源，也 能更快的定位到問題。而進行重構的話，只要寫得足夠詳細的單元測試都是綠色的話，那麼基本就能夠保證在羣集運行的結果也是正常的。

    MRUnit不在Apache標準的Hadoop的發行版中，而是在Couldera公司的加強版本中hadoop- 0.20.1+133.tar.gz的contrib\mrunit\hadoop-0.20.1+169.56-mrunit.jar，已經貼在附件 中。只要把它和Junit4的jar添加到Hadoop程序項目的classpath中，就可使用MRUnit了。

4準備測試數據：

4.1創建一個text文檔，test.text內容以下

china2013inbeijing023isok
china2013inbeijing024isok
china2013inbeijing+25isok
china2013inbeijing026isok
china2013inbeijing027isok
china2013inbeijing028isok
china2013inbeijing029isok
china2013inbeijing030isok
china2013inbeijing031isok
china2013inbeijing132isok
china2013inbeijing033isok
china2013inbeijing034isok

4.2在HDFS的工做目錄下創建一個文件夾input，把test.text上傳到input目錄裏：

4.3配置運行參數：

注意在程序運行以前output文件夾是不存在的。hdfs://localhost:9000/user/XXX/input/test.text是輸入文件的絕對目錄。

4.4運行：

    運行結果：

5 在集羣上運行：

5.1打包

Prpject點選右鍵--Export

按照提示將程序打包爲jar文件。在命令行方式下按照以下命令格式調用便可：

Hadoop jar /home/.../xxx/.../test.jar /home/.../xx/.../output

6 MapReduce的Web界面：

用瀏覽器訪問：http://localhost:50030/就能夠找到用戶界面信息，在這裏能夠很方便的進行做業跟蹤，查找做業完成狀況後的統計信息，以及一些相關的日誌信息：

    時間有限，中間的一些具體的細節，好比Hadoop集羣的配置，Hadoop的Eclipse開發環境的配置，Hadoop命令行方式下的操做，後面再作專門講解。