電信數據稽核及處理技術方案

數據稽核及處理技術方案

編寫與審核人

修改歷史

註釋：“更改請求號”為文檔正式發佈後需要變更時的編號。

目錄

1.數據稽核 

1.1數據稽核概念 

1.2數據稽核優點 

1.3稽核策略 

1.3.1數據完整性 

1.3.2數據一致性 

1.3.3數據準確性 

1.4數據稽核流程 

2稽核方案 

2.1稽核數據 

2.2稽核規則 

2.3數據稽核方式 

2.4稽核方式比較和分析 

3數據處理與分析 

3.1技術方案 

3.1.1 Hadoop平臺方案 

3.1.2 MapReduce中的Shuffle和Sort分析 

3.1.3 Hive處理流程 

3.1.4 Hive的服務端組件 

3.1.5 Hive的數據類型 

3.2採集數據稽核 

3.2.1數據完整性稽核 

3.2.2數據一致性稽核 

3.2.3數據準確性稽核 

3.3基站數據分析 

4結語 

1.數據稽核

1.1數據稽核概念

目前通信市場競爭激烈，資費下調，利潤空間減少，應加強業務稽核，保障數據完整，是企業完善治理結構的內在需求。成本高、效率低、稽核過程繁瑣、管理風險高是營收業務管理模式運行中的主要瓶頸。實際工作中，出於降低運營成本和偏重市場營銷的經營思想等因素考慮，業務稽核工作是運營商管理工作中比較薄弱的環節；對電信業務進行稽核，保證受理業務合理、完整、準確也更顯複雜性和必要性為適應內控需要，減少潛在風險，業務稽核將成為電信運營商日常管理工作中重要的一個環節。實行嚴格的業務稽核是電信企業發展的迫切需要，也是建立規範有效的內控制度的必要環節。

電信運營支撐系統中，用戶數據存在以下特點：

第一業務屬性維度多，包含各種訂購關係、營銷活動信息、產品業務屬性、用戶信息、客戶信息、賬戶信息、資費信息等，各項數據分別存放於資料庫中不同的表中。

第二數據量大，每項數據表都是海量的，採用原來的關聯查詢處理方法處理業務，幾乎是不可能的。

第三數據間關係複雜，由於數據屬性維度多，數據存儲比較分散，表間關係已不僅僅是兩兩關係或主從關係，數據間大多呈現星形關係和樹形關係。

第四用戶數據稽核指標要求多，精度要求高，要核對每一項稽核指標，都須從海量的各項數據表中，查找相應的業務屬性數據，數據稽核要求涉及交易性質，有時需要統計分析，而不僅僅是簡單的查詢彙總。

第五稽核指標多且變化頻繁，由於各種交易操作頻繁，發生差異的幾率也比較高，錯誤可能是某數據項的偏差、也可能是數據項間的關係不對。基於以上特點，一方面，用戶數據關係及業務規則複雜，發生錯誤的幾率高；另一方面，由於數據量大且稽核事務複雜，難於及時、準確、全面的發現錯誤數據，以便及時修複，保障計費準確，規避相關投訴。因此，急需一種高效、精準的用戶數據稽核方法或措施，在海量的用戶數據及複雜的關係中，及時發現錯誤數據。

數據稽核是數據質量管控的一個核心內容，重點就是實現數據的完整性和一致性檢查，提升數據質量，數據稽核是一個從數據採集，預處理，比對，分析，預警，通知，問題修複的完整數據質量管控鏈條。我在前面也談到過，在當前的應用和架構下，企業業務系統間的數據集成模式導致了核心的主數據和跨系統共用的動態數據全部落地，由於本身數據集成的問題或者由於數據源頭管理不善等原因導致了大量的數據不一致性。雖然一直在做數據清理工作，但是這種不一致性和問題將持續存在於整個應用架構體系裡面。在這裡簡單談下數據稽核的系統解決方面的事情。

1.2數據稽核優點

提升客戶對服務的感知

1.提升跨域、跨平臺以及跨系統間的數據質量.

2.提高服務的響應效率，進而提升用戶的感知度.

3.有效杜絕數據的差錯，減少用戶投訴.

提供網路運行、生產效率

1.有利於提高網路資源的利用率.

2.減少企業部門間在協同處理業務時因數據質量問題而帶來的溝通和協調時間，同時高效、持續的數據自動質量稽核手段有利於提高企業的生產效率.

