分布式計算開源框架Hadoop入門實踐（三）

Hadoop基本流程

一個圖片太大了，只好分割成為兩部分。根據流程圖來說一下具體一個任務執行的情況。

1. 在分布式環境中客戶端創建任務并提交。
2. InputFormat做Map前的預處理，主要負責以下工作：
   
   1. 驗證輸入的格式是否符合JobConfig的輸入定義，這個在實現Map和構建Conf的時候就會知道，不定義可以是Writable的任意子類。
   2. 將input的文件切分為邏輯上的輸入InputSplit，其實這就是在上面提到的在分布式文件系統中blocksize是有大小限制的，因此大文件會被劃分為多個block。
   3. 通過RecordReader來再次處理inputsplit為一組records，輸出給Map。（inputsplit只是邏輯切分的第一步，但是如何根據文件中的信息來切分還需要RecordReader來實現，例如最簡單的默認方式就是回車換行的切分）
3. RecordReader處理后的結果作為Map的輸入，Map執行定義的Map邏輯，輸出處理后的key和value對應到臨時中間文件。
4. Combiner可選擇配置，主要作用是在每一個Map執行完分析以后，在本地優先作Reduce的工作，減少在Reduce過程中的數據傳輸量。
5. Partitioner可選擇配置，主要作用是在多個Reduce的情況下，指定Map的結果由某一個Reduce處理，每一個Reduce都會有單獨的輸出文件。（后面的代碼實例中有介紹使用場景）
6. Reduce執行具體的業務邏輯，并且將處理結果輸出給OutputFormat。
7. OutputFormat的職責是，驗證輸出目錄是否已經存在，同時驗證輸出結果類型是否如Config中配置，最后輸出Reduce匯總后的結果。

業務場景和代碼范例

業務場景描述： 可設定輸入和輸出路徑（操作系統的路徑非HDFS路徑），根據訪問日志分析某一個應用訪問某一個API的總次數和總流量，統計后分別輸出到兩個文件中。這里僅僅為了測試，沒有去細分很多類，將所有的類都歸并于一個類便于說明問題。

測試代碼類圖

LogAnalysiser就是主類，主要負責創建、提交任務，并且輸出部分信息。內部的幾個子類用途可以參看流程中提到的角色職責。具體地看看幾個類和方法的代碼片斷：

LogAnalysiser::MapClass

    
      ??? public static class MapClass extends MapReduceBase
??? ??? implements Mapper<LongWritable, Text, Text, LongWritable> 
??? {
??? ??? public void map(LongWritable key, Text value, OutputCollector<Text, LongWritable> output, Reporter reporter)
??? ??? ??? ??? throws IOException
??? ??? {??? 
??? ??? ??? String line = value.toString();//沒有配置RecordReader，所以默認采用line的實現，key就是行號，value就是行內容
??? ??? ??? if (line == null || line.equals(""))
??? ??? ??? ??? return;
??? ??? ??? String[] words = line.split(",");
??? ??? ??? if (words == null || words.length < 8)
??? ??? ??? ??? return;
??? ??? ??? String appid = words[1];
??? ??? ??? String apiName = words[2];
??? ??? ??? LongWritable recbytes = new LongWritable(Long.parseLong(words[7]));
??? ??? ??? Text record = new Text();
??? ??? ??? record.set(new StringBuffer("flow::").append(appid)
??? ??? ??? ??? ??? ??? ??? .append("::").append(apiName).toString());
??? ??? ??? reporter.progress();
??? ??? ??? output.collect(record, recbytes);//輸出流量的統計結果，通過flow::作為前綴來標示。
??? ??? ??? record.clear();
??? ??? ??? record.set(new StringBuffer("count::").append(appid).append("::").append(apiName).toString());
??? ??? ??? output.collect(record, new LongWritable(1));//輸出次數的統計結果，通過count::作為前綴來標示
??? ??? }??? 
??? }


    
  

LogAnalysiser:: PartitionerClass

    
      ??? public static class PartitionerClass implements Partitioner<Text, LongWritable>
??? {
??? ??? public int getPartition(Text key, LongWritable value, int numPartitions)
??? ??? {
??? ??? ??? if (numPartitions >= 2)//Reduce 個數，判斷流量還是次數的統計分配到不同的Reduce
??? ??? ??? ??? if (key.toString().startsWith("flow::"))
??? ??? ??? ??? ??? return 0;
??? ??? ??? ??? else
??? ??? ??? ??? ??? return 1;
??? ??? ??? else
??? ??? ??? ??? return 0;
??? ??? }
??? ??? public void configure(JobConf job){}??? 
}
    
  

