1. 程式人生 > >Hadoop學習筆記:MapReduce框架詳解

Hadoop學習筆記:MapReduce框架詳解

object 好的 單點故障 提高 apr copy 普通 exce 代表性

開始聊mapreduce,mapreduce是hadoop的計算框架,我學hadoop是從hive開始入手,再到hdfs,當我學習hdfs時候,就感覺到hdfs和mapreduce關系的緊密。這個可能是我做技術研究的思路有關,我開始學習某一套技術總是想著這套技術到底能幹什麽,只有當我真正理解了這套技術解決了什麽問題時候,我後續的學習就能逐步的加快,而學習hdfs時候我就發現,要理解hadoop框架的意義,hdfs和mapreduce是密不可分,所以當我寫分布式文件系統時候,總是感覺自己的理解膚淺,今天我開始寫mapreduce了,今天寫文章時候比上周要進步多,不過到底能不能寫好本文了,只有試試再說了。

Mapreduce初析

Mapreduce是一個計算框架,既然是做計算的框架,那麽表現形式就是有個輸入(input),mapreduce操作這個輸入(input),通過本身定義好的計算模型,得到一個輸出(output),這個輸出就是我們所需要的結果。

我們要學習的就是這個計算模型的運行規則。在運行一個mapreduce計算任務時候,任務過程被分為兩個階段:map階段和reduce階段,每個階段都是用鍵值對(key/value)作為輸入(input)和輸出(output)。而程序員要做的就是定義好這兩個階段的函數:map函數和reduce函數。

Mapreduce的基礎實例

講解mapreduce運行原理前,首先我們看看mapreduce裏的hello world實例WordCount,這個實例在任何一個版本的hadoop安裝程序裏都會有,大家很容易找到,這裏我還是貼出代碼,便於我後面的講解,代碼如下:

 1 /**
 2  * Licensed to the Apache Software Foundation (ASF) under one
 3  * or more contributor license agreements.  See the NOTICE file
 4  * distributed with this work for additional information
 5  * regarding copyright ownership.  The ASF licenses this file
 6  * to you under the Apache License, Version 2.0 (the
7 * "License"); you may not use this file except in compliance 8 * with the License. You may obtain a copy of the License at 9 * 10 * http://www.apache.org/licenses/LICENSE-2.0 11 * 12 * Unless required by applicable law or agreed to in writing, software 13 * distributed under the License is distributed on an "AS IS" BASIS, 14 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. 15 * See the License for the specific language governing permissions and 16 * limitations under the License. 17 */ 18 package org.apache.hadoop.examples; 19 20 import java.io.IOException; 21 import java.util.StringTokenizer; 22 23 import org.apache.hadoop.conf.Configuration; 24 import org.apache.hadoop.fs.Path; 25 import org.apache.hadoop.io.IntWritable; 26 import org.apache.hadoop.io.Text; 27 import org.apache.hadoop.mapreduce.Job; 28 import org.apache.hadoop.mapreduce.Mapper; 29 import org.apache.hadoop.mapreduce.Reducer; 30 import org.apache.hadoop.mapreduce.lib.input.FileInputFormat; 31 import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat; 32 import org.apache.hadoop.util.GenericOptionsParser; 33 34 public class WordCount { 35 36 public static class TokenizerMapper 37 extends Mapper<Object, Text, Text, IntWritable>{ 38 39 private final static IntWritable one = new IntWritable(1); 40 private Text word = new Text(); 41 42 public void map(Object key, Text value, Context context 43 ) throws IOException, InterruptedException { 44 StringTokenizer itr = new StringTokenizer(value.toString()); 45 while (itr.hasMoreTokens()) { 46 word.set(itr.nextToken()); 47 context.write(word, one); 48 } 49 } 50 } 51 52 public static class IntSumReducer 53 extends Reducer<Text,IntWritable,Text,IntWritable> { 54 private IntWritable result = new IntWritable(); 55 56 public void reduce(Text key, Iterable<IntWritable> values, 57 Context context 58 ) throws IOException, InterruptedException { 59 int sum = 0; 60 for (IntWritable val : values) { 61 sum += val.get(); 62 } 63 result.set(sum); 64 context.write(key, result); 65 } 66 } 67 68 public static void main(String[] args) throws Exception { 69 Configuration conf = new Configuration(); 70 String[] otherArgs = new GenericOptionsParser(conf, args).getRemainingArgs(); 71 if (otherArgs.length != 2) { 72 System.err.println("Usage: wordcount <in> <out>"); 73 System.exit(2); 74 } 75 Job job = new Job(conf, "word count"); 76 job.setJarByClass(WordCount.class); 77 job.setMapperClass(TokenizerMapper.class); 78 job.setCombinerClass(IntSumReducer.class); 79 job.setReducerClass(IntSumReducer.class); 80 job.setOutputKeyClass(Text.class); 81 job.setOutputValueClass(IntWritable.class); 82 FileInputFormat.addInputPath(job, new Path(otherArgs[0])); 83 FileOutputFormat.setOutputPath(job, new Path(otherArgs[1])); 84 System.exit(job.waitForCompletion(true) ? 0 : 1); 85 } 86 }