3.可靠的數據質量保證企業網路運行的效率.

支撐精確化運營，提升企業生產運營水平

1.為企業的數據共用提供數據質量保證.

2.利於梳理和完善企業的數據質量管控流程.

3.為企業的精確化運營管理提供數據質量保證

1.3稽核策略

如圖所示，DM數據稽核的大致思路是通過數據完整性、數據一致性、數據準確性三方面依次對DM層數據進行稽核，每一步都為下一步做準備，層層遞進，環環相扣，以保證DM獲取層、基礎層、衍生層、複合指標層以及視圖層的數據質量。

1.3.1數據完整性

數據完整性稽核主要包括，實體是否在規定的時間點提供了並加工生成了數據，實體中指標是否完整覆蓋訂閱指標兩個方面，首先考慮實體中各賬期各省份是否有數據（即判斷數據是否缺失），只有在實體有數據的基礎上才能做進一步的數據稽核，其次檢查數據中指標是否滿足需求，是否包含指標訂購的指標。

1.3.2數據一致性

數據在由數據源到數據獲取層，數據獲取層到基礎數據層，再由基礎數據層到衍生數據層的傳遞過程中，數據能否保持一致也成為縱向實體間稽核的內容。在此基礎上，檢查橫向實體間在相同口徑下的相同指標的指標值是否一致。雖然實體間相同口徑下相同的指標是建設集市極力避免出現的，但是一旦出現並使用，就要要對此進行嚴格的稽核管控。這種大量橫縱十字交叉的方式進行一致性的檢查，便形成了一種網狀稽核。數據一致性網狀稽核的目標便是無“漏網之魚”。複合指標層的一致性稽核主要包括複合指標層實體內上期值、累計值等對應一致的稽核，這不僅保證了複合指標層的數據一致，而且便於數據的準確性稽核。

1.3.3數據準確性

數據在時間推移的過程中不可能一成不變，會按著一定規律波動，我們依照以往指標數據，確定不同指標的波動上限，波動下限，形成一個指標的正常波動範圍。在數據保證完整一致的基礎上，對當前更新的月數據作環比來表現月指標的變化狀況，對當前更新的日數據作同比來表現日指標的變化狀況，嚴格控制閥門，一旦超出指標正常波動範圍，準確及時地找到異常數據。另外，我們用排名對比的方法體現複合指標層指標較上月的排名變化，把指標省內排名和全國排名變化較大的標記為異常指標。以上是本月比起上月同期值的變化情況，如果指標為異常，我們並不能確定哪個月的數據異常，因此，引出在時間序列上的指標數據展現，從而確定異常數據來源。

1.4數據稽核流程

首先說下數據稽核的整個流程，首先是數據的採集和適配，這個常見方式是通過ETL工具來完成，ETL工具採集到的數據做初步的數據清理和預處理。在這個步驟完成後根據預定義的數據稽核和校驗規則，對數據進行差異分析和異常分析，對於分析的結果，一方面是實時的預警和通知，一方面是根據預先定義的報表模版生產數據稽核統計報表。以上完全可以配置為一個自動化的流程，當然對於核心的業務對象或實體，我們還可以定義稽核的時間範圍，稽核的業務規則進行實時的數據比對工作。

其次，來看下數據稽核中跨系統數據比對的內容。數據比對本身是一個由粗到細的過程，首先是數據表級別的比較，但是這個往往並不需要；然後是數據表中記錄層級的數據比較，A系統同步了一條數據到B系統，是否正常成功同步到，首先要比對的就是兩個數據表的key值關聯是否存在。行記錄級別比較完成後是欄位級別的數據比對工作，欄位級的比對分為兩個層面，一個是數據表表結構和欄位結構元數據的一致性，如相同的表兩邊欄位數量不一致，相同的欄位的欄位類型或長度不一致等；其次是欄位內數據和內容的一致性比對。還有些數據稽核工具會提供數據參考完整性和通用性業務規則校驗的功能。

2稽核方案

2.1稽核數據

