hadoop梳理

Hadoop簡單介紹

Hadoop是apache基金會組織的一個頂級項目，其核心爲HDFS和MapReduce，HDFS爲海量的數據提供存儲，而MapReduce爲海量的數據提供並行計算，因此簡單說就是大數據的分佈式存儲與計算平臺。
Hadoop 基礎架構負責處理分佈式處理的全部複雜方面：並行化、調度、資源管理、機器間通訊、軟件和硬件故障處理，等等。得益於這種乾淨的抽象，實現處理數百（或者甚至數千）個機器上的數 TB 數據的分佈式應用程序很是容易。
Apache Hadoop :http://archive.apache.org/dis...
Apache Hadoop文檔：http://hadoop.apache.org/docs

經常使用於一下場景：

場景1：數據分析平臺
場景2：推薦系統
場景3：業務系統的底層存儲系統
場景4：業務監控系統

生態系統：
HBase位於結構化存儲層，Hadoop HDFS爲HBase提供了高可靠性的底層存儲支持，Hadoop MapReduce/Hive爲HBase提供了高性能的計算能力，Zookeeper爲HBase提供了穩定服務和failover機制。

hadoop版本

①Apache:官方版本。
②Cloudera：使用下載最多的版本，穩定，有商業支持，在Apache的基礎上打上了一些patch。推薦使用。
③Yahoo：Yahoo內部使用的版本，發不過兩次，已有的版本都放到Apache上，後續再也不繼續發佈，在集中在Apache的版本上

hadoop Cloudera版本

在Apache Hadoop的基礎上，Cloudera公司將其進行商業化改進和更新，它的發行版本就是CDH(Cloudera Distribution Hadoop)。
下載地址：
CDH4.x: http://archive.cloudera.com/c...
CDH5.x: http://archive.cloudera.com/c...
CDH5.3.6文檔：http://archive.cloudera.com/c...

hadoop的4個模塊

Hadoop Common：爲其餘模塊提供基礎設施。
HDFS：高可靠、高吞吐量的分佈式系統。
MapReduce：分佈式離線並行計算框架。
Yarn：任務調度與資源管理的MapReduce框架。

端口號

hadoop2.x經常使用端口
http://www.cnblogs.com/jancco...
8020：接收Client鏈接的RPC端口，用於獲取文件系統metadata信息。core-site.xml。
10020：Hadoop自帶歷史服務器配置。mapred-site.xml
50070：HDFS外部管理界面端口號。經過 jetty 啓動的 Web 服務。

HDFS

Hadoop Distributed File Syetem

主從結構

主節點，只有一個：namenode
從節點，有不少個：datanodes
secondarynamenode

namenode

做用：
①接收用戶請求，返回客戶端數據信息
②維護文件系統的目錄結構
③管理文件元數據，block與datanode之間關係。

文件內容：

1. fsimage：元數據鏡像文件。儲存某一時段NameNode內存元數據信息。
2. edits：操做日誌文件
3. fstime：保存最近一次checkpoint的時間

datanode

做用：
①存儲文件
②文件被分紅block存儲在磁盤上
③爲保證數據安全，文件會有多個副本

存儲的形式：
塊block，大小默認128MB，能夠用戶自定義大小。dfs.blocksize 自定義塊大小屬性，單位字節。
若是文件大小小於數據塊的大小，它是不會佔據整個塊的空間。
多個文件不能放到一個塊中。
一次寫入，屢次讀取。

如：文件大小爲500MB，塊大小是256MB，第一個塊：256，第二個塊：244

SecondaryNameNode

hdfs首次啓動須要格式化文件，生成fsimage元數據鏡像文件，edits文件是用戶行爲記錄文件，secondarynamenode會讀取兩類文件，加載到內存，使得namenode獲取元數據文件。以後會從NameNode上下載元數據信息（fsimage，edits），而後把兩者合併，生成新的fsimage，在本地保存，並將其推送到NameNode，同時重置NameNode的edits。
它的目的是幫助 NameNode 合併編輯日誌，減小 NameNode 啓動時間。

