大數據技術之Hadoop(MapReduce)

詳細介紹

目錄

一MapReduce概念..............................................................4

1.1為什么要MapReduce...........................................................................................................4

1.2MapReduce核心思想.....................................................5

1.3MapReduce進程.........................................................6

1.4MapReduce編程規范（八股文）...........................................6

1.5MapReduce程序運行流程分析.............................................7

二MapReduce理論篇............................................................8

2.1Writable序歹ij化..........................................................8

2.1.1常用數據序列化類型................................................8

2.1.2自定義bean對象實現序列化接口.....................................8

2.2InputFormat數據切片機制...............................................10

2.2.1FilelnputFormat切片機制...........................................10

2.2.2CombineTextlnputFormat切片機制...................................13

2.2.3自定義InputFormat.................................................................................................13

2.3MapTask工作機制.......................................................14

2.4Shuffle機制.............................................................15

2.4.1Shuffle機制.......................................................15

2.4.2MapReduce工作流程..............................................15

2.4.3partition分區......................................................17

2.4.4排序..............................................................18

2.4.5Groupingcomparator分組...........................................20

2.4.6Combiner合并.....................................................20

2.4.7數據傾斜&Distributedcache....................................................................................21

2.5ReduceTask工作機制.....................................................21

2.6自定義OutputFormat.........................................................................................................22

2.7MapReduce數據壓縮....................................................23

2.7.1概述..............................................................23

2.7.2MR支持的壓縮編碼...............................................23

2.7.3采用壓縮的位置...................................................24

2.7.4壓縮配置參數.....................................................25

2.8計數器應用.............................................................26

2.9數據清洗...............................................................26

2.10MapReduce與Yarn...........................................................................................................26

2.10.1Yarn概述........................................................26

2.10.2Yarn的重要概念..................................................26

2.10.3Yarn工作機制....................................................27

2.11作業提交全過程........................................................28

2.12MapReduce開發總結...................................................30

2.13MapReduce參數優化...................................................31

2.13.1資源相關參數....................................................31

2.13.2容錯相關參數....................................................32

三MapReduce實戰篇............................................................33

3.1WordCount案例.........................................................33

3.1.1需求1：統計一堆文件中單詞出現的個數（WordCount案例）..........33

3.1.2需求2：把單詞按照ASCII碼奇偶分區（Partitioner）....................................37

3.1.3需求3：對每一個maptask的輸出局部匯總（Combiner）............................38

3.1.4需求4：大量小文件的切片優化（CombineTextlnputFormat）......................40

3.2流量匯總程序案例.......................................................41

3.2.1需求1：統計手機號耗費的總上行流量、下行流量、總流量（序列化）....41

3.2.2需求2：將統計結果按照手機歸屬地不同省份輸出到不同文件中（Partitioner）

46

3.2.3需求3:將統計結果按照總流量倒序排序（排序）....................49

3.3求每個訂單中最貴的商品（Groupingcomparator）....................................................54

3.4MapReduce中多表合并案例..............................................59

3.4.1需求1：reduce端表合并（數據傾斜）...............................60

3.4.2需求2：map端表合并（Distributedcache）.....................................................66

3.5小文件處理（自定義InputFormat）...............................................................................69

3.6過濾日志及自定義日志輸出路徑（自定義OutputFormat）......................................74

3.7日志清洗（數據清洗）...................................................78

3.7.1簡單解析版.......................................................78

3.7.2復雜解析版.......................................................81

3.8倒排索引實戰（多job串聯）.............................................86

3.9找微信共同好友分析實戰................................................91

3.10壓縮/解壓縮...........................................................98

3.10.1對數據流的壓縮和解壓縮..........................................98

3.10.2在M叩輸出端采用壓縮..........................................100

3.10.3在Reduce輸出端采用壓縮........................................102

四、常見錯誤...................................................................104

一MapReduce概念

Mapreduce是一個分布式運算程序的編程框架，是用戶開發“基于hado叩的數據分析

應用”的核心框架；

Mapreduce核心功能是將用戶編寫的業務邏輯代碼和自帶默認組件整合成一個完整的