此次稽核需要對採集數據進行稽核，並對3G、4G基站數據分析，由於用戶數據業務屬性多、屬性間關聯規則複雜、數據量龐大、變更頻率高等原因，實現高效、精準的用戶數據稽核一直是較大的難題；大數據最大的優點是針對傳統手段捕捉到的數據之外的非結構化數據。這意味著不能保證輸入的數據是完整的，清洗過的和沒有任何的錯誤。因此用大數據技術稽核是非常有必要的，對於HDFS上的海量數據而言，編寫Mapreduce程式代碼對於類似數據倉庫的需求來說總是顯得相對於難以維護和重用，Hive作為一種基於Hadoop的數據倉庫解決方案應運而生，並得到了廣泛應用。

採集數據的稽核主要是比對採集的數據和上傳到HDFS後的數據，採集的數據是通過程式上傳到HDFS存儲，所以採集數據的稽核要制定數據完整性、數據一致性、數據準確性稽核。基站數據的分析是整合3G、4G基站數據，收集各類商圈的位置信息與基站數據結合，對基站數據打上位置標簽。

2.2稽核規則

根據稽核策略，我們制定了數據完整性、數據一致性、數據準確性稽核的標準，提供數據稽核時參照的依據，也就是稽核規則，此外還對數據處理分析、實時數據分析比對。

2.3數據稽核方式

數據稽核使用實時流處理打破了傳統的數據分析和處理的模式，即數據最終積累和落地後再針對海量數據進行拆分處理，然後進行分析統計，傳統的模式很難真正達到實時性和速度要求。而實時流處理模型的重點正是既有類似Hadoop中MapReduce和PIG一樣的數據處理和調度引擎，由解決了直接通過輸入適配接駁到流的入口，流實時達到實時處理，實時進行分組匯聚等增量操作。在這種模式下一個重點就是前面談到過的對於數據稽核規則需要進行拆分，形成多個可以並行處理的PE任務，對實時達到的數據流進行處理，形成某種結果信息後再向後續節點推送，最終實時的監控和查詢數據比對結果。

在建立數據倉庫時，ETL通常都採用批處理的方式，一般來說是每天的夜間進行跑批。

隨著數據倉庫技術的逐步成熟，企業對數據倉庫的時間延遲有了更高的要求，也就出現了目前常說的實時ETL（Real-Time ETL）。實時ETL是數據倉庫領域里比較新的一部分內容。微批處理的方式和我們通常的ETL處理方式很相似，但是處理的時間間隔要短，例如間隔一個小時處理一次。數據稽核處理方式有以下幾種方式。

1.觸發器方式

是普遍採取的一種增量抽取機制。該方式是根據抽取要求，在要被抽取的源表上建立插入、修改、刪除3個觸發器，每當源表中的數據發生變化，就被相應的觸發器將變化的數據寫入一個增量日誌表，ETL的增量抽取則是從增量日誌表中而不是直接在源表中抽取數據，同時增量日誌表中抽取過的數據要及時被標記或刪除。為了簡單起見，增量日誌表一般不存儲增量數據的所有欄位信息，而只是存儲源表名稱、更新的關鍵字值和更新操作類型(knsen、update或delete)，ETL增量抽取進程首先根據源表名稱和更新的關鍵字值，從源表中提取對應的完整記錄，再根據更新操作類型，對目標表進行相應的處理。

2.時間戳方式

時間戳方式是指增量抽取時，抽取進程通過比較系統時間與抽取源表的時間戳欄位的值來決定抽取哪些數據。這種方式需要在源表上增加一個時間戳欄位，系統中更新修改表數據的時候，同時修改時間戳欄位的值。有的資料庫(例如Sql Server)的時間戳支持自動更新，即表的其它欄位的數據發生改變時，時間戳欄位的值會被自動更新為記錄改變的時刻。在這種情下，進行ETL實施時就只需要在源表加上時間戳欄位就可以了。對於不支持時間戳自動更新的資料庫，這就要求業務系統在更新業務數據時，通過編程的方式手工更新時間戳欄位。使用時間戳方式可以正常捕獲源表的插入和更新操作，但對於刪除操作則無能為力，需要結合其它機制才能完成。

3.全表刪除插入方式

全表刪除插入方式是指每次抽取前先刪除目標表數據，抽取時全新載入數據。該方式實際上將增量抽取等同於全量抽取。對於數據量不大，全量抽取的時間代價小於執行增量抽取的演算法和條件代價時，可以採用該方式。

4.全表比對方式                                                                