1. HDFS的高可用性（HA）的一個解決方案。但不支持熱備。配置便可。
2. 執行過程：從NameNode上下載元數據信息（fsimage，edits），而後把兩者合併，生成新的fstime， 在本地保存，並將其推送到NameNode，同時重置NameNode的edits。

可靠性

1.block默認有3個副本。
2.3個副本中，第一個副本通常放置在client所在的同一節點上（客戶端無節點則隨機放），第二個副本在與第一個所在節點同一機架的不一樣節點，第三個副本放到不一樣機架的節點，取用時遵循就近原則。
3.datanode以block爲單位，每3s報告心跳狀態（節點間經過ssh通訊，須要節點間設置免密登入），10min不報告的block，nn會認做死掉，並備份其數據到其餘一個dn，以保證副本數量。
4.datanode也會默認每小時報告其全部塊狀態信息。
5.safemode模式，nn接受報告會計算block的損壞率，達到0.01%時，進入安全模式，HDFS只讀不寫。
6.HA

MapReduce

對大規模的數據集的操做分配給各子節點進行操做，而後經過整合各子節點的中間結果，獲得最終結果。簡單來講就是「任務的分解和結果的彙總」。Hadoop將這個工做高度抽象成爲兩個函數，分別爲map和reduce。map負責將任務分解爲多個任務，reduce負責將多個map任務的中間結果合併爲最終結果。至於在此過程當中的其餘步驟，均由hadoop的MapReduce框架自行負責處理，包括工做調度、任務分配、各節點通訊等。
Map 任務 在輸入數據的子集上調用 map 函數。在完成這些調用後，reduce 任務 開始在 map 函數所生成的中間數據上調用 reduce 任務，生成最終的輸出。 map 和 reduce 任務彼此單獨運行，這支持並行和容錯的計算。
MapReduce是一個框架，能夠理解是一種思想，能夠選擇使用語言開發。

也是主從架構:JobTracker，接收客戶提交的計算任務，把計算任務分給TaskTrackers執行，監控TaskTracker的執行狀況。
TaskTracker執行一個簡單的循環，按期發送心跳（heartbeat）給JobTracker。

一個mapreduce程序分爲五塊

input(split) -> map -> shuffle -> reduce -> output
1.input輸入來自 HDFS的 block 。Split 與 block 的對應關係多是多對一，默認爲一對一。

2.map任務
讀取split，其中每一個鍵值對都會調用一次 map 函數
mapper數量直接由split數決定，間接由 1）輸入文件數目，2）輸入文件的大小， 3）配置參數決定。
mapreduce.input.fileinputformat.split.minsize //啓動map最小的split size大小，默認0
mapreduce.input.fileinputformat.split.maxsize //啓動map最大的split size大小，默認256M
dfs.block.size//block塊大小，默認64M
計算公式：splitSize =  Math.max(minSize, Math.min(maxSize, blockSize));

例如一個文件800M，Block大小是128M，那麼Mapper數目就是7個。6個Mapper處理的數據是128M，1個Mapper處理的數據是32M；

若是要處理不少小文件，能夠經過mapreduce.input.fileinputformat.split.minsize調整Mapper的個數，由於小於這個值文件會合並。

map任務的結果都會被寫入環形緩衝區（這個數據結構其實就是個字節數組，大小默認是mapreduce.task.io.sort.mb: 100瞭解更多）。
寫入以前，key 與value 值都會被序列化成字節數組。

3.Map端Shuffle
3.1切分Partition
mapreduce 提供的Partitioner接口，對key 進行 hash 後，再以 reducetask 數量取模，而後到指定的 job 上。如一個鍵值對的partition值爲1，意味着這個鍵值對會交給第一個Reducer處理。

總結：Partition主要做用將map階段產生的全部kv對分配給不一樣的reducer task處理，能夠將reduce階段的處理負載進行分攤。

3.2 排序sort
每一個內存塊中，按key排序