LogAnalysiser:: CombinerClass

參看ReduceClass，通常兩者可以使用一個，不過這里有些不同的處理就分成了兩個。在ReduceClass中藍色的行表示在CombinerClass中不存在。

LogAnalysiser:: ReduceClass

    
      ??? public static class ReduceClass extends MapReduceBase
??? ??? implements Reducer<Text, LongWritable,Text, LongWritable> 
??? {
??? ??? public void reduce(Text key, Iterator<LongWritable> values,
??? ??? ??? ??? OutputCollector<Text, LongWritable> output, Reporter reporter)throws IOException
??? ??? {
??? ??? ??? Text newkey = new Text();
??? ??? ??? newkey.set(key.toString().substring(key.toString().indexOf("::")+2));
??? ??? ??? LongWritable result = new LongWritable();
??? ??? ??? long tmp = 0;
??? ??? ??? int counter = 0;
??? ??? ??? while(values.hasNext())//累加同一個key的統計結果
??? ??? ??? {
??? ??? ??? ??? tmp = tmp + values.next().get();
??? ??? ??? ??? 
??? ??? ??? ??? counter = counter +1;//擔心處理太久，JobTracker長時間沒有收到報告會認為TaskTracker已經失效，因此定時報告一下
??? ??? ??? ??? if (counter == 1000)
??? ??? ??? ??? {
??? ??? ??? ??? ??? counter = 0;
??? ??? ??? ??? ??? reporter.progress();
??? ??? ??? ??? }
??? ??? ??? }
??? ??? ??? result.set(tmp);
??? ??? ??? output.collect(newkey, result);//輸出最后的匯總結果
??? ??? }??? 
??? }
    
  

LogAnalysiser

    
      	public static void main(String[] args)
	{
		try
		{
			run(args);
		} catch (Exception e)
		{
			e.printStackTrace();
		}
	}
	public static void run(String[] args) throws Exception
	{
		if (args == null || args.length <2)
		{
			System.out.println("need inputpath and outputpath");
			return;
		}
		String inputpath = args[0];
		String outputpath = args[1];
		String shortin = args[0];
		String shortout = args[1];
		if (shortin.indexOf(File.separator) >= 0)
			shortin = shortin.substring(shortin.lastIndexOf(File.separator));
		if (shortout.indexOf(File.separator) >= 0)
			shortout = shortout.substring(shortout.lastIndexOf(File.separator));
		SimpleDateFormat formater = new SimpleDateFormat("yyyy.MM.dd");
		shortout = new StringBuffer(shortout).append("-")
			.append(formater.format(new Date())).toString();
		
		
		if (!shortin.startsWith("/"))
			shortin = "/" + shortin;
		if (!shortout.startsWith("/"))
			shortout = "/" + shortout;
		shortin = "/user/root" + shortin;
		shortout = "/user/root" + shortout;			
		File inputdir = new File(inputpath);
		File outputdir = new File(outputpath);
		if (!inputdir.exists() || !inputdir.isDirectory())
		{
			System.out.println("inputpath not exist or isn't dir!");
			return;
		}
		if (!outputdir.exists())
		{
			new File(outputpath).mkdirs();
		}
		
		JobConf conf = new JobConf(new Configuration(),LogAnalysiser.class);//構建Config
		FileSystem fileSys = FileSystem.get(conf);
		fileSys.copyFromLocalFile(new Path(inputpath), new Path(shortin));//將本地文件系統的文件拷貝到HDFS中

		conf.setJobName("analysisjob");
		conf.setOutputKeyClass(Text.class);//輸出的key類型，在OutputFormat會檢查
		conf.setOutputValueClass(LongWritable.class); //輸出的value類型，在OutputFormat會檢查
		conf.setMapperClass(MapClass.class);
		conf.setCombinerClass(CombinerClass.class);
		conf.setReducerClass(ReduceClass.class);
		conf.setPartitionerClass(PartitionerClass.class);
		conf.set("mapred.reduce.tasks", "2");//強制需要有兩個Reduce來分別處理流量和次數的統計
		FileInputFormat.setInputPaths(conf, shortin);//hdfs中的輸入路徑
		FileOutputFormat.setOutputPath(conf, new Path(shortout));//hdfs中輸出路徑
		
		Date startTime = new Date();
	    	System.out.println("Job started: " + startTime);
	    	JobClient.runJob(conf);
	    	Date end_time = new Date();
	    	System.out.println("Job ended: " + end_time);
	    	System.out.println("The job took " + (end_time.getTime() - startTime.getTime()) /1000 + " seconds.");
	    	//刪除輸入和輸出的臨時文件
		fileSys.copyToLocalFile(new Path(shortout),new Path(outputpath));
		fileSys.delete(new Path(shortin),true);
		fileSys.delete(new Path(shortout),true);
	}


    
  

