一個完整的 mapreduce 程式在分散式運行時有三類實例進程：

1. MRAppMaster：負責整個程式的過程調度及狀態協調
2. MapTask：負責 map 階段的整個數據處理流程
3. ReduceTask：負責 reduce 階段的整個數據處理流程

map階段：2步

　　第一步：設置InputFormat（通常使用TextInputFormat）的類型和數據的路徑—獲取數據的過程（可以得到K1,V1）

　　第二步：自定義Mapper–將K1，V1轉為K2，V2

shuffle階段：4步

　　第三步：分區的動作，如果有多個reduce才去考慮分區，默認只有一個reduce，分區可以省略

　　第四步：排序，默認對K2進行排序（字典序）–管好K2就行

　　第五步：規約，combiner是一個局部的reduce，map端的合併，是對mapreduce的優化操作，不會影響任何結果，減少網路傳輸，默認可以省略

　　第六步：分組，相同的K（K2）對應的V會放到同一個集合中—將map傳遞的K2，V2變成新的K2，V2

Reduce階段：2步

　　第七步：自定義Reducer 得到K2，V2,轉為K3，V3

　　第八步：設置OutputFormat和數據的路徑–生成結果文件

序列化（Serialization）是指把結構化對象轉化為位元組流。

反序列化（Deserialization）是序列化的逆過程。把位元組流轉為結構化對象。

//反序列化的方法，反序列化時，
//從流中讀取到的各個欄位的順序應該與序列化時
//寫出去的順序保持一致
 @Override 
 public void readFields(DataInput in) throws IOException { 
  upflow = in.readLong(); 
  dflow = in.readLong(); 
  sumflow = in.readLong(); 
 } 
 //序列化的方法 
 @Override 
 public void write(DataOutput out) throws IOException { 
  out.writeLong(upflow); 
  out.writeLong(dflow); 
  out.writeLong(sumflow); 
 } 
 @Override 
 public int compareTo(FlowBean o) { 
//實現按照 sumflow 的大小倒序排序 
  return sumflow>o.getSumflow()?-1:1; 
 } 

compareTo方法用於將當前對象與方法的參數進行比較。 如果指定的數與參數相等返回 0。 如果指定的數小於參數返回 -1。 如果指定的數大於參數返回 1。

例如：o1.compareTo(o2) 返回正數的話，當前對象（調用 compareTo 方法的對象 o1）要排在比較對象（compareTo 傳參對象 o2）後面，返回負數的話，放在前面。

Mapreduce 中會將 map 輸出的 kv 對，按照相同 key 分組，然後分發給不同的 reducetask。 默認的分發規則為：根據 key 的 hashcode%reducetask 數來分發

 

每一個 map 都可能會產生大量的本地輸出，Combiner 的作用就是對 map 端的輸出先做一次合併，以減少在 map 和 reduce 節點之間的數據傳輸量，以提高網路 IO 性能，是 MapReduce 的一種優化手段之一。

1. combiner 是 MR 程式中 Mapper 和 Reducer 之外的一種組件
2. combiner 組件的父類就是 Reducer
3. combiner 和 reducer 的區別在於運行的位置：
4. combiner 是在每一個 maptask 所在的節點運行
5. reducer 是接收全局所有 Mapper 的輸出結果；
6. combiner 的意義就是對每一個 maptask 的輸出進行局部匯總，以減小網路傳輸量

具體實現步驟：

1. 自定義一個 combiner 繼承 Reducer，重寫 reduce 方法
2. 在 job 中設置： job.setCombinerClass(CustomCombiner.class)

 

 

map 階段處理的數據如何傳遞給 reduce 階段，是 MapReduce 框架中最關鍵的一個流程，這個流程就叫 shuffle。

1. 將map端結果輸出到環形緩衝區, 默認為100M, 保存的是<key, value>和分區資訊多個ReduceTask時才需要分區
2. 當環形緩衝去到達80%時, 寫入磁碟中, 在寫入之前對數據進行快排, 如果配置了combiner, 還會對有相同分區號和key進行排序

　　　　Combiner規約：作用就是對map端的輸出先做一次合併, 以減少在map和reduce節點之間的數據傳輸量, 以提高網路IO性能, 是MR的一種優化手段之一