分布式運算程序，并發運行在一個hadoop集群上。

1.1為什么要M叩Reduce

1）海量數據在單機上處理因為硬件資源限制，無法勝任

2）而一旦將單機版程序擴展到集群來分布式運行，將極大增加程序的復雜度和開發難

度

3）引入mapreduce框架后，開發人員可以將絕大部分工作集中在業務邏輯的開發上，

而將分布式計算中的復雜性交由框架來處理。

4）nuipreduce分布式方案考慮的問題

（1）運算邏輯要不要先分后合？

（2）程序如何分配運算任務（切片）？

（3）兩階段的程序如何啟動？如何協調？

（4）整個程序運行過程中的監控？容錯？重試？

分布式方案需要考慮很多問題，但是我們可以將分布式程序中的公共功能封裝成框架，

讓開發人員將精力集中于業務邏輯上。而m叩reduce就是這樣一個分布式程序的通用框架。

1.2MapReduce核心思想

上圖簡單的闡明了map和reduce的兩個過程或者作用，雖然不夠嚴謹，但是足以提供

一個大概的認知，map過程是?個蔬菜到制成食物前的準備工作，reduce將準備好的材料合

并進而制作出食物的過程

向硅谷

ujuiuj.atguigu.com

MapReduce核心編程思想

需求：統計其中每Map和Reduce階段

一個單詞出現的總1）分布式的運算程序往往需要分成至少2個階段

1）讀數據

次數（查詢結果：a-2）第?個階段的maptask并發實例，完全并行運行，

）按行處理

P一個文件，q-z-2互不相干

個文件）maptask"心

4）Hashmap<單向.value+1）3）第二個階段的reducetask并發實例互不相干，但是

他們的數據依穎于個階段的所仃并發實

Hadoopspark等到分給白己的數甥片全曲:讀完之后I.maptask

格按照首個字母范例的輸出

hive5）hashmap

、用分成2個hashmap

Hbase4）MapReduce編程模型只能包含一個map階段和一個

Hadoopreduce階段，如果用戶的業務邏輯非常復雜,那就只

能多個程序,中行運行

sparkmapreduce

統計a-p開頭的單詞

maptask

輸出數據

輸出結果若干問題細節

Oreducetask;>

Hashmap(a-p)一一一到文件1)maptask如何進行任務分配

Hashmap(q-z)2)Reducetask如何進行任務分配

3)MaptaskfllReducetask之間如何銜接

統計中z開頭的單詞4)如果某maptask運行失敗，如何處理

maptask

5)Maptask如果都要自己負費輸出數

輸出結果據的分區，很麻煩

reducetask

輸入數據到文件

Hashmab(q-z).

1MrAppMaster負責整個程序的過程調度邈燙媽

1）分布式的運算程序往往需要分成至少2個階段

2）第一個階段的maptask并發實例，完全并行運行，互不相干

3）第二個階段的reducetask并發實例互不相干，但是他們的數據依賴于上一個階段的所有

maptask并發實例的輸出

4)MapReduce編程模型只能包含一個map階段和一個reduce階段，如果用戶的業務邏輯非

常復雜，那就只能多個mapreduce程序，串行運行

1.3MapReduce進程

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

1)MrAppMaster：負責整個程序的過程調度及狀態協調

2)MapTask：負責map階段的整個數據處理流程

3)ReduceTask：負責reduce階段的整個數據處理流程

1.4MapReduce編程規范(八股文)

用戶編寫的程序分成三個部分：Nipper,Reducer,Driver(提交運行mr程序的客戶端)

1)Mapper階段

(1)用戶自定義的Mapper要繼承自己的父類

(2)Mapper的輸入數據是KV對的形式(KV的類型可自定義)

(3)Mapper中的業務邏輯寫在map()方法中

(4)Mapper的輸出數據是KV對的形式(KV的類型可自定義)

(5)map()方法(maptask進程)對每一個＜1＜、＞調用一次

2)Reducer階段

(1)用戶自定義的Reducer要繼承自己的父類

(2)Reducer的輸入數據類型對應Mapper的輸出數據類型，也是KV

(3)Reducer的業務邏輯寫在reduce。方法中