如何運行它,這裏不做累述了,大夥可以百度下,網上這方面的資料很多。這裏的實例代碼是使用新的api,大家可能在很多書籍裏看到講解mapreduce的WordCount實例都是老版本的api,這裏我不給出老版本的api,因為老版本的api不太建議使用了,大家做開發最好使用新版本的api,新版本api和舊版本api有區別在哪裏:

  1. 新的api放在:org.apache.hadoop.mapreduce,舊版api放在:org.apache.hadoop.mapred
  2. 新版api使用虛類,而舊版的使用的是接口,虛類更加利於擴展,這個是一個經驗,大家可以好好學習下hadoop的這個經驗。

其他還有很多區別,都是說明新版本api的優勢,因為我提倡使用新版api,這裏就不講這些,因為沒必要再用舊版本,因此這種比較也沒啥意義了。

下面我對代碼做簡單的講解,大家看到要寫一個mapreduce程序,我們的實現一個map函數和reduce函數。我們看看map的方法:

1 public void map(Object key, Text value, Context context) throws IOException, InterruptedException {…}

這裏有三個參數,前面兩個Object key, Text value就是輸入的key和value,第三個參數Context context這是可以記錄輸入的key和value,例如:context.write(word, one);此外context還會記錄map運算的狀態。

對於reduce函數的方法:

1 public void reduce(Text key, Iterable<IntWritable> values, Context context) throws IOException, InterruptedException {…}

reduce函數的輸入也是一個key/value的形式,不過它的value是一個叠代器的形式Iterable<IntWritable> values,也就是說reduce的輸入是一個key對應一組的值的value,reduce也有context和map的context作用一致。

至於計算的邏輯就是程序員自己去實現了。

下面就是main函數的調用了,這個我要詳細講述下,首先是:

1 Configuration conf = new Configuration();

運行mapreduce程序前都要初始化Configuration,該類主要是讀取mapreduce系統配置信息,這些信息包括hdfs還有mapreduce,也就是安裝hadoop時候的配置文件例如:core-site.xml、hdfs-site.xml和mapred-site.xml等等文件裏的信息,有些童鞋不理解為啥要這麽做,這個是沒有深入思考mapreduce計算框架造成,我們程序員開發mapreduce時候只是在填空,在map函數和reduce函數裏編寫實際進行的業務邏輯,其它的工作都是交給mapreduce框架自己操作的,但是至少我們要告訴它怎麽操作啊,比如hdfs在哪裏啊,mapreduce的jobstracker在哪裏啊,而這些信息就在conf包下的配置文件裏。

接下來的代碼是:

1 String[] otherArgs = new GenericOptionsParser(conf, args).getRemainingArgs();
2     if (otherArgs.length != 2) {
3       System.err.println("Usage: wordcount <in> <out>");
4       System.exit(2);
5     }

If的語句好理解,就是運行WordCount程序時候一定是兩個參數,如果不是就會報錯退出。至於第一句裏的GenericOptionsParser類,它是用來解釋常用hadoop命令,並根據需要為Configuration對象設置相應的值,其實平時開發裏我們不太常用它,而是讓類實現Tool接口,然後再main函數裏使用ToolRunner運行程序,而ToolRunner內部會調用GenericOptionsParser。

接下來的代碼是:

1 Job job = new Job(conf, "word count");
2 job.setJarByClass(WordCount.class);
3 job.setMapperClass(TokenizerMapper.class);
4 job.setCombinerClass(IntSumReducer.class);
5 job.setReducerClass(IntSumReducer.class);

第一行就是在構建一個job,在mapreduce框架裏一個mapreduce任務也叫mapreduce作業也叫做一個mapreduce的job,而具體的map和reduce運算就是task了,這裏我們構建一個job,構建時候有兩個參數,一個是conf這個就不累述了,一個是這個job的名稱。