全表比對即在增量抽取時，ETL進程逐條比較源表和目標表的記錄，將新增和修改的記錄讀取出來。優化之後的全部比對方式是採用MD5校驗碼，需要事先為要抽取的表建立一個結構類似的MD5臨時表，該臨時表記錄源表的主鍵值以及根據源表所有欄位的數據計算出來的MD5校驗碼，每次進行數據抽取時，對源表和MD5臨時表進行MD5校驗碼的比對，如有不同，進行update操作：如目標表沒有存在該主鍵值，表示該記錄還沒有，則進行insert操作。然後還需要對在源表中已不存在而目標表仍保留的主鍵值，執行delete操作。

5.日誌表方式

對於建立了業務系統的生產資料庫，可以在資料庫中創建業務日誌表，當特定需要監控的業務數據發生變化時，由相應的業務系統程式模塊來更新維護日誌表內容。增量抽取時，

通過讀日誌表數據決定載入哪些數據及如何載入。日誌表的維護需要由業務系統程式用代碼來完成。

6.系統日誌分析方式

該方式通過分析資料庫自身的日誌來判斷變化的數據。關係犁資料庫系統都會將所有的DML操作存儲在日誌文件中，以實現資料庫的備份和還原功能。ETL增暈抽取進程通過對資料庫的日誌進行分析，提取對相關源表在特定時間後發生的DML操作信息，就可以得知自上次抽取時刻以來該表的數據變化情況，從而指導增量抽取動作。有些資料庫系統提供了訪問日誌的專用的程式包(例如Oracle的LogMinder)，使資料庫日誌的分析工作得到大大簡化。

2.4稽核方式比較和分析

ETL在進行增量抽取操作時，有以上各種機制可以選擇。現從相容性、完備性、性能和侵入性3個方面對這些機制的優劣進行比較分析。數據抽取需要面對的源系統，並不一定都是關係型資料庫系統。某個ETL過程需要從若幹年前的遺留系統中抽取Excel或者CSV文本數據的情形是經常發牛的。這時，所有基於關係型資料庫產品的增量機制都無法工作，時間戳方式和全表比對方式可能有一定的利用價值，在最壞的情況下，只有放棄增量抽取的思路，轉而採用全表刪除插入方式。完備性方面，時間戳方式不能捕獲delete操作，需要結合其它方式一起使用。增量抽取的性能因素表現在兩個方面，一是抽取進程本身的性能，二是對源系統性能的負面影響。觸發器方式、日誌表方式以及系統日誌分析方式由於不需要在抽取過程中執行比對步驟，所以增量抽取的性能較佳。全表比對方式需要經過複雜的比對過程才能識別出更改的記錄，抽取性能最差。在對源系統的性能影響方面，觸發器方式由於是直接在源系統業務表上建立觸發器，同時寫臨時表，對於頻繁操作的業務系統可能會有一定的性能損失，尤其是當業務表上執行批量操作時，行級觸發器將會對性能產生嚴重的影響；同步CDC方式內部採用觸發器的方式實現，也同樣存在性能影響的問題；全表比對方式和日誌表方式對數據源系統資料庫的性能沒有任何影響，只是它們需要業務系統進行額外的運算和資料庫操作，會有少許的時間損耗；時間戳方式、系統日誌分析方式以及基於系統日誌分析的方式(非同步CDC和閃回查詢)對資料庫性能的影響也是非常小的。對數據源系統的侵入性是指業務系統是否要為實現增抽取機製做功能修改和額外操作，在這一點上，時間戳方式值得特別關註該方式除了要修改數據源系統表結構外，對於不支持時間戳欄位自動更新的關係型資料庫產品，還必須要修改業務系統的功能，讓它在源表t執行每次操作時都要顯式的更新表的時間戳欄位，這在ETL實施過程中必須得到數據源系統高度的配合才能達到，並且在多數情況下這種要求在數據源系統看來是比較“過分”的，這也是時間戳方式無法得到廣泛運用的主要原因。另外，觸發器方式需要在源表上建立觸發器，這種在某些場合中也遭到拒絕。還有一些需要建立臨時表的方式，例如全表比對和日誌表方式。可能因為開放給ETL進程的資料庫許可權的限制而無法實施。同樣的情況也可能發生在基於系統日誌分析的方式上，因為大多數的資料庫產品只允許特定組的用戶甚至只有DBA才能執行日誌分析。閃回杏詢在侵入性方面的影響是最小的。

3數據處理與分析