3.3溢寫spill
因爲環形緩衝區的大小限制，當map task輸出結果不少時就可能發生內存溢出，因此須要在必定條件下將緩衝區的數據臨時寫入磁盤，而後從新利用這塊緩衝區。這個從內存往磁盤寫數據的過程被稱爲Spill，中文可譯爲溢寫。
這個溢寫是由另外單獨線程來完成，不影響往緩衝區寫map結果的線程。
整個緩衝區有個溢寫的比例spill.percent。這個比例默認是0.8（由於spill的時候還有可能別的線程在往裏寫數據）。能夠經過mapreduce.map.sort.spill.percent進行調整。

3.4合併merge
產生不少spill文件後，在Map Task結束前會對這些spill文件進行合併，這個過程就是merge的過程。
最多能同時mergemapreduce.task.io.sort.factor（default：10）個spill文件，若是有100個spill個文件，此時就沒法一次完成整個merge的過程，這個時候須要調大factor來減小merge的次數，從而減小磁盤的操做。

3.5combiner 將有相同key的 key/value 對加起來。（可選）
Combiner的適用場景：因爲Combiner的輸出是Reducer的輸入，Combiner毫不能改變最終的計算結果。故大多數狀況下，combiner適用於輸入輸出的key/value類型徹底一致，且不影響最終結果的場景（好比累加、最大值等……須要本身設置）。
min.num.spill.for.combine的參數決定何時進行Combiner操做，默認是3，也就是說spill的文件數在默認狀況下有三個的時候就要進行combine操做，減小溢寫spill到磁盤的數據量，最終減小磁盤數據。

3.6 分組group
默認按key進行分組

3.7壓縮（可選）
爲了減小磁盤IO和網絡IO還能夠進行：壓縮，對spill，merge文件均可以進行壓縮。中間結果很是的大，IO成爲瓶頸的時候壓縮就很是有用，能夠經過mapreduce.map.output.compress（default：false）設置爲true進行壓縮，數據會被壓縮寫入磁盤，讀數據讀的是壓縮數據須要解壓，在實際經驗中Hive在Hadoop的運行的瓶頸通常都是IO而不是CPU，壓縮通常能夠10倍的減小IO操做，壓縮的方式Gzip，Lzo,BZip2,Lzma等，其中Lzo是一種比較平衡選擇，mapreduce.map.output.compress.codec（default：org.apache.hadoop.io.compress.DefaultCodec）參數設置。但這個過程會消耗CPU，適合IO瓶頸比較大。

4.reduce任務前，進行reduce端Shuffle
注意：數據達到reducer以前，mapreduce框架已經對這些數據按鍵排序了。
在 reduce task 以前，不斷拉取當前 job 裏每一個 maptask 的最終結果，而後對從不一樣地方拉取過來的數據不斷地作 merge ，也最終造成一個文件做爲 reduce task 的輸入文件。
4.1複製
一旦一個map任務完成以後，就會經過常規心跳通知應用程序的Application Master。
reduce的一個線程也會週期性地向master詢問，所以reduce知道去哪些機器取數據。
數據被reduce提走以後，map機器不會馬上刪除數據，這是爲了預防reduce任務失敗須要重作。所以map輸出數據是在整個做業完成以後才被刪除掉的。

Reduce進程啓動一些數據copy線程(Fetcher)，經過HTTP方式請求map task所在的TaskTracker獲取map task的輸出文件。由於maptask早已結束，這些文件就歸TaskTracker管理在本地磁盤中。