第二行就是裝載程序員編寫好的計算程序,例如我們的程序類名就是WordCount了。這裏我要做下糾正,雖然我們編寫mapreduce程序只需要實現map函數和reduce函數,但是實際開發我們要實現三個類,第三個類是為了配置mapreduce如何運行map和reduce函數,準確的說就是構建一個mapreduce能執行的job了,例如WordCount類。

第三行和第五行就是裝載map函數和reduce函數實現類了,這裏多了個第四行,這個是裝載Combiner類,這個我後面講mapreduce運行機制時候會講述,其實本例去掉第四行也沒有關系,但是使用了第四行理論上運行效率會更好。

接下來的代碼:

1 job.setOutputKeyClass(Text.class);
2 job.setOutputValueClass(IntWritable.class);

這個是定義輸出的key/value的類型,也就是最終存儲在hdfs上結果文件的key/value的類型。

最後的代碼是:

1     FileInputFormat.addInputPath(job, new Path(otherArgs[0]));
2     FileOutputFormat.setOutputPath(job, new Path(otherArgs[1]));
3     System.exit(job.waitForCompletion(true) ? 0 : 1);

第一行就是構建輸入的數據文件,第二行是構建輸出的數據文件,最後一行如果job運行成功了,我們的程序就會正常退出。FileInputFormat和FileOutputFormat是很有學問的,我會在下面的mapreduce運行機制裏講解到它們。

好了,mapreduce裏的hello word程序講解完畢,我這個講解是從新辦api進行,這套講解在網絡上還是比較少的,應該很具有代表性的。

Mapreduce運行機制

下面我要講講mapreduce的運行機制了,前不久我為公司出了一套hadoop面試題,裏面就問道了mapreduce運行機制,出題時候我發現這個問題我自己似乎也將不太清楚,因此最近幾天惡補了下,希望在本文裏能說清楚這個問題。

下面我貼出幾張圖,這些圖都是我在百度圖片裏找到的比較好的圖片:

圖片一:

技術分享

  圖片二:

技術分享

  圖片三:

技術分享

  圖片四:

技術分享

  圖片五:

技術分享

  圖片六:

技術分享

我現在學習技術很喜歡看圖,每次有了新理解就會去看看圖,每次都會有新的發現。

談mapreduce運行機制,可以從很多不同的角度來描述,比如說從mapreduce運行流程來講解,也可以從計算模型的邏輯流程來進行講解,也許有些深入理解了mapreduce運行機制還會從更好的角度來描述,但是將mapreduce運行機制有些東西是避免不了的,就是一個個參入的實例對象,一個就是計算模型的邏輯定義階段,我這裏講解不從什麽流程出發,就從這些一個個牽涉的對象,不管是物理實體還是邏輯實體。

首先講講物理實體,參入mapreduce作業執行涉及4個獨立的實體:

  1. 客戶端(client):編寫mapreduce程序,配置作業,提交作業,這就是程序員完成的工作;
  2. JobTracker:初始化作業,分配作業,與TaskTracker通信,協調整個作業的執行;
  3. TaskTracker:保持與JobTracker的通信,在分配的數據片段上執行Map或Reduce任務,TaskTracker和JobTracker的不同有個很重要的方面,就是在執行任務時候TaskTracker可以有n多個,JobTracker則只會有一個(JobTracker只能有一個就和hdfs裏namenode一樣存在單點故障,我會在後面的mapreduce的相關問題裏講到這個問題的)
  4. Hdfs:保存作業的數據、配置信息等等,最後的結果也是保存在hdfs上面

那麽mapreduce到底是如何運行的呢?

首先是客戶端要編寫好mapreduce程序,配置好mapreduce的作業也就是job,接下來就是提交job了,提交job是提交到JobTracker上的,這個時候JobTracker就會構建這個job,具體就是分配一個新的job任務的ID值,接下來它會做檢查操作,這個檢查就是確定輸出目錄是否存在,如果存在那麽job就不能正常運行下去,JobTracker會拋出錯誤給客戶端,接下來還要檢查輸入目錄是否存在,如果不存在同樣拋出錯誤,如果存在JobTracker會根據輸入計算輸入分片(Input Split),如果分片計算不出來也會拋出錯誤,至於輸入分片我後面會做講解的,這些都做好了JobTracker就會配置Job需要的資源了。分配好資源後,JobTracker就會初始化作業,初始化主要做的是將Job放入一個內部的隊列,讓配置好的作業調度器能調度到這個作業,作業調度器會初始化這個job,初始化就是創建一個正在運行的job對象(封裝任務和記錄信息),以便JobTracker跟蹤job的狀態和進程。