3. 將所有溢出的臨時文件進行一次合併操作, 確保一個MapTask最終只生成一個文件
4. Reduce複製一份數據, 默認保存在記憶體的緩衝區中, 到達閾值, 將數據寫到磁碟
5. Reduce複製數據同時, 開啟兩個執行緒對記憶體到本地的數據進行合併
6. 進行合併數據同時, 進行排序, 因為Map端已進行局部排序, Reduce只需保證數據整體有效

　　　　壓縮機制map輸出壓縮（減少shuffle過程中網路傳輸）reduce輸出壓縮（減少HDFS存儲）

　　DEFLATE：自帶，不可切分，和文本處理一樣, 不需要修改

　　Gzip：自帶，不可切分，和文本處理一樣, 不需要修改

　　Bzip2：自帶，可以切分，和文本處理一樣, 不需要修改

　　LZO：需要安裝，可以切分，建立索引指定輸入格式

　　snappy：需要安裝，不可切分，和文本處理一樣, 不需要修改

 

1. 單純按照文件內容切分
2. 切片大小等於block塊大小
3. 不考慮數據集整體, 逐個針對文件切片
4. 切片公式: max(0, min(Long_max, blockSize))

 

　　Map 端 Join 解決數據傾斜，我們為每一個 MapTask 準備一個表的全表數據文件。這種機制叫做 Map Side Join。適用於關聯表中有小表的情形。

CombineTextInputFormat

　　小文件處理場景 將 HDFS 上多個小文件合併到一個 InputSplit中，然後會啟用一個 Map 來處理這裡面的文件，以此減少 MR 整體作業的運行時間。

 

NameNode 維護整個文件系統元數據

按照類型劃分：文件資訊、塊資訊、datanode節點資訊

按照形式劃分：記憶體元數據、文件元數據

1. 記憶體元數據：文件資訊、塊資訊、datanode節點資訊
2. 文件元數據：只包含文件資訊，其他資訊是在datanode啟動的時候進行上報

fsimage鏡像文件

　　通常很大，保存的是一個小時前文件資訊，是元數據的一個持久化的檢查點，包含 Hadoop 文件系統中的所有目錄和文件元數據資訊，但不包含文件塊位置的資訊。文件塊位置資訊只存儲在記憶體中，是在 datanode 加入集群的時候，namenode 詢問 datanode 得到的，並且間斷的更新。

edits日誌

　　通常比較小，保存最近更改的文件資訊，存放的是 Hadoop 文件系統的所有更改操作（文件創建，刪除或修改）的日誌，文件系統客戶端執行的更改操作首先會被記錄到 edits 文件中，只有當所有的寫操作都執行完成之後，寫操作才會返回成功。

SecondaryNameNode

　　職責是合併NameNode的edit logs到fsimage文件中, 減小edit logs文件的大小和得到一個最新的fsimage文件

Secondary Namenode將namenode上積累的所有edits和一個最新的fsimage下載到本地, 並載入到記憶體進行merge, 這個過程稱為checkpoint. 一個小時檢查一次fsimage，更新edits的數據到fsimage中。

 

HDFS所處的一種特殊狀態，在這種狀態下，文件系統只接受讀數據請求，而不接受刪除、修改等變更請求，是一種保護機制，用於保證集群中的數據塊的安全性。 如果 HDFS 處於安全模式下，不允許 HDFS 客戶端進行任何修改文件的操作,包括上傳文件，刪除文件，重命名，創建文件夾,修改副本數等操作。

 

是一個資源管理, 任務調度系統

　　ResourceManager：所有資源的監控, 分配和管理, 是一個全局的資源管理系統

　　NodeManager：每一個節點的維護，是每個節點上的資源和任務管理器，它是管理這台機器的代理，負責該節點程式的運行，以及該節點資源的管理和監控

　　ApplicationMaster：具體每一個應用程式的調度和協調, 分配任務, 用戶提交的每個應用程式均包含一個AppMaster

對於所有的applications, RM擁有絕對的控制權和對資源的分配權. 而每個AM則會和RM協商資源, 同時和NodeManager通訊來執行和監控task.

 