下載也能夠是並行的從多個map下載，這個並行度是能夠經過mapreduce.reduce.shuffle.parallelcopies(default5）調整。默認狀況下，每一個Reducer只會有5個map端並行的下載線程在從map下數據，這個參數比較適合在map不少而且完成的比較快的job的狀況下調大，有利於reduce更快的獲取屬於本身部分的數據。 在Reducer內存和網絡都比較好的狀況下，也能夠調大該參數。

reduce的每個下載線程在下載某個map數據的時候，有可能由於那個map所在機器發生錯誤，或者中間結果的文件丟失，或者網絡瞬斷等等狀況，這樣reduce的下載就有可能失敗，因此reduce的下載線程並不會無休止的等待下去，當必定時間後下載仍然失敗，那麼下載線程就會放棄此次下載，並在隨後嘗試從另外的地方下載（由於這段時間map可能重跑）。
reduce下載線程的這個最大的下載時間段是能夠經過mapreduce.reduce.shuffle.read.timeout（default180000秒）調整的。若是集羣環境的網絡自己是瓶頸，那麼用戶能夠經過調大這個參數來避免reduce下載線程被誤判爲失敗的狀況。通常狀況下都會調大這個參數。

4.2 排序
默認按key進行排序

4.3合併merge
Copy 過來的數據會先放入內存緩衝區中，這裏的緩衝區大小要比 map 端的更爲靈活，它基於 JVM 的 heap size 設置，由於 Shuffle 階段 Reducer 不運行，因此應該把絕大部分的內存都給 Shuffle 用。

參數 mapreduce.reduce.shuffle.input.buffer.percent（default 0.7f 源碼裏面寫死了） 這個一個百分比，意思是說，shuffile在reduce內存中的數據最多使用內存量爲：0.7 × maxHeap of reduce task，即JVM的heapsize的70%。
好比reducetask的max heapsize爲1G（一般經過mapreduce.admin.reduce.child.java.opts來設置，好比設置爲-Xmx1024m），那麼用來作下載數據緩存的內存就爲大概700MB左右。這700M的內存，跟map端同樣，也不是要等到所有寫滿纔會往磁盤刷的，而是當這700M中被使用到了必定的限度（一般是一個百分比），就會開始往磁盤刷（刷磁盤前會先作sortMerge）。這個限度閾值也是能夠經過參數 mapreduce.reduce.shuffle.merge.percent（default0.66）來設定。

這個過程當中若是你設置有Combiner，也是會啓用的，而後在磁盤中生成了衆多的溢寫文件。這種merge方式一直在運行，直到沒有map端的數據時才結束，而後啓動磁盤到磁盤的merge方式生成最終的那個文件。

補充：merge有三種形式：
1)內存到內存（memToMemMerger）
這種合併將內存中的map輸出合併，而後再寫入內存。這種合併默認關閉，能夠經過mapreduce.reduce.merge.memtomem.enabled(default:false)打開，當map輸出文件達到mapreduce.reduce.merge.memtomem.threshold時，觸發這種合併。

2）內存中Merge（inMemoryMerger）
當緩衝中數據達到配置的閾值時，這些數據在內存中被合併、寫入機器磁盤。
閾值有2種配置方式：        配置內存比例，前面提到reduceJVM堆內存的一部分用於存放來自map任務的輸入，在這基礎之上配置一個開始合併數據的比例。假設用於存放map輸出的內存爲500M，mapreduce.reduce.shuffle.merge.percent配置爲0.66，則當內存中的數據達到330M的時候，會觸發合併寫入。

配置map輸出數量： 經過mapreduce.reduce.merge.inmem.threshold配置。在合併的過程當中，會對被合併的文件作全局的排序。若是做業配置了Combiner，則會運行combine函數，減小寫入磁盤的數據量。
這種merge方式一直在運行，直到沒有 map 端的數據時才結束。

3）磁盤上的Merge（onDiskMerger）具體包括兩個：（一）Copy過程當中磁盤合併（二）磁盤到磁盤。
磁盤到磁盤的 merge 方式生成最終的那個文件（有多少個reduce就生成幾個文件）。當reducer 輸入文件已定，整個 Shuffle 階段纔算結束。而後就是 Reducer 執行，把結果放到 HDFS 上。

5.reduce任務
默認狀況下，reduce是所有從磁盤開始讀處理數據。能夠用mapreduce.reduce.input.buffer.percent（default 0.0）(源代碼MergeManagerImpl.java：674行)來設置reduce的緩存。若是這個參數大於0，那麼就會有必定量的數據被緩存在內存並輸送給reduce，當reduce計算邏輯消耗內存很小時，能夠分一部份內存用來緩存數據，能夠提高計算的速度。因此默認狀況下都是從磁盤讀取數據，若是內存足夠大的話，務必設置該參數讓reduce直接從緩存讀數據

