一. MapReduce 程式設計模型

還是以一個經典的圖片來說明問題.

1. 首先, 我們能確定我們有一份輸入, 而且他的資料量會很大

2. 通過split之後, 他變成了若干的分片, 每個分片交給一個Map處理

3. map處理完後, tasktracker會把資料進行復制和排序, 然後通過輸出的key 和value進行 partition的劃分, 並把partition相同的map輸出, 合併為相同的reduce的輸入. 4. ruducer通過處理, 把資料輸出, 每個相同的key, 一定在一個reduce中處理完, 每一個reduce至少對應一份輸出(可以通過擴充套件MultipleOutputFormat來得到多分輸出) 5. 來看一個例子, 如下圖:(來自 《hadoop權威指南》 一書)

  說明幾點:   5.1 輸入的資料可能就是一堆文字   5.2 mapper會解析每行資料, 然後提取有效的資料, 作為輸出. 這裡的例子是 從日誌檔案中提取每一年每天的氣溫, 最後會計算每年的最高氣溫   5.3 map的輸出就是一條一條的 key-value   5.4 通過shuffle之後, 變成reduce的輸入, 這是相同的key對應的value被組合成了一個迭代器   5.5 reduce的任務是提取每一年的最高氣溫, 然後輸出

二. Mapper

1. mapper可以選擇性地繼承 MapreduceBase這個基類, 他只是把一些方法實現了而已, 即使方法體是空的. 2. mapper必須實現 Mapper 介面(0.20以前的版本), 這是一個泛型介面, 需要執行輸入和輸出的key-value的型別, 這些型別通常都是Wriable介面的實現類 3. 實現map方法, 方法有四個引數, 前面兩個就是輸入的 Key 和 value, 第三個引數是 OuputCollector, 用於收集輸出的, 第四個是reporter,用來報告一些狀態的,可以用於debug   3.1 input 預設是一行一條記錄, 每天記錄都放在value裡邊   3.2 output  每次蒐集一條 K-V記錄, 一個K可以對應多個value, 在reduce 裡面體現為一個 iterator 4. 覆蓋 configure方法可以得到JobConf的例項, 這個JobConf是在Job執行時傳遞過來的, 可以跟外部資源進行資料互動

三. Reducer

1. reduce也可以選擇繼承 MapreduceBase這個基類, 功能跟mapper一樣. 2. reducer必須實現Reducer介面, 這個介面同樣是泛型介面, 意義跟Mapper的類似 3. 實現reduce方法, 這個方法也有四個引數, 第一個是輸入的key, 第二個是輸入的 value的迭代器, 可以遍歷所有的value,相當於一個列表, outputCollector跟map的一樣, 是輸出的蒐集器, 每次蒐集都是key-value的形式, report的作用跟map的相同. 4. 在新版本中, hadoop已經將後面兩個引數合併到一個context物件裡邊了, 當然還會相容就版本的 介面. >0.19.x 5. 覆蓋configure方法, 作用跟map的相同 6. 覆蓋close 方法,可以做一些reduce結束後的處理工作.(clean up)

四. Combiner

1. combiner的作用是, 將map的輸出,先計算一遍,得到初步的合併結果, 減少reduce的計算壓力. 2. combiner的編寫方法跟reduce是一樣的, 他本來就是一個Reducer的實現類 3. 當reducer符合函式  F(a,b) = F(F(a), F(b)) 時, combinner可以與reduce相同. 比如 sum(a,b,c,d,e,f,g) = sum(sum(a,b) ,sum(c,d,e,f) , sum(g)) 還有max, min等等. 4. 編寫正確的combiner可以優化整個mapreduce程式的效能.(特別是當reduce是效能瓶頸的時候.) 5. combiner可以跟reducer不同.

五. Configuration

1. 後加的屬性的值會覆蓋前面定義的相同名稱的屬性的值. 2. 被定義為 final的屬性(在屬性定義中加上<final>true</final>標籤)不會被後面的同名屬性定義的值給覆蓋. 3. 系統屬性比通過資源定義的屬性優先順序高, 也就是通過System.setProperty()方法會覆蓋在資原始檔中定義的屬性的值. 4. 系統屬性定義必須在資原始檔中有相應的定義才會生效. 5. 通過 -D 選項定義的屬性, 比在資原始檔中定義的屬性優先順序要高.

六. Run Jobs