以上的代碼就完成了所有的邏輯性代碼，然后還需要一個注冊驅動類來注冊業務Class為一個可標示的命令，讓hadoop jar可以執行。

    
      public class ExampleDriver {
? public static void main(String argv[]){
??? ProgramDriver pgd = new ProgramDriver();
??? try {
????? pgd.addClass("analysislog", LogAnalysiser.class, "A map/reduce program that analysis log .");
????? pgd.driver(argv);
??? }
??? catch(Throwable e){
????? e.printStackTrace();
??? }
? }
}
    
  

將代碼打成jar，并且設置jar的mainClass為ExampleDriver這個類。在分布式環境啟動以后執行如下語句：

    
      hadoop jar analysiser.jar analysislog /home/wenchu/test-in /home/wenchu/test-out
    
  

在/home/wenchu/test-in中是需要分析的日志文件，執行后就會看見整個執行過程，包括了Map和Reduce的進度。執行完畢會 在/home/wenchu/test-out下看到輸出的內容。有兩個文件：part-00000和part-00001分別記錄了統計后的結果。 如果需要看執行的具體情況，可以看在輸出目錄下的_logs/history/xxxx_analysisjob，里面羅列了所有的Map，Reduce 的創建情況以及執行情況。在運行期也可以通過瀏覽器來查看Map,Reduce的情況：http://MasterIP:50030 /jobtracker.jsp

Hadoop集群測試

首先這里使用上面的范例作為測試，也沒有做太多的優化配置，這個測試結果只是為了看看集群的效果，以及一些參數配置的影響。

文件復制數為1，blocksize 5M

Slave數	處理記錄數(萬條)	執行時間（秒）
2	95	38
2	950	337
4	95	24
4	950	178
6	95	21
6	950	114

Blocksize 5M

Slave數	處理記錄數(萬條)	執行時間（秒）
2（文件復制數為1）	950	337
2（文件復制數為3）	950	339
6（文件復制數為1）	950	114
6（文件復制數為3）	950	117

文件復制數為1

Slave數	處理記錄數(萬條)	執行時間（秒）
6(blocksize 5M)	95	21
6(blocksize 77M)	95	26
4(blocksize 5M)	950	178
4(blocksize 50M)	950	54
6(blocksize 5M)	950	114
6(blocksize 50M)	950	44
6(blocksize 77M)	950	74

測試的數據結果很穩定，基本測幾次同樣條件下都是一樣。通過測試結果可以看出以下幾點：

1. 機器數對于性能還是有幫助的（等于沒說^_^）。
2. 文件復制數的增加只對安全性有幫助，但是對于性能沒有太多幫助。而且現在采取的是將操作系統文件拷貝到HDFS中，所以備份多了，準備的時間很長。
3. blocksize對于性能影響很大，首先如果將block劃分的太小，那么將會增加job的數量，同時也增加了協作的代價，降低了性能，但是配置的太大也會讓job不能最大化并行處理。所以這個值的配置需要根據數據處理的量來考慮。
4. 最后就是除了這個表里面列出來的結果，應該去仔細看輸出目錄中的_logs/history中的xxx_analysisjob這個文件，里面記錄了全部的執行過程以及讀寫情況。這個可以更加清楚地了解哪里可能會更加耗時。

隨想

“云計算”熱的燙手，就和SAAS、Web2及SNS等一樣，往往都是在搞概念，只有真正踏踏實實的大型互聯網公司，才會投入人力物力去研究符合自 己的分布式計算。其實當你的數據量沒有那么大的時候，這種分布式計算也就僅僅只是一個玩具而已，只有在真正解決問題的過程中，它深層次的問題才會被挖掘出 來。

這三篇文章（分布式計算開源框架Hadoop介紹，Hadoop中的集群配置和使用技巧）僅僅是為了給對分布式計算有興趣的朋友拋個磚，要想真的掘到金 子，那么就踏踏實實的去用、去想、去分析。或者自己也會更進一步地去研究框架中的實現機制，在解決自己問題的同時，也能夠貢獻一些什么。

前幾日看到有人跪求成為架構師的方式，看了有些可悲，有些可笑，其實有多少架構師知道什么叫做架構？架構師的職責是什么？與其追求這么一個名號，還不如踏踏實實地做塊石頭沉到水底。要知道，積累和沉淀的過程就是一種成長。

分布式計算開源框架Hadoop入門實踐（三）