Reduce在這個階段，框架爲已分組的輸入數據中的每一個 <key, (list of values)>對調用一次 reduce(WritableComparable,Iterator, OutputCollector, Reporter)方法。Reduce任務的輸出一般是經過調用 OutputCollector.collect(WritableComparable,Writable)寫入文件系統的。Reducer的輸出是沒有排序的。

Shuffle過程的做用

Hadoop 集羣中，大部分 map task 與 reduce task 的執行是在不一樣的節點上。固然不少狀況下 Reduce 執行時須要跨節點去拉取其它節點上的map task結果。若是集羣正在運行的 job 有不少，那麼 task 的正常執行對集羣內部的網絡資源消耗會很嚴重。而對於必要的網絡資源消耗，最終的目的就是最大化地減小沒必要要的消耗。還有在節點內，相比於內存，磁盤 IO 對 job 完成時間的影響也是可觀的。從最基本的要求來講，對於 MapReduce 的 job 性能調優的 Shuffle 過程，目標指望能夠有：
完整地從map task端拉取數據到reduce 端；
在跨節點拉取數據時，儘量地減小對帶寬的沒必要要消耗；
減小磁盤IO對task執行的影響。

整體來說這段Shuffle過程，能優化的地方主要在於減小拉取數據的量及儘可能使用內存而不是磁盤。

優化

運行map和reduce任務的JVM，內存經過mapred.child.java.opts屬性來設置，儘量設大內存。容器的內存大小經過mapreduce.map.memory.mb和mapreduce.reduce.memory.mb來設置，默認都是1024M。

過估計map的輸出大小，設置合理的mapreduce.task.io.sort.*屬性，使得spill文件數量最小。例如儘量調大mapreduce.task.io.sort.mb。

在reduce端，若是可以讓全部數據都保存在內存中，能夠達到最佳的性能。一般狀況下，內存都保留給reduce函數，可是若是reduce函數對內存需求不是很高，將mapreduce.reduce.merge.inmem.threshold（觸發合併的map輸出文件數）設爲0，mapreduce.reduce.input.buffer.percent（用於保存map輸出文件的堆內存比例）設爲1.0，能夠達到很好的性能提高。在2008年的TB級別數據排序性能測試中，Hadoop就是經過將reduce的中間數據都保存在內存中勝利的。

Hadoop默認使用4KB做爲緩衝，這個算是很小的，能夠經過io.file.buffer.size來調高緩衝池大小。

cpu調整mapreduce.map.cpu.vcores
https://blog.csdn.net/aijiudu...

Yarn

Yet Another Resource Negotiator

MapReduce on Yarn流程

客戶端提交job任務，resourcemanager處理任務請求，首先會選擇一臺nodemanager啓動一個applicationmaster。
applicationmaster啓動後向resourcemanager申請資源，申請到資源後調度nodemanager分配任務，
nodemanager接受任務並運行任務（map/reduce），任務運行結束後會向applicationmaster報告，
applicationmaster最後向resourcemanager報告，並反饋結果給客戶端 。

ApplicationMaster應用管理者：每個應用都會有一個應用管理者，去爲應用程序申請資源並分配、監控任務。
Container：每一個map都是在各自獨立的環境中去運行（資源獨立）。

Hadoop環境搭建

https://blog.csdn.net/gitchat...
不要用root用戶搭建。

啓動命令

啓動當前節點的服務，如namenode、secondarynamenode、datanode、journalnode、dfs、dfsadmin、fsck、balancer、zkfc等：

$HADOOP_HOME/sbin/hadoop-daemon.sh start namenode
$HADOOP_HOME/sbin/hadoop-daemon.sh start datanode