1. 設定 inputs & output     1.1 先判斷輸入是否存在 (不存在會導致出錯,最好利用程式來判斷.)     1.2 判斷輸出是否已經存在(存在也會導致出錯)     1.3 養成一種好的習慣(先判斷,再執行) 2. 設定 mapper、reducer、combiner. 各個實現類的class物件.  XXXX.class 3. 設定 inputformat & outputformat & types     3.1 input和output format都有兩種, 一種是 textfile, 一種是sequencefile. 簡單理解, textfile是文字組織的形式,sequence file是 二進位制組織的形式.     3.2 Types的設定, 根據輸入和輸出的資料型別, 設定各種Writable介面的實現類的class物件. 4. 設定reduce count     4.1 reduce count可以為0, 當你的資料無需reduce的時候.     4.2 reduce數量最好稍微少於當前可用的slots的數量, 這樣reduce就能在一波計算中算好. (一個slot可以理解為一個計算單元(資源).)

七. 其他的一些細節.

1. ChainMapper可以實現鏈式執行mapper 他本身就是一個Mapper的實現類. 提供了一個addMapper的方法. 2. ChainReducer 跟ChainMapper類似, 可以實現鏈式執行reducer, 他是Reducer的實現類. 3. 多個job先後執行, 可以通過先後執行 JobClient.runJob方法來實現先後順序 4. 擴充套件MultipleOutputFormat介面, 可以實現一個reduce對應多份輸出 (而且可以指定檔名哦) 5. Partitioner 介面用於將 Map的輸出結果進行分割槽, 分割槽相同的key對應的資料會被同一個reducer處理     5.1 提供了一個介面方法: public int getPartition(K2 key, V2 value, int numReduceTasks)     5.2 可以自己定義, 根據key的某些特指來劃分, 也可以根據value的某些特質來劃分.     5.3 numReduceTasks就是設定的reduce的個數.一般返回的partition的值應該都小於這個值.(%) 6. reporter的作用     6.1 reporter.incrCounter(key, amount). 比如對資料計算是, 一些不合規範的髒資料, 我們可以通過counter來記錄有多少     6.2 reporter.setStatus(status); 方法可以設定一條狀態訊息, 當我們發現job執行出現這條訊息是, 說明出現了我們預期的(錯誤或者正確)的情況, 用於debug.     6.3 reporter.progress(), 像mapreduce框架報告當前執行進度. 這個progress可以起到心跳的作用. 一個task要是超過10分鐘沒有想mapreduce框架報告情況, 這個reduce會被kill掉. 當你的任務處理會比較舊是, 最好定時向mapreduce彙報你的狀態. 7. 通過實現Wriable介面, 我們可以自定義key和value的型別, 使用起來就像pojo, 不需要每次都進行parse. 如果你的自定義型別是Key的型別, 則需要同時實現Comparable 介面, 用於排序. 比如MapWritable就是一個例子.

八. 實戰.(簡單篇)

簡單篇:

1. 需求: 統計某個站點每天的PV

2. 資料輸入: 以天為分割槽存放著的日誌資料, 一條日誌代表一個PV

3. 資料輸出: 日期   PV

4. Mapper編寫

主要的工作很簡單, split每一條日誌, 取出日期, 並對該日期的PV蒐集一條記錄, 記錄的value為ONE(1, 一條記錄代表一個PV)

5. Reducer編寫

reduce的任務是將每天(key相同的為同一天) 的日誌進行彙總(sum), 最後以天為key輸出彙總結果.

6. 設定環境, 指定job(Run)

6.1 設定輸入路徑.

6.2 設定輸出路徑

6.3 設定Mapper/Reducer 和 輸入資料的資料格式和資料型別

6.4  執行命令:

hadoop jar site-pv-job.jar org.jiacheo.SitePVSumSampleJob

6.5 檢視hadoop的web 工具, 顯示當前job進度.

可以看出, 此次輸入產生了14292個map,和29個reduce. reduce數這麼少是因為我的reduce的slots數只有30, 所以設定為29, 以防一個掛了, 還能在一波reduce中算好.

6.6 計算結果.

上面部分是hadoop cli客戶端顯示的進度, 中間是web工具顯示的輸入輸出的一些資料的統計.可以看出, 此次輸入資料總共有1.6TB大小, 設計的總記錄數為69.6億. 也就是這份資料記錄了該站點的69.6億的PV. 左下角可以看出, 執行時間比較長, 用了18分鐘+46秒.這裡慢的原因不在於reduce, 而是我的map的slots太少, 只有300個, 總共一萬多個map, 那要分好幾百波才能算完map, 所以瓶頸在map這裡.右下角是統計的結果資料, 可以看出, 該站點的整體的PV是呈現上升趨勢的.

至此, 一個簡單的map/reduce程式就寫好並運行了.