針對不同數據類型與應用的大數據處理系統是支持大數據科學研究的基礎平臺，對於規模巨大、結構複雜、價值稀疏的大數據，其處理亦面臨計算複雜度高、任務周期長、實時性要求強等難題，大數據及其處理的這些難點不僅對大數據處理系統的系統架構、計算框架、處理方法提出了新的挑戰，更對大數據處理系統的運行效率及單位能耗提出了苛刻要求，要求大數據處理系統必須具有高效能的特點，對於以高效能為目標的大數據處理系統的系統架構設計、計算框架設計、處理方法設計和測試基準設計研究，其基礎是大數據處理系統的效能評價與優化問題研究，這些問題的解決可奠定大數據處理系統設計、實現、測試與優化的基本準則，是構建能效優化的分散式存儲和處理的硬體及軟體系統架構的重要依據和基礎，因此是大數據分析處理所必須解決的關鍵問題。

3.1技術方案

大數據，首先你要能存的下大數據，HDFS（Hadoop Distributed FileSystem）的設計本質上是為了大量的數據能橫跨成百上千台機器，但是你看到的是一個文件系統而不是很多文件系統。存的下數據之後，我們就得對這些數據進行處理，此次數據稽核中使用了hadoop+hive的大數據處理技術，hive是Hadoop的一個組件，作為數據廠庫，hive的數據是存儲在Hadoop的文件系統中的，hive為Hadoop提供SQL語句，是Hadoop可以通過SQL語句操作文件系統中的數據，hive是依賴Hadoop而存在的。Hive是基於Hadoop的數據倉庫平臺，由Facebook貢獻，其支持類似SQL的結構化查詢功能。Facebook設計開發Hive的初衷就是讓那些熟悉SQL編程方式的人也可以更好的利用hadoop，hive可以讓數據分析人員只關註於具體業務模型，而不需要深入瞭解Map/Reduce的編程細節，但是這並不意味著使用hive不需要瞭解和學習Map/Reduce編程模型和Hadoop，複雜的業務需求和模型總是存在的，對於Hive分析人員來說，深入瞭解Hadoop和Hive的原理和Mapreduce模型，對於優化查詢總有益處。

hive把海量數據存儲於 hadoop 文件系統，而不是資料庫，但提供了一套類資料庫的數據存儲和處理機制，並採用 HQL （類 SQL ）語言對這些數據進行自動化管理和處理。我們可以把 hive 中海量結構化數據看成一個個的表，而實際上這些數據是分散式存儲在 HDFS 中的。 Hive 經過對語句進行解析和轉換，最終生成一系列基於 Hadoop 的 map/reduce 任務，通過執行這些任務完成數據處理。

藉助於Hadoop和HDFS的大數據存儲能力，數據仍然存儲於Hadoop的HDFS中，Hive提供了一種類SQL的查詢語言：HiveQL（HQL），對數據進行管理和分析。Hive 構建在基於靜態批處理的Hadoop 之上，Hadoop 通常都有較高的延遲並且在作業提交和調度的時候需要大量的開銷。因此，Hive 並不能夠在大規模數據集上實現低延遲快速的查詢，例如，Hive 在幾百MB 的數據集上執行查詢一般有分鐘級的時間延遲。

Hive 並不適合那些需要低延遲的應用，例如，聯機事務處理（OLTP）。Hive 查詢操作過程嚴格遵守Hadoop MapReduce 的作業執行模型，Hive 將用戶的HiveQL 語句通過解釋器轉換為MapReduce 作業提交到Hadoop 集群上，Hadoop 監控作業執行過程，然後返回作業執行結果給用戶。

Hive的執行入口是Driver，執行的SQL語句首先提交到Drive驅動，然後調用compiler解釋驅動，最終解釋成MapReduce任務去執行。

3.1.1 Hadoop平臺方案

Hadoop是一個由Apache基金會所開發的分散式系統基礎架構。用戶可以在不瞭解分散式底層細節的情況下，開發分散式程式。充分利用集群的威力進行高速運算和存儲。

Hadoop實現了一個分散式文件系統（Hadoop Distributed File System），簡稱HDFS。HDFS有高容錯性的特點，並且設計用來部署在低廉的（low-cost）硬體上，而且它提供高吞吐量（high throughput）來訪問應用程式的數據，適合那些有著超大數據集（large data set）的應用程式。HDFS放寬了（relax）POSIX的要求，可以以流的形式訪問（streaming access）文件系統中的數據。