注意：hadoop-daemons.sh其實就是在hadoop-daemon.sh的基礎上還調用了salves.sh，通知其餘機器執行命令，也就是說hadoop-daemon.sh只對一臺機器起做用，可是hadoop-daemons.sh會對多臺機器起做用。
yarn-daemon.sh 同理。

或者分兩個模塊啓動hdfs、yarn。
hdfs裏面按順序分別調用hadoop-daemons.sh腳本啓動namenode，datanode，secondarynamenode，journalnode，zkfc。
stop-dfs.sh和start-dfs.sh同樣，按照啓動的順序調用hadoop-daemons.sh來關閉服務進程；
yarn裏面按順序分別調用yarn-daemons.sh腳本啓動resourcemanager，nodemanager服務。
stop-yarn.sh和start-yarn.sh同樣，按照啓動的順序調用yarn-daemons.sh來關閉服務進程。

$HADOOP_HOME/sbin/start-dfs.sh 
$HADOOP_HOME/sbin/start-yarn.sh

一個命令同時啓動兩個模塊，不推薦：

$HADOOP_HOME/sbin/start-all.sh

sbin/mr-jobhistory-daemon.sh start historyserver 啓動歷史服務

HDFS shell命令

$HADOOP_HOME/bin/hdfs dfs是對hdfs文件的操做命令，和linux中文件操做命令差很少。每一條命令都會讀取配置文件。
$HADOOP_HOME/bin/hdfs dfs -help ls便可查看ls相關幫助信息。
經常使用有：
$HADOOP_HOME/bin/hdfs dfs -get 下載文件，-P遞歸下載全部文件。
$HADOOP_HOME/bin/hdfs dfs -text 讀文件。
$HADOOP_HOME/bin/hdfs dfs -cat 讀文件。
$HADOOP_HOME/bin/hdfs dfs -ls -R遞歸查看全部文件。
$HADOOP_HOME/bin/hdfs dfs -cp -r遞歸複製。
$HADOOP_HOME/bin/hdfs dfs -getmerge 能夠把多個文件（空格隔開）合併下載到本地。
-du -h 文件夾 查看文件夾大小、-put上傳、-mkdir建立目錄（-p遞歸建立）、-rm刪除（-f刪文件）（-R刪目錄）、-Dfs.defaultFS=file:/// -ls /讀本地文件
-chown、 -chmod等。

$HADOOP_HOME/bin/hdfs dfsadmin是對hdfs文件系統的管理命令。
-safemode enter | leave | get | wait即安全模式相關的
-report 查看集羣報告

MR-jar執行

$HADOOP_HOME/bin/yarn jar xxx.jar mainClass /hdfs文件系統上的input文件夾 /hdfs文件系統上的output文件夾

HDFS api

Apache Hadoop Main 2.5.0-cdh5.3.6 API
Hadoop文件系統的接口

MapReducer-Java程序編寫

WIN+eclipse+hadoop2.5.2

二次排序
top N問題

數據類型

全部數據類型都實現Writable接口，以即可被序列化，最終實現網絡傳輸和文件存儲。
write()序列化
readFields()反序列化

由於MR中有個基於鍵的排序過程，因此做爲鍵的類型必須實現Comparable<T>接口。重寫compare()方法。

WritableComparable接口則是結合了以上兩個接口。
還有一個接口RawComparaotor。
WritableComparable是須要把數據流反序列化爲對象後，而後作對象之間的比較，而RawComparator是直接比較數據流的數據，不須要數據流反序列化成對象，省去了新建對象的開銷。

1.基本數據類型
BooleanWritable、ByteWritable、DoubleWritable、FloatWritable
經常使用的來了：
IntWritable、LongWritable
Text文本類型，UTF-8編碼的，
NullWritable，<key,value>中key或value是空時。

2.自定義數據類型
實際開發中經常須要自定義數據類型。
須要實現Writable接口或WritableComparable接口，而後重寫hashCode()、equals()、compareTo()方法便可。

自定義Partitioner

繼承Partitioner<KEY, VALUE>，並實現getPartition(KEY key, VALUE value, int numPartitions)方法