(4)Reducetask進程對每一組相同k的＜k,v＞組調用一次reduce。方法

3)Driver階段

整個程序需要一個Drvier來進行提交，提交的是一個描述了各種必要信息的job對象

4)案例實操

詳見3.1.1統ii?堆文件中單詞出現的個數(WordCount案例)。

1.5MapReduce程序運行流程分析

他尚理看

M叩Reduce程序運行流程分析

1待處理文本s讀取數據「Maptaskss.tx,0-128一7收柒kv到媛存

/user/atguigu/inputInputFormat------?K,voutputcollector

8按照k分區排

atguiguss.txt序后寫入磁盤

bigdata200mwordcount

Mapper二理a-g

邏輯其克

Javaphp6reducetask11輸出結果到文件

map(K,v)oteui[u,l><Hbsse,l><$pak.l>

Android<btgdita,l><hivt1>...

Context.write(k,v)wordcountReduce{OutputFormat

Html5

reduce(k,v)

Bigdatabb.txt

Context.wrlte(k,v);

python100mPart-r-00000

Maptaskss.txt128-200}

2客戶斯submit。前，獲取reducetask

行處理數據的信息,然后

OutputFormat、

根據參數配置，形成一個wordcountReducef

他務分配的規劃.reduce(k,v)

Maptaskbb.txt0-100

10reducet>sk^Context.write{k,v);

ss.txt0-128Part-r-00001

取數據,月運算}

ss.txt128-2003提交切片信息

bb.txtJob.split

wc.jar4計算出maptask數星reducetask

Job.xml

9所有maptask任務完wordcountReducefOutputFormat

Mrappmaster成后，啟動相應數量的reduce(k,v)

reducetask,并告知Contextwrite(k,v);

Part-r-00002

NodeManagerreducetask處理數抿范)

圍(數據分區)

1)在MapReduce程序讀取文件的輸入目錄上存放相應的文件。

2)客戶端程序在submit。方法執行前，獲取待處理的數據信息，然后根據集群中參數的配

置形成一個任務分配規劃。

3)客戶端提交job.split、jar包、job.xml等文件給yarn,yarn中的resourcemanager啟動

MRAppMaster?

4)MRAppMaster啟動后根據本次job的描述信息，計算出需要的maplask實例數量，然后

向集群申請機器啟動相應數量的maptask進程。

5)maptask利用客戶指定的inputformat來讀取數據，形成輸入KV對。

6)maptask將輸入KV對傳遞給客戶定義的map()方法，做邏輯運算

7)map()運算完畢后將KV對收集到maptask緩存。

8)maptask緩存中的KV對按照K分區排序后不斷寫到磁盤文件

9)MRAppMaster監控到所有maptask進程任務完成之后，會根據客戶指定的參數啟動相應

數量的reducetask進程，并告知reducetask進程要處理的數據分區。

10)Reducetask進程啟動之后，根據MRAppMaster告知的待處理數據所在位置，從若干臺

maptask運行所在機器上獲取到若干個maptask輸出結果文件，并在本地進行重新歸并排序,

然后按照相同key的KV為一個組，調用客戶定義的reduce。方法進行邏輯運算。

11)Reducetask運算完畢后，調用客戶指定的outputformat將結果數據輸出到外部存儲。

二MapReduce理論篇

2.1Writable序列化

序列化就是把內存中的對象，轉換成字節序列(或其他數據傳輸協議)以便于存儲(持

久化)和網絡傳輸。

反序列化就是將收到字節序列(或其他數據傳輸協議)或者是硬盤的持久化數據，轉換

成內存中的對象。

Java的序列化是一個重量級序列化框架(Serializable),一個對象被序列化后，會附帶

很多額外的信息(各種校驗信息，header,繼承體系等)，不便于在網絡中高效傳輸。所以，

hadoop自己開發了一套序列化機制(Writable),精簡、高效。

2.1.1常用數據序列化類型

常用的數據類型對應的hadoop數據序列化類型

Java類型HadoopWritable類型

booleanBooleanWritable

byteByteWritable

intIntWritable

floatFloatWritable

longLongWritable

doubleDoubleWritable

