云智慧（北京）科技有限公司陳鑫
NullWritable
不想輸出的時候，把它當做key。NullWritable是Writable的一個特殊類，序列化的長度為0，實現(xiàn)方法為空實現(xiàn)，不從數(shù)據(jù)流中讀數(shù)據(jù)，也不寫入數(shù)據(jù)，只充當占位符，如在MapReduce中，如果你不需要使用鍵或值，你就可以將鍵或值聲明為NullWritable,NullWritable是一個不可變的單實例類型。
FileInputFormat繼承于InputFormat
InputFormat的作用：
驗證輸入規(guī)范；
切分輸入文件為InputSpilts;
提供RecordReader來收集InputSplit中的輸入記錄，給Mapper進行執(zhí)行。
RecordReader
將面向字節(jié)的InputSplit轉(zhuǎn)換為面向記錄的視圖，供Mapper或者Reducer使用運行。因此假定處理記錄的責(zé)任界限,為任務(wù)呈現(xiàn)key-value。
SequenceFile:
SequenceFile是包含二進制kv的扁平文件（序列化）。它提供Writer、Reader、Sorter來進行寫、讀、排序功能?；贑ompressionType，SequenceFile有三種對于kv的壓縮方式：
Writer：不壓縮records；
RecordCompressWriter：只壓縮values;
BlockCompressWriter: 壓縮records，keys和values都被分開壓縮在block中，block的大小可以配置；
壓縮方式由合適的CompressionCodec指定。推薦使用此類的靜態(tài)方法createWriter來選擇格式。Reader作為橋接可以讀取以上任何一種壓縮格式。
CompressionCodec：
封裝了關(guān)于流式壓縮/解壓縮的相關(guān)方法。
Mapper
Mapper 將輸入的kv對映射成中間數(shù)據(jù)kv對集合。Maps 將輸入記錄轉(zhuǎn)變?yōu)橹虚g記錄，其中被轉(zhuǎn)化后的記錄不必和輸入記錄類型相同。一個給定的輸入對可以映射為0或者多個輸出對。
在MRJob執(zhí)行過程中，MapReduce框架根據(jù)提前指定的InputFormat(輸入格式對象)產(chǎn)生InputSplit（輸入分片），而每個InputSplit將會由一個map任務(wù)處理。
總起來講，Mapper實現(xiàn)類通過JobConfigurable.configure(JobConf)方法傳入JobConf對象來初始化，然后在每個map任務(wù)中調(diào)用map(WritableComparable,Writable,OutputCollector,Reporter)方法處理InputSplit的每個kv對。MR應(yīng)用可以覆蓋Closeable.close方法去處理一些必須的清理工作。
輸出對不一定和輸入對類型相同。一個給定的輸入對可能映射成0或者很多的輸出對。輸出對是框架通過調(diào)用OutputCollector.colect(WritableComparable,Writable)得到。
MR應(yīng)用可以使用Reporter匯報進度，設(shè)置應(yīng)用層級的狀態(tài)信息，更新計數(shù)器或者只是顯示應(yīng)用處于運行狀態(tài)等。
所有和給定的輸出key關(guān)聯(lián)的中間數(shù)據(jù)都會隨后被框架分組處理，并傳給Reducer處理以產(chǎn)生最終的輸出。用戶可以通過JobConf.setOutputKeyComparatorClass(Class)指定一個Comparator控制分組處理過程。
Mapper輸出都被排序后根據(jù)Reducer數(shù)量進行分區(qū)，分區(qū)數(shù)量等于reduce任務(wù)數(shù)量。用戶可以通過實現(xiàn)自定義的Partitioner來控制哪些keys(記錄)到哪個Reducer中去。
此外，用戶還可以指定一個Combiner，調(diào)用JobConf.setCombinerClass(Class)來實現(xiàn)。這個可以來對map輸出做本地的聚合，有助于減少從mapper到reducer的數(shù)據(jù)量。
經(jīng)過排序的中間輸出數(shù)據(jù)通常以一種簡單的格式(key-len,key,value-len,value)存儲在SequenceFile中。應(yīng)用可以決定是否或者怎樣被壓縮以及壓縮格式，可以通過JobConf來指定CompressionCodec.
如果job沒有reducer，那么mapper的輸出結(jié)果會不經(jīng)過分組排序，直接寫進FileSystem.
Map數(shù)
通常map數(shù)由輸入數(shù)據(jù)總大小決定，也就是所有輸入文件的blocks數(shù)目決定。
每個節(jié)點并行的運行的map數(shù)正常在10到100個。由于Map任務(wù)初始化本身需要一段時間所以map運行時間至少在1分鐘為好。
如此，如果有10T的數(shù)據(jù)文件，每個block大小128M，大使用為82000map數(shù)，除非使用setNumMapTasks(int)（這個方法僅僅對MR框架提供一個建議值）將map數(shù)值設(shè)置到更高。
Reducer
Reducer根據(jù)key將中間數(shù)據(jù)集合處理合并為更小的數(shù)據(jù)結(jié)果集。
用戶可以通過JobConf.setNumReduceTasks(int)設(shè)置作業(yè)的reducer數(shù)目。
整體而言，Reducer實現(xiàn)類通過JobConfigurable.configure(JobConf)方法將JobConf對象傳入,并為Job設(shè)置和初始化Reducer。MR框架調(diào)用 reduce(WritableComparable, Iterator, OutputCollector,Reporter) 來處理以key被分組的輸入數(shù)據(jù)。應(yīng)用可以覆蓋Closeable.close()處理必要的清理操作。
Reducer由三個主要階段組成:shuffle，sort，reduce。
shuffle
輸入到Reducer的輸入數(shù)據(jù)是Mapper已經(jīng)排過序的數(shù)據(jù).在shuffle階段，根據(jù)partition算法獲取相關(guān)的mapper地址，并通過Http協(xié)議將mapper的相應(yīng)輸出數(shù)據(jù)由reducer拉取到reducer機器上處理。
sort
框架在這個階段會根據(jù)key對reducer的輸入進行分組(因為不同的mapper輸出的數(shù)據(jù)中可能含有相同的key)。
shuffle和sort是同時進行的，同時reducer仍然在拉取map的輸出。
Secondary Sort
如果對中間數(shù)據(jù)key進行分組的規(guī)則和在處理化簡階段前對key分組規(guī)則不一致時，可以通過JobConf.setOutputValueGroupingComparator(Class)設(shè)置一個Comparator。因為中間數(shù)據(jù)的分組策略是通過JobConf.setOutputKeyComparatorClass(Class) 設(shè)置的，可以控制中間數(shù)據(jù)根據(jù)哪些key進行分組。而JobConf.setOutputValueGroupingComparator(Class)則可用于在數(shù)據(jù)連接情況下對value進行二次排序。
Reduce(化簡)
這個階段框架循環(huán)調(diào)用 reduce(WritableComparable, Iterator, OutputCollector,Reporter) 方法處理被分組的每個kv對。
reduce 任務(wù)一般通過OutputCollector.collect(WritableComparable, Writable)將輸出數(shù)據(jù)寫入文件系統(tǒng)FileSystem。應(yīng)用可以使用Reporter匯報作業(yè)執(zhí)行進度、設(shè)置應(yīng)用層級的狀態(tài)信息并更新計數(shù)器(Counter),或者只是提示作業(yè)在運行。
注意，Reducer的輸出不會再進行排序。
Reducer數(shù)目
合適的reducer數(shù)目可以這樣估算：(節(jié)點數(shù)目mapred.tasktracker.reduce.tasks.maximum)乘以0.95 或乘以1.75。因子為0.95時，當所有map任務(wù)完成時所有reducer可以立即啟動，并開始從map機器上拉取數(shù)據(jù)。因子為1.75時,最快的一些節(jié)點將完成第一輪reduce處理，此時框架開始啟動第二輪reduce任務(wù)，這樣可以達到比較好的作業(yè)負載均衡。提高reduce數(shù)目會增加框架的運行負擔，但有利于提升作業(yè)的負載均衡并降低失敗的成本。上述的因子使用最好在作業(yè)執(zhí)行時框架仍然有reduce槽為前提，畢竟框架還需要對作業(yè)進行可能的推測執(zhí)行和失敗任務(wù)的處理。
不使用Reducer
如果不需要進行化簡處理，可以將reduce數(shù)目設(shè)為0。這種情況下，map的輸出會直接寫入到文件系統(tǒng)。輸出路徑通過setOutputPath(Path)指定?？蚣茉趯懭霐?shù)據(jù)到文件系統(tǒng)之前不再對map結(jié)果進行排序。
Partitioner
Partitioner對數(shù)據(jù)按照key進行分區(qū)，從而控制map的輸出傳輸?shù)侥膫€reducer上。默認的Partitioner算法是hash（哈希。分區(qū)數(shù)目由作業(yè)的reducer數(shù)目決定。HashPartitioner是默認的Partitioner。
Reporter
Reporter為MR應(yīng)用提供了進度報告、應(yīng)用狀態(tài)信息設(shè)置，和計數(shù)器(Counter)更新等功能.
Mapper和Reducer實現(xiàn)可以使用Reporter匯報進度或者提示作業(yè)在正常運行。在一些場景下，應(yīng)用在處理一些特殊的kv對時耗費了過多時間，這個可能會因為框架假定任務(wù)超時而強制停止了這些作業(yè)。為避免該情況，可以設(shè)置mapred.task.timeout為一個比較高的值或者將其設(shè)置為0以避免超時發(fā)生。
應(yīng)用也可以使用Reporter來更新計數(shù)(Counter)。
OutputCollector
OutputCollector是MR框架提供的通用工具來收集Mapper或者Reducer輸出數(shù)據(jù)(中間數(shù)據(jù)或者最終結(jié)果數(shù)據(jù))。
HadoopMapReduce提供了一些經(jīng)常使用的mapper、reducer和partioner的實現(xiàn)類供我們進行學(xué)習(xí)。
以上有關(guān)configuration和job的部分在新的API中有所改變，簡單說就是在Mapper和Reducer中引入了MapContext和ReduceContext，它們封裝了configuration和outputcollector，以及reporter。
另外有需要云服務(wù)器可以了解下創(chuàng)新互聯(lián)scvps.cn，海內(nèi)外云服務(wù)器15元起步，三天無理由+7*72小時售后在線，公司持有idc許可證，提供“云服務(wù)器、裸金屬服務(wù)器、高防服務(wù)器、香港服務(wù)器、美國服務(wù)器、虛擬主機、免備案服務(wù)器”等云主機租用服務(wù)以及企業(yè)上云的綜合解決方案，具有“安全穩(wěn)定、簡單易用、服務(wù)可用性高、性價比高”等特點與優(yōu)勢，專為企業(yè)上云打造定制，能夠滿足用戶豐富、多元化的應(yīng)用場景需求。
重慶分公司
重慶分公司