須要自定義分區器的情形1：
每個Reduce的輸出都是有序的，可是將全部Reduce的輸出合併到一塊兒卻並不是是全局有序的，若是要作到全局有序，咱們該怎麼作呢？最簡單的方式，只設置一個Reduce task，可是這樣徹底發揮不出集羣的優點，並且能應對的數據量也很受限。最佳的方式是本身定義一個Partitioner
用輸入數據的最大值除以系統Reduce task數量的商做爲分割邊界，也就是說分割數據的邊界爲此商的1倍、2倍至numPartitions-1倍，這樣就能保證執行partition後的數據是總體有序的。

須要自定義分區器的情形2：
各個Reduce task處理的鍵值對數量極不平衡。對於某些數據集，因爲不少不一樣的key的hash值都同樣，致使這些鍵值對都被分給同一個Reducer處理，而其餘的Reducer處理的鍵值對不多，從而拖延整個任務的進度。（數據傾斜問題）

須要自定義分區器的情形3：
業務須要，如二次排序問題，不是根據key進行分區，而是key的某個屬性進行分區。

數據格式

1.數據輸入格式
抽象類InputFormat用於描述MR做業的數據輸入格式規範。MapReduce框架依賴InputFormat進行輸入數據分片以及提供讀取分片數據的RecordReader實例對象。每個InputFormat類都會有一個對應的RecordReader類，RecordReader類主要做用是將輸入數據轉換爲鍵值對，傳輸給mapper階段的map方法。Map input records記錄了mapper的輸入數據條數。

MapReduce默認的數據輸入格式是:TextInputFormat(LineRecordReader)。除了這個格式器之外，還有KeyValueTextInputFormat, CombineTextInputFormat, SequenceFileInputFormat, DBInputFormat等。

自定義時須要繼承：
抽象類org.apache.hadoop.mapreduce.InputFormat：
getSplits：返回值是分片信息集合；做用：經過分片個數肯定mappre的個數，並根據分片信息中的數據地址信息決定是否採用數據本地化策略。
createRecordReader：建立一個具體input數據並構造key/value鍵值對的RecordReader實例對象。

抽象類org.apache.hadoop.mapreduce.InputSplit：
getLength：獲取分片長度。
getLocations：獲取該分片數據對應的位置信息，肯定數據本地化時候有用。

抽象類org.apache.hadoop.mapreduce.RecordReader：
initialize：根據對應的分片信息進行初始化操做。
nextKeyValue：判斷是否還有下一個key/value鍵值對，若是有返回true；不然返回false。
getCurrentKey/getCurrentValue：獲取當前key/value鍵值對。
getProgress：獲取操做進度信息。
close：關閉資源讀取相關鏈接。

2.數據輸出格式
(OutputFormat)用於描述MR做業的數據輸出格式規範。MapReduce框架依賴OutputFormat進行輸出路徑(輸出空間)檢測、獲取提交job的OutputCommitter實例對象以及提供一個具體定義如何輸出數據的RecordWriter實例對象。每個OutputFormat類都會有一個對應的RecordWriter類，RecordWriter類主要做用是明肯定義如何寫入以及寫入的格式，接收reducer階段輸出的key/value鍵值對。
MapReduce默認的數據輸出格式是:TextOutputFormat(LineRecordWriter)。除了這個格式器之外，還有SequenceFileOutputFormat, DBOutputFormat等。

自定義時須要繼承：
抽象類org.apache.hadoop.mapreduce.OutputFormat：
getRecordWriter：建立一個具體寫數據的RecordWriter實例。
checkOutputSpecs：檢測輸出空間相關信息，若是檢測失敗，直接拋出異常。
getOutputCommitter：獲取一個提交job的committer對象。通常狀況下，直接使用FileOutputCommitter對象便可。若是以爲FileOutputCommitter內容比較多，也能夠本身實現一個徹底爲空的類

抽象類org.apache.hadoop.mapreduce.RecordWriter：write：接收reducer階段產生的輸出key/value鍵值對數據，並將其寫出。close：關閉流，進行一些其餘操做。