stringText

mapMapWritable

arrayArrayWritable

2.1.2自定義bean對象實現序列化接口

1)自定義bean對象要想序列化傳輸，必須實現序列化接口，需要注意以下7項。

(1)必須實現Writable接口

(2)反序列化時，需要反射調用空參構造函數，所以必須有空參構造

(3)重寫序列化方法

(4)重寫反序列化方法

(5)注意反序列化的順序和序列化的順序完全一致

(6)要想把結果顯示在文件中，需要重寫toString(),且用“\t“分開，方便后續用

(7)如果需要將自定義的bean放在key中傳輸，則還需要實現comparable接口，因為

mapreduce框中的shuffle過程一定會對key進行排序

//1必須實現Writable接口

publicclassFlowBeanimplementsWritable{

privatelongupFlow;

privatelongdownFlow;

privatelongsumFlow;

111反序列化時，需要反射調用空參構造函數，所以必須有

publicFlowBean(){

super();

*3重寫序列化方法

*

*@paramout

*@throwslOException

*/

@Override

publicvoidwrite(DataOutputout)throwslOException{

out.writeLong(upFlow);

out.writeLong(downFlow);

out.writeLong(sumFlow);

*4重寫反序列化方法

5注意反序列化的順序和序列化的順序完全一致

*

*@paramin

*@throwslOException

*/

@Override

publicvoidreadFields(DataInputin)throwslOException{

upFlow=in.readLong();

downFlow=in.readLongO;

sumFlow=in.readLong();

)

〃6要想把結果顯示在文件中，需要重寫loString。，且用分開，方便后續用

@Override

publicStringtoStringO{

returnupFlow+n\t"+downFlow+"\t"+sumFlow;

//7如果需要將自定義的bean放在key中傳輸，則還需要實現comparable接口，因

為mapreduce框中的shuffle過程一定會對key進行排序

@Override

publicintcompareTo(FlowBeano){

//倒序排列，從大到小

returnthis.sumFlow>o.getSumFlow()?-1:1;

2)案例實操

詳見3.2.1統計每一個手機號耗費的總上行流量、下行流量、總流量(序列化)。

2.2InputFormat數據切片機制

2.2.1FilelnputFormat切片機制

1)job提交流程源碼詳解

waitForCompletion()

submit();

//1建立連接

connect();

//I)創建提交job的代理

newCluster(getConfiguration());

//(1)判斷是本地yarn還是遠程

initialize(jobTrackAddr,conf)；

//2提交job

submitter.submitJobInternal(Job.this,cluster)

//I)創建給集群提交數據的Stag路徑

PathjobStagingArea=JobSubmissionFiles.ge04g山gZ)ir(cluster,conf);

//2)獲取jobid,并創建job路徑

JoblDjobld=submitClient.getNewJobID();

//3)拷貝jar包到集群

copyAndConfigureFiles(job,submitJobDir);

rUploader.uploadFiles(job,jobSubmitDir);

//4)計算切片，生成切片規劃文件

writeSplits(job,submitJobDir);

maps=writeNewSplits(job,jobSubmitDir);

input.getSplits(job);

//5)向Stag路徑寫xml配置文件

writeConf(conf,submitJobFile);

conf.writeXml(out);

//6)提交job,返回提交狀態

status=submitClient.submitJob(jobId,submitJobDir.toStringO,job.getCredentials());

2)FilelnputFormat源碼解析(input.getSplits(job))

（1）找到你數據存儲的目錄。

（2）開始遍歷處理（規劃切片）目錄下的每一個文件

（3）遍歷第一個文件ss.txt

a）獲取文件大小fs.sizeOf（ss.txt）;

b）計算切片大小

computeSliteSize（Math.max（minSize,Math.max（maxSize,blocksize）））=blocksize=128M

c）默認情況下，切片大小=blocksize

d）開始切，形成第1個切片：ss.txt—0:128M第2個切片ss.txt—128:256M第3

個切片ss.txt—256M:300M（每次切片時，都要判斷切完剩下的部分是否大于塊的1.1倍,

不大于1.1倍就劃分一塊切片）

e）將切片信息寫到一個切片規劃文件中

f）整個切片的核心過程在getSplit（）方法中完成。

g)數據切片只是在邏輯上對輸入數據進行分片，并不會再磁盤上將其切分成分片