初始化完畢後,作業調度器會獲取輸入分片信息(input split),每個分片創建一個map任務。接下來就是任務分配了,這個時候tasktracker會運行一個簡單的循環機制定期發送心跳給jobtracker,心跳間隔是5秒,程序員可以配置這個時間,心跳就是jobtracker和tasktracker溝通的橋梁,通過心跳,jobtracker可以監控tasktracker是否存活,也可以獲取tasktracker處理的狀態和問題,同時tasktracker也可以通過心跳裏的返回值獲取jobtracker給它的操作指令。任務分配好後就是執行任務了。在任務執行時候jobtracker可以通過心跳機制監控tasktracker的狀態和進度,同時也能計算出整個job的狀態和進度,而tasktracker也可以本地監控自己的狀態和進度。當jobtracker獲得了最後一個完成指定任務的tasktracker操作成功的通知時候,jobtracker會把整個job狀態置為成功,然後當客戶端查詢job運行狀態時候(註意:這個是異步操作),客戶端會查到job完成的通知的。如果job中途失敗,mapreduce也會有相應機制處理,一般而言如果不是程序員程序本身有bug,mapreduce錯誤處理機制都能保證提交的job能正常完成。

下面我從邏輯實體的角度講解mapreduce運行機制,這些按照時間順序包括:輸入分片(input split)、map階段、combiner階段、shuffle階段和reduce階段

1. 輸入分片(input split):在進行map計算之前,mapreduce會根據輸入文件計算輸入分片(input split),每個輸入分片(input split)針對一個map任務,輸入分片(input split)存儲的並非數據本身,而是一個分片長度和一個記錄數據的位置的數組,輸入分片(input split)往往和hdfs的block(塊)關系很密切,假如我們設定hdfs的塊的大小是64mb,如果我們輸入有三個文件,大小分別是3mb、65mb和127mb,那麽mapreduce會把3mb文件分為一個輸入分片(input split),65mb則是兩個輸入分片(input split)而127mb也是兩個輸入分片(input split),換句話說我們如果在map計算前做輸入分片調整,例如合並小文件,那麽就會有5個map任務將執行,而且每個map執行的數據大小不均,這個也是mapreduce優化計算的一個關鍵點。

2. map階段:就是程序員編寫好的map函數了,因此map函數效率相對好控制,而且一般map操作都是本地化操作也就是在數據存儲節點上進行;

3. combiner階段:combiner階段是程序員可以選擇的,combiner其實也是一種reduce操作,因此我們看見WordCount類裏是用reduce進行加載的。Combiner是一個本地化的reduce操作,它是map運算的後續操作,主要是在map計算出中間文件前做一個簡單的合並重復key值的操作,例如我們對文件裏的單詞頻率做統計,map計算時候如果碰到一個hadoop的單詞就會記錄為1,但是這篇文章裏hadoop可能會出現n多次,那麽map輸出文件冗余就會很多,因此在reduce計算前對相同的key做一個合並操作,那麽文件會變小,這樣就提高了寬帶的傳輸效率,畢竟hadoop計算力寬帶資源往往是計算的瓶頸也是最為寶貴的資源,但是combiner操作是有風險的,使用它的原則是combiner的輸入不會影響到reduce計算的最終輸入,例如:如果計算只是求總數,最大值,最小值可以使用combiner,但是做平均值計算使用combiner的話,最終的reduce計算結果就會出錯。

4. shuffle階段:將map的輸出作為reduce的輸入的過程就是shuffle了,這個是mapreduce優化的重點地方。這裏我不講怎麽優化shuffle階段,講講shuffle階段的原理,因為大部分的書籍裏都沒講清楚shuffle階段。Shuffle一開始就是map階段做輸出操作,一般mapreduce計算的都是海量數據,map輸出時候不可能把所有文件都放到內存操作,因此map寫入磁盤的過程十分的復雜,更何況map輸出時候要對結果進行排序,內存開銷是很大的,map在做輸出時候會在內存裏開啟一個環形內存緩沖區,這個緩沖區專門用來輸出的,默認大小是100mb,並且在配置文件裏為這個緩沖區設定了一個閥值,默認是0.80(這個大小和閥值都是可以在配置文件裏進行配置的),同時map還會為輸出操作啟動一個守護線程,如果緩沖區的內存達到了閥值的80%時候,這個守護線程就會把內容寫到磁盤上,這個過程叫spill,另外的20%內存可以繼續寫入要寫進磁盤的數據,寫入磁盤和寫入內存操作是互不幹擾的,如果緩存區被撐滿了,那麽map就會阻塞寫入內存的操作,讓寫入磁盤操作完成後再繼續執行寫入內存操作,前面我講到寫入磁盤前會有個排序操作,這個是在寫入磁盤操作時候進行,不是在寫入內存時候進行的,如果我們定義了combiner函數,那麽排序前還會執行combiner操作。