Hadoop的框架最核心的設計就是：HDFS和MapReduce。HDFS為海量的數據提供了存儲，則MapReduce為海量的數據提供了計算。

MapReduce是一個軟體框架，基於該框架能夠容易地編寫應用程式，這些應用程式能夠運行在由上千個商用機器組成的大集群上，並以一種可靠的，具有容錯能力的方式並行地處理上TB級別的海量數據集。這個定義裡面有著這些關鍵詞，一是軟體框架，二是並行處理，三是可靠且容錯，四是大規模集群，五是海量數據集。因此，對於MapReduce，可以簡潔地認為，它是一個軟體框架。MapReduce擅長處理大數據，它為什麼具有這種能力呢？這可由MapReduce的設計思想發覺。MapReduce的思想就是“分而治之”。Mapper負責“分”，即把複雜的任務分解為若幹個“簡單的任務”來處理。“簡單的任務”包含三層含義：一是數據或計算的規模相對原任務要大大縮小；二是就近計算原則，即任務會分配到存放著所需數據的節點上進行計算；三是這些小任務可以並行計算，彼此間幾乎沒有依賴關係。Reducer負責對map階段的結果進行彙總。至於需要多少個Reducer，用戶可以根據具體問題，通過在mapred-site.xml配置文件里設置參數mapred.reduce.tasks的值，預設值為1。

3.1.2 MapReduce中的Shuffle和Sort分析

MapReduce是現今一個非常流行的分散式計算框架，它被設計用於並行計算海量數據。第一個提出該技術框架的是Google公司，而Google的靈感則來自於函數式編程語言，如LISP，Scheme，ML等。MapReduce框架的核心步驟主要分兩部分：Map和Reduce。當你向MapReduce框架提交一個計算作業時，它會首先把計算作業拆分成若幹個Map 任務，然後分配到不同的節點上去執行，每一個Map任務處理輸入數據中的一部分，當Map任務完成後，它會生成一些中間文件，這些中間文件將會作為Reduce 任務的輸入數據。Reduce任務的主要目標就是把前面若幹個Map的輸出彙總到一起並輸出。從高層抽象來看，MapReduce的數據流圖如圖 所示：

Shuffle是指從Map產生輸出開始，包括系統執行排序以及傳送Map輸出到Reducer作為輸入的過程。在這裡我們將去探究Shuffle是如何工作的，因為對基礎的理解有助於對MapReduce程式進行調優。

首先從Map端開始分析，當Map開始產生輸出的時候，他並不是簡單的把數據寫到磁碟，因為頻繁的操作會導致性能嚴重下降，他的處理更加複雜，數據首先是寫到記憶體中的一個緩衝區，並作一些預排序，以提升效率，如圖：

每個Map任務都有一個用來寫入輸出數據的迴圈記憶體緩衝區，這個緩衝區預設大小是100M，可以通過io.sort.mb屬性來設置具體的大小，當緩衝區中的數據量達到一個特定的閥值(io.sort.mb * io.sort.spill.percent，其中io.sort.spill.percent 預設是0.80)時，系統將會啟動一個後臺線程把緩衝區中的內容spill 到磁碟。在spill過程中，Map的輸出將會繼續寫入到緩衝區，但如果緩衝區已經滿了，Map就會被阻塞直道spill完成。spill線程在把緩衝區的數據寫到磁碟前，會對他進行一個二次排序，首先根據數據所屬的partition排序，然後每個partition中再按Key排序。輸出包括一個索引文件和數據文件，如果設定了Combiner，將在排序輸出的基礎上進行。Combiner就是一個Mini Reducer，它在執行Map任務的節點本身運行，先對Map的輸出作一次簡單的Reduce，使得Map的輸出更緊湊，更少的數據會被寫入磁碟和傳送到Reducer。Spill文件保存在由mapred.local.dir指定的目錄中，Map任務結束後刪除。

每當記憶體中的數據達到spill閥值的時候，都會產生一個新的spill文件，所以在Map任務寫完他的最後一個輸出記錄的時候，可能會有多個spill文件，在Map任務完成前，所有的spill文件將會被歸併排序為一個索引文件和數據文件。如圖3 所示。這是一個多路歸併過程，最大歸併路數由io.sort.factor 控制(預設是10)。如果設定了Combiner，並且spil文件的數量至少是3（由min.num.spills.for.combine 屬性控制），那麼Combiner將在輸出文件被寫入磁碟前運行以壓縮數據。