進行存儲。InputSplit只記錄了分片的元數據信息，比如起始位置、長度以及所在的節

點列表等。

h)注意：block是HDFS上物理上存儲的存儲的數據，切片是對數據邏輯上的劃分。

(4)提交切片規劃文件到yarn±,yarn上的MrAppMaster就可以根據切片規劃文件計

算開啟maptask個數。

3)FilelnputFormat中默認的切片機制：

(1)簡單地按照文件的內容長度進行切片

(2)切片大小，默認等于block大小

(3)切片時不考慮數據集整體，而是逐個針對每一個文件單獨切片

比如待處理數據有兩個文件：

filel.txt320M

file2,txt10M

經過FilelnputFormat的切片機制運算后，形成的切片信息如下:

filel.txt.splitl-0?128

filel.txt.split2-128?256

filel.txt.split3—256?320

file2.txt.splitl—(MOM

4)FilelnputFormat切片大小的參數配置

(1)通過分析源碼，在FilelnputFormat中，計算切片大小的邏輯：Math.max(minSize,

Math.min(maxSize,blockSize));

切片主要由這幾個值來運算決定

mapreduce.input.fileinputformat.split.minsize=l默認值為I

mapreduce.input.fileinputformat.split.maxsize=Long.MAXValue默認值Long.MAXValue

因此，默認情況下，切片大小=1)1(?：1(^26。

maxsize(切片最大值)：參數如果調得比blocksize小，則會讓切片變小，而且就等于

配置的這個參數的值。

minsize(切片最小值)：參數調的比blockSize大，則可以讓切片變得比blocksize還

大。

5)獲取切片信息API

〃根據文件類型獲取切片信息

FileSplitinputSplit=(FileSplit)context.getInputSplit();

//獲取切片的文件名稱

Stringname=inputSplit.getPath().getName();

2.2.2CombineTextlnputFormat切片機制

關于大量小文件的優化策略

1)默認情況下Textinputformat對任務的切片機制是按文件規劃切片，不管文件多小，都會

是一個單獨的切片，都會交給一個maptask,這樣如果有大量小文件，就會產生大量的

maplask,處理效率極其低下。

2)優化策略

(1)最好的辦法，在數據處理系統的最前端(預處理/采集)，將小文件先合并成大文

件，再上傳到HDFS做后續分析。

(2)補救措施：如果已經是大量小文件在HDFS中了，可以使用另一種InpulFormat

來做切片(CombineTextlnputFormat),它的切片邏輯跟TextFilelnputFormat不同：它可以

將多個小文件從邏輯上規劃到一個切片中，這樣，多個小文件就可以交給一個maptask。

(3)優先滿足最小切片大小，不超過最大切片大小

CombineTextInputFormat.se/Mar/叩w/Sp/〃Size(job,4194304);//4m

CombineTextlnputFormat.2097152);//2m

舉例：0.5m+1m+0.3m+5m=2m+4.8m=2m+4m+0.8m

3)具體實現步驟

//9如果不設置InputFormat,它默認用的是TextlnputFormat.class

job.setlnputFormatClass(CombineTextlnputFormat.class)

CombineTextlnpulFormaLsefMax/叩”rSp〃rSize(job,4194304);//4m

CombineTextlnputFormat.se/Mi”/叩"/Sp/"Size(job,2097152);//2m

4)案例實操

詳見3.1.4需求4：大量小文件的切片優化(CombineTexlInpulFormal)。

2.2.3自定義InputFormat

1)概述

(1)自定義一個InputFormat

(2)改寫RecordReader,實現一次讀取一個完整文件封裝為KV

(3)在輸出時使用SequenceFileOutPutFormat輸出合并文件

2)案例實操

詳見3.5小文件處理(自定義InputFormat)。

2.3M叩Task工作機制

i)問題引出

maptask的并行度決定map階段的任務處理并發度，進而影響到整個job的處理速度。

那么，mapTask并行任務是否越多越好呢？