每次spill操作也就是寫入磁盤操作時候就會寫一個溢出文件,也就是說在做map輸出有幾次spill就會產生多少個溢出文件,等map輸出全部做完後,map會合並這些輸出文件。這個過程裏還會有一個Partitioner操作,對於這個操作很多人都很迷糊,其實Partitioner操作和map階段的輸入分片(Input split)很像,一個Partitioner對應一個reduce作業,如果我們mapreduce操作只有一個reduce操作,那麽Partitioner就只有一個,如果我們有多個reduce操作,那麽Partitioner對應的就會有多個,Partitioner因此就是reduce的輸入分片,這個程序員可以編程控制,主要是根據實際key和value的值,根據實際業務類型或者為了更好的reduce負載均衡要求進行,這是提高reduce效率的一個關鍵所在。到了reduce階段就是合並map輸出文件了,Partitioner會找到對應的map輸出文件,然後進行復制操作,復制操作時reduce會開啟幾個復制線程,這些線程默認個數是5個,程序員也可以在配置文件更改復制線程的個數,這個復制過程和map寫入磁盤過程類似,也有閥值和內存大小,閥值一樣可以在配置文件裏配置,而內存大小是直接使用reduce的tasktracker的內存大小,復制時候reduce還會進行排序操作和合並文件操作,這些操作完了就會進行reduce計算了。

5. reduce階段:和map函數一樣也是程序員編寫的,最終結果是存儲在hdfs上的。

Mapreduce的相關問題

這裏我要談談我學習mapreduce思考的一些問題,都是我自己想出解釋的問題,但是某些問題到底對不對,就要廣大童鞋幫我確認了。

① jobtracker的單點故障:jobtracker和hdfs的namenode一樣也存在單點故障,單點故障一直是hadoop被人詬病的大問題,為什麽hadoop的做的文件系統和mapreduce計算框架都是高容錯的,但是最重要的管理節點的故障機制卻如此不好,我認為主要是namenode和jobtracker在實際運行中都是在內存操作,而做到內存的容錯就比較復雜了,只有當內存數據被持久化後容錯才好做,namenode和jobtracker都可以備份自己持久化的文件,但是這個持久化都會有延遲,因此真的出故障,任然不能整體恢復,另外hadoop框架裏包含zookeeper框架,zookeeper可以結合jobtracker,用幾臺機器同時部署jobtracker,保證一臺出故障,有一臺馬上能補充上,不過這種方式也沒法恢復正在跑的mapreduce任務。

② 做mapreduce計算時候,輸出一般是一個文件夾,而且該文件夾是不能存在,我在出面試題時候提到了這個問題,而且這個檢查做的很早,當我們提交job時候就會進行,mapreduce之所以這麽設計是保證數據可靠性,如果輸出目錄存在reduce就搞不清楚你到底是要追加還是覆蓋,不管是追加和覆蓋操作都會有可能導致最終結果出問題,mapreduce是做海量數據計算,一個生產計算的成本很高,例如一個job完全執行完可能要幾個小時,因此一切影響錯誤的情況mapreduce是零容忍的。

③ Mapreduce還有一個InputFormat和OutputFormat,我們在編寫map函數時候發現map方法的參數是之間操作行數據,沒有牽涉到InputFormat,這些事情在我們new Path時候mapreduce計算框架幫我們做好了,而OutputFormat也是reduce幫我們做好了,我們使用什麽樣的輸入文件,就要調用什麽樣的InputFormat,InputFormat是和我們輸入的文件類型相關的,mapreduce裏常用的InputFormat有FileInputFormat普通文本文件,SequenceFileInputFormat是指hadoop的序列化文件,另外還有KeyValueTextInputFormat。OutputFormat就是我們想最終存儲到hdfs系統上的文件格式了,這個根據你需要定義了,hadoop有支持很多文件格式,這裏不一一列舉,想知道百度下就看到了。

好了,文章寫完了,呵呵,這篇我自己感覺寫的不錯,是目前hadoop系列文章裏寫的最好的,我後面會再接再厲的。加油!!!

Hadoop學習筆記:MapReduce框架詳解