1. 客戶端向RM提交程式, 包括AppMaster的必須資訊，例如 ApplicationMaster 程式、啟動 ApplicationMaster 的命令、用戶程式等。
2. RM啟動Container用來運行AppMaster
3. 啟動中的AppMaster向RM註冊, 啟動完成後與RM保持心跳
4. AppMaster向RM申請相應數量的Container
5. RM返回AppMaster申請的Container資訊, 由AppMaster對其進行初始化
6. AppMaster與NM進行通訊, 要求NM啟動Container, 兩者保持心跳, 從而對NM運行的任務進行管理和監控
7. AppMaster對運行中的Container進行監控, Container通過RPC協議對相應的NM彙報資訊
8. 客戶端通過AppMaster獲取運行狀態進度等資訊
9. 結束後, AppMaster向RM進行註銷, 並回收Container

 

　　1）先進先出調度器(企業一定不會)：先進先出, 同一個時間隊列只有一個任務執行

　　

　　2）容量調度器(Hadoop2.7.2默認調度器)：多隊列, 每個隊列內部先進先出, 同一個時間隊列只有一個任務執行, 隊列並行度為隊列個數，

1. 1. 每個隊列可配置資源
   2. 對同一用戶提交的作業所佔資源進行限定
   3. 首先, 計算每個隊列中正在進行的任務與所分得資源的比值, 選出一個值最小的隊列, 也就是最閑的
   4. 其次, 按照作業優先順序和提交時間順序, 同時考慮用戶資源量限制和記憶體限制, 對隊列中的任務排序
   5. 隊列同時按照提交任務的順序並行運行

　　　

 

HA(High Available), 高可用，是保證業務連續性的有效解決方案，一般有兩個或兩個以上的節點，分為活動節點（Active）及備用節點（Standby）。

 

在HA環境中, active的NM出現假死狀態, stanby接收不到active的心跳, 判斷active的NM處於宕機, 但實際上未死亡, stanby轉換為active, 若此時原來active的NM復活, 此時有兩台active的NM, 稱為腦裂

危害

　　1. 造成資源爭奪

　　2. 造成數據不統一

　　聯邦機制就是會有多個NameNode。多個 NameNode 的情況意味著有多個 namespace(命名空間)，區別於 HA 模式下的多 NameNode，它們是擁有著同一個 namespace。

概括：

1. 多個 NN 共用一個集群里的存儲資源，每個 NN 都可以單獨對外提供服務。
2. 每個 NN 都會定義一個存儲池，有單獨的 id，每個 DN 都為所有存儲池提供存儲。
3. DN 會按照存儲池 id 向其對應的 NN 彙報塊資訊，同時，DN 會向所有 NN 彙報本地存儲可用資源情況。

缺點

　　並沒有完全解決單點故障問題。 所以一般集群規模真的很大的時候，會採用 HA+Federation 的部署方案。也就是每個聯合的 namenodes 都是 ha 的。

 

1. 在hdfs-site.xml文件中配置多目錄，最好提前配置好，否則更改目錄需要重新啟動集群
2. NameNode有一個工作執行緒池，用來處理不同DataNode的並發心跳以及客戶端並發的元數據操作。
3. dfs.namenode.handler.count=20 * log2(Cluster Size)，比如集群規模為10台時，此參數設置為60
4. 編輯日誌存儲路徑dfs.namenode.edits.dir設置與鏡像文件存儲路徑dfs.namenode.name.dir盡量分開，達到最低寫入延遲
5. 伺服器節點上YARN可使用的物理記憶體總量，默認是8192（MB），注意，如果你的節點記憶體資源不夠8GB，則需要調減小這個值，而YARN不會智慧的探測節點的物理記憶體總量。　　yarn.nodemanager.resource.memory-mb
6. 單個任務可申請的最多物理記憶體量，默認是8192（MB）。yarn.scheduler.maximum-allocation-mb

 

Hadoop宕機

1. 如果MR造成系統宕機。此時要控制Yarn同時運行的任務數，和每個任務申請的最大記憶體。調整參數：yarn.scheduler.maximum-allocation-mb（單個任務可申請的最多物理記憶體量，默認是8192MB）
2. 如果寫入文件過量造成NameNode宕機。那麼調高Kafka的存儲大小，控制從Kafka到HDFS的寫入速度。高峰期的時候用Kafka進行快取，高峰期過去數據同步會自動跟上。

 

Hadoop基準測試