2)MapTask并行度決定機制

一個job的map階段MapTask并行度(個數)，由客戶端提交job時的切片個數決定。

3)MapTask工作機制

(1)Read階段：MapTask通過用戶編寫的RecordReader,從輸入InputSplit中解析出

一個個key/value。

(2)Map階段：該節點主要是將解析出的key/value交給用戶編寫map()函數處理，并

產生一系列新的key/value?

(3)Collect階段：在用戶編寫map()函數中，當數據處理完成后，一般會調用

OutputCollector.collect。輸出結果。在該函數內部，它會將生成的key/value分區(調用

Partitioner),并寫入一個環形內存緩沖區中。

(4)Spill階段：即“溢寫”，當環形緩沖區滿后，MapReduce會將數據寫到本地磁盤上,

生成一個臨時文件。需要注意的是，將數據寫入本地磁盤之前，先要對數據進行一次本地排

序，并在必要時對數據進行合并、壓縮等操作。

溢寫階段詳情：

步驟1：利用快速排序算法對緩存區內的數據進行排序，排序方式是，先按照分區編號

partition進行排序，然后按照key進行排序。這樣，經過排序后，數據以分區為單位聚集在

一起，且同一分區內所有數據按照key有序。

步驟2：按照分區編號由小到大依次將每個分區中的數據寫入任務工作目錄下的臨時文

件output/spillN.out(N表示當前溢寫次數)中。如果用戶設置了Combiner,則寫入文件之

前，對每個分區中的數據進行一次聚集操作。

步驟3：將分區數據的元信息寫到內存索引數據結構SpiDRecord中，其中每個分區的元

信息包括在臨時文件中的偏移量、壓縮前數據大小和壓縮后數據大小。如果當期內存索引大

小超過1MB,則將內存索引寫到文件output/spillN.out.index中。

(5)Combine階段：當所有數據處理完成后，MapTask對所有臨時文件進行一次合并,

以確保最終只會生成一個數據文件。

當所有數據處理完后，MapTask會將所有臨時文件合并成一個大文件，并保存到文件

output/file.out中，同時生成相應的索引文件output/file.out.index。

在進行文件合并過程中，MapTask以分區為單位進行合并。對于某個分區，它將采用多

輪遞歸合并的方式。每輪合并io.sort.factor（默認100）個文件，并將產生的文件重新加入

待合并列表中，對文件排序后，重復以上過程，直到最終得到一個大文件。

讓每個MapTask最終只生成一個數據文件，可避免同時打開大量文件和同時讀取大量

小文件產生的隨機讀取帶來的開銷。

2.4Shuffle機制

2.4.1Shuffle機制

M叩reduce確保每個reducer的輸入都是按鍵排序的。系統執行排序的過程（即將map

輸出作為輸入傳給reducer）稱為shuffleo

Copy"Sort"Reduce

phasephasephase

maptask端嚅reducetask

spilltodisk幽m...........?>1卜”A..

bufferinmerge1------------1

memoryi_________________________）口▼4n

■-?!|AllII『II1/merge位

/_n__u

s■'▼蹦tk

mergeJr?-----------1output

partitiom?/y

/mixtunofin-memoryandon-diskdata

Othermaps**..............Otherreduces

2.4.2MapReduce工作流程

1）流程示意圖

?向硅谷

ss.txt0128MapReduce詳細工作流程（一）

5默認TextlnputFormat,__Maptask

7環形緩沖區默認100M

RecorderReaderInputFormat>K,v

......<a,lxc.

K,vMapper!___

readerO6邏輯運

map(K,v)溢出到文件（分區且區內有序）

spiller9

待處理文本Context.write(k,v)

1Merwe底并排序

/user/inputHashPartitioner分區<a,lxc,X><b,l><b,l>

o」xa,lxc4xe,l>

ss.txtoutputcollector

Key.compareT。排序

200m

Combiner介并<a,lxc.l>va,2xGlxe,l><b,2xd,lxf,l>

2客戶端submit。前,獲取partitionOpartitionl

8分區、排序11合并

待處理數據的信息，然后ss.txt128-200

根據參數配置，形成一個