對寫入到磁碟的數據進行壓縮（這種壓縮同Combiner 的壓縮不一樣）通常是一個很好的方法，因為這樣做使得數據寫入磁碟的速度更快，節省磁碟空間，並減少需要傳送到Reducer 的數據量。預設輸出是不被壓縮的，但可以很簡單的設置mapred.compress.map.output為true啟用該功能。壓縮所使用的庫由mapred.map.output.compression.codec來設定。當spill文件歸併完畢後，Map將刪除所有的臨時spill 文件，並告知TaskTracker 任務已完成。Reducers 通過HTTP 來獲取對應的數據。用來傳輸partitions數據的工作線程個數由tasktracker.http.threads控制，這個設定是針對每一個TaskTracker的，並不是單個Map，預設值為40，在運行大作業的大集群上可以增大以提升數據傳輸速率。

現在讓我們轉到Shuffle的Reduce部分。Map的輸出文件放置在運行Map任務的TaskTracker的本地磁碟上（註意：Map輸出總是寫到本地磁碟，但是Reduce輸出不是，一般是寫到HDFS），它是運行Reduce任務的TaskTracker所需要的輸入數據。Reduce任務的輸入數據分佈在集群內的多個Map任務的輸出中，Map任務可能會在不同的時間內完成，只要有其中一個Map任務完成，Reduce任務就開始拷貝他的輸出。這個階段稱為拷貝階段，Reduce任務擁有多個拷貝線程，可以並行的獲取Map輸出。可以通過設定mapred.reduce.parallel.copies來改變線程數。

Reduce是怎麼知道從哪些TaskTrackers中獲取Map的輸出呢？當Map任務完成之後，會通知他們的父TaskTracker，告知狀態更新，然後TaskTracker再轉告JobTracker，這些通知信息是通過心跳通信機制傳輸的，因此針對以一個特定的作業，jobtracker知道Map輸出與tasktrackers的映射關係。Reducer中有一個線程會間歇的向JobTracker詢問Map輸出的地址，直到把所有的數據都取到。在Reducer取走了Map輸出之後，TaskTracker不會立即刪除這些數據，因為Reducer可能會失敗，他們會在整個作業完成之後，JobTracker告知他們要刪除的時候才去刪除。

如果Map輸出足夠小，他們會被拷貝到Reduce TaskTracker的記憶體中（緩衝區的大小由mapred.job.shuffle.input.buffer.percnet控制），或者達到了Map輸出的閥值的大小(由mapred.inmem.merge.threshold控制)，緩衝區中的數據將會被歸併然後spill到磁碟。

拷貝來的數據疊加在磁碟上，有一個後臺線程會將它們歸併為更大的排序文件，這樣做節省了後期歸併的時間。對於經過壓縮的Map 輸出，系統會自動把它們解壓到記憶體方便對其執行歸併。

當所有的Map輸出都被拷貝後，Reduce任務進入排序階段（更恰當的說應該是歸併階段，因為排序在Map端就已經完成），這個階段會對所有的Map輸出進行歸併排序，這個工作會重覆多次才能完成。

假設這裡有50個Map輸出（可能有保存在記憶體中的），並且歸併因數是10（由io.sort.factor控制，就像Map端的merge一樣），那最終需要5次歸併。每次歸併會把10個文件歸併為一個，最終生成5個中間文件。在這一步之後，系統不再把5個中間文件歸併成一個，而是排序後直接“喂”給Reduce 函數，省去向磁碟寫數據這一步。最終歸併的數據可以是混合數據，既有記憶體上的也有磁碟上的。由於歸併的目的是歸併最少的文件數目，使得在最後一次歸併時總文件個數達到歸併因數的數目，所以每次操作所涉及的文件個數在實際中會更微妙些。譬如，如果有40個文件，並不是每次都歸併10個最終得到4個文件，相反第一次只歸併4個文件，然後再實現三次歸併，每次10個，最終得到4個歸併好的文件和6 個未歸併的文件。要註意，這種做法並沒有改變歸併的次數，只是最小化寫入磁碟的數據優化措施，因為最後一次歸併的數據總是直接送到Reduce 函數那裡。在Reduce階段，Reduce函數會作用在排序輸出的每一個key上。這個階段的輸出被直接寫到輸出文件系統，一般是HDFS