Hadoop——MapReduce（3）

MapReduce：自己處理業務相關代碼 + 自身的默認代碼

1.MapReduce優缺點

優點：

• 1、易于編程，用戶只關心業務邏輯；實作框架的介面
• 2、良好的擴展性，可以動態增加服務器，解決計算資源不夠問題
• 3、高容錯性，任何一臺機器掛掉，可以將任務轉移到其他節點
• 4、適合海量資料計算（TP/PB），幾千臺服務器共同計算

缺點：

• 1、不擅長實時計算，Mysql
• 2、不擅長流式計算，Sparkstreaming flink
• 3、不擅長DAG有向無環圖計算，spark

2.MapReduce行程

一個完整的MapReduce在分布式運行時有三類實體行程：

• MrAppMaster：負責整個程式的程序調度及狀態協調
• MapTask：負責Map階段的整個資料處理流程
• ReduceTask：負責Reduce階段的整個資料處理流程

3.序列化

• 序列化：將記憶體中的物件——>位元組序列（或其他資料傳輸協議），以便于存盤到磁盤（持久化）和網路傳輸
• 反序列化：將收到的位元組序列（或其他資料傳輸協議）->記憶體中的物件
• 為什么要序列化：序列化可以存盤“活的”物件，將“活的”物件發送到遠程計算機
• 為什么不用Java的序列化：Java的序列化是一個重量級序列化框架（Serializable）,一個物件被序列化后，會附帶很多額外的資訊（如：校驗資訊，Header，繼承體系），不便于在網路中高效傳輸，Hadoop因此開發了一套自己的序列化機制（Writable）
• Hadoop序列化的特點：
  • （1）緊湊：高效使用存盤空間
  • （2）快速：讀寫資料的額外開銷小
  • （3）互操作：支持多語言的互動

4 InputFormat資料輸入

4.1 切片與MapTask并行度決定機制

• 問題：MapTask的并行度決定Map階段的任務處理并發度，進而影響到整個Job的處理速度
• MapTask并行度決定機制
  • 數資料塊：Block是HDFS物理上把資料分成一塊一塊，資料塊是HDFS存盤資料單位
  • 資料切片：資料切片只是在邏輯上對輸入進行分片，并不會在磁盤上將其切分成片進行存盤，資料切片是MapReduce程式計算輸入資料的單位，一個切片會對應啟動一個MapTask

4.2 Job提交流程原始碼詳解

waitForCompletion()

submit();

// 1建立連接
	connect();	
		// 1）創建提交Job的代理
		new Cluster(getConfiguration());
			// （1）判斷是本地運行環境還是yarn集群運行環境
			initialize(jobTrackAddr, conf); 

// 2 提交job
submitter.submitJobInternal(Job.this, cluster)

	// 1）創建給集群提交資料的Stag路徑
	Path jobStagingArea = JobSubmissionFiles.getStagingDir(cluster, conf);

	// 2）獲取jobid ，并創建Job路徑
	JobID jobId = 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.toString(), job.getCredentials());

4.3 FileInputFormat 切片機制

切片機制：

• 1.簡單地按照檔案的內容長度進行切片
• 2.切片大小，默認等于Block大小
• 3.切片時不考慮資料集整體，而是逐個針對每一個檔案單獨切片

切片原始碼決議：

(1)原始碼中計算切片大小的公式
Math.max(minSize,Math.min(maxSize,blockSize))
mapreduce.input.fileinputformat.split.minsize=1	//默認值為1
mapreduce.input.fileinputformat.split.maxsize=Long.MAXValue	//默認值Long.MAXValue

(2)切片大小設定
maxsize(切片最大值)：引數如果調得比blockSize小，則會讓切片變小，此時切片大小=maxsize
minsize(切片最小值)：引數調的比blockSize大，讓切片變大,此時切片大小=minSize

(3)獲取切片資訊API
String name = inputSplit.getPath().getName();	//獲取切片的檔案名稱
FileSplit inputSplit = (FileSplit) contxt.getInputSplit();	//根據檔案型別獲取切片資訊

4.4 FileInputFormat

FileInputFormat常見的介面實作類：

TextInputFormat、KeyValueTextInputFormat、NLineInputFormat、CombineTextInputFormat和自定義InputFormat等

• TextInputFormat：FileInputFormat默認實作類，按行讀取每條記錄，
  • 鍵：存盤該行在整個檔案中的起始位元組偏移量，LongWritable型別
  • 值：改行內容，不包括任何行終止符（換行符和回車符），Text型別

4.5 CombineTextInputFormat切片機制

Combine使用案例
? 說明：框架默認的TextInputFormat切片機制是對任務按檔案規劃切片，不管檔案多小，都會是一個單獨的切片，都會交給一個MapTast，這樣如果有大量小檔案，就會產生大量的MapTask，處理效率及其低下，

(1)應用場景
CombineTextInputFormat用于小檔案過多的場景，它可以將多個小檔案從邏輯上規劃到一個切片中，這樣，多個小檔案交給一個MapTask處理

(2)虛擬存盤切片最大值設定
CombineTextInputFormat.setMaxInputSplitSize(job, 4194304);// 4m
注意：虛擬存盤切片最大值設定最好根據實際的小檔案大小情況來設定具體的值

(3)切片機制，生成切片程序包括：虛擬存盤程序和切片程序兩部分
	虛擬存盤程序：將輸入目錄下所有檔案大小，依次和設定的setMaxInputSplitSize值比較，如果不大于設定的最大	 值，邏輯上劃分一個塊，如果輸入檔案大于設定的最大值且大于兩倍，那么以最大值切割一塊；當剩余資料大小超過設定的最大值且不大于最大值2倍，此時將檔案均分成2個虛擬存盤塊（防止出現太小切片），
	例如setMaxInputSplitSize值為4M，輸入檔案大小為8.02M，則先邏輯上分成一個4M，剩余的大小為4.02M，如果按照4M邏輯劃分，就會出現0.02M的小的虛擬存盤檔案，所以將剩余的4.02M檔案切分成（2.01M和2.01M）兩個檔案
	
	切片程序：
（a）判斷虛擬存盤的檔案大小是否大于setMaxInputSplitSize值，大于等于則單獨形成一個切片，
（b）如果不大于則跟下一個虛擬存盤檔案進行合并，共同形成一個切片，
（c）測驗舉例：有4個小檔案大小分別為1.7M、5.1M、3.4M以及6.8M這四個小檔案，則虛擬存盤之后形成6個檔案塊，大小分別為：1.7M，（2.55M、2.55M），3.4M以及（3.4M、3.4M）
最侄訓形成3個切片，大小分別為：（1.7+2.55）M，（2.55+3.4）M，（3.4+3.4）M

5.MapReduce作業流程

6.Shffle機制

6.1Shuffle機制大綱流程

Map方法之后，Reduce方法之前的資料處理程序稱之為Shuffle

(1)MapTask收集我們的map方法輸出kv對，放到記憶體緩沖區中
(2)從記憶體緩沖區不斷溢位本地磁盤檔案，可能會溢位多個檔案
(3)多個溢位檔案會被合并成大的溢位檔案
(4)在溢位程序及合并的程序中，都要呼叫Partition進行磁區和針對key進行排序
(5)ReduceTask根據自己的磁區號，去各個MapTask機器上取相應的結果磁區資料
(6)ReduceTask會抓取到同一個磁區的來自不同MapTask的結果檔案，ReduceTask會將這些檔案再進行合并(歸并排序)
(7)合并成大檔案后，Shuffle的程序也就結束了，后面進入ReduceTask的邏輯運算程序(從檔案中取出一個一個的鍵值對Group，呼叫用戶自定義的reduce()方法)

注意：(1)Shuffle中的緩沖區大小會影響到MapReduce程式的執行效率，原則上說，緩沖區越大，磁盤io的次數越少，執行速度就越快，
(2)緩沖區的大小可以通過引數調整，引數：mapreduce.task.io.sort.mb默認100M

6.2 Partition磁區

Partition磁區的使用案例

1.問題引出：
要求將統計結果按照條件輸出到不同檔案中(磁區)

2.默認Partitioner磁區
public class HashPartitioner<K, V> extends Partitioner<K, V> {
	public int getPartition(K key, V value, int numReduceTasks) {
		return (key.hashCode() & Integer.MAX_VALUE) % numReduceTasks;
		}
}
默認磁區是根據key的hashCode對ReduceTask個數取模得到的，用戶沒法控制哪個key存盤到哪個磁區

3.自定義Partition步驟
	(1)自定義類繼承Partitioner，重寫getPartition方法
	public class CustomPartitioner extends Partitioner<Text, FlowBean> {
		@Override
		public int getPartition(Text key, FlowBean value, int numPartitions) {
		// 控制磁區代碼邏輯
		… …
			return partition;
		}
	}
	(2)在Job驅動中，設定自定義Partitioner
		job.setPartitionerClass(CustomPartitioner.class)
	(3)自定義Partition后，要根據自定義Partitioner的邏輯設定相應數量的ReduceTask
		job.setNumReduceTasks(5); //輸出到五個檔案,磁區設定為5

4.磁區總結
	(1)如果ReduceTask的數量 > getPartition的結果數，則會多產出幾個空的輸出檔案part-r-000xx;
	(2)如果1 < ReduceTask的數量 < getPartition的結果數，則有一部分磁區資料無處安放，會Exception;
	(3)如果ReduceTask的數量=1，則不管MapTask端出數多少個磁區檔案，最終結果都交給這一個ReduceTask，最終也就只會產出一個結果檔案part-r-00000;
	(4)磁區號必須從零開始，逐一累加

5.案例分析
例如：假設自定義磁區數為5，則
	(1)job.setNumReduceTasks(1);	會正常運行，只不過會產生一個輸出檔案
	(2)job.setNumReduceTasks(2);	會報錯
	(3)job.setNumReduceTasks(6);	大于5，程式會正常運行，會產生空檔案

6.3 WritableComparable排序

WritableComparable排序的使用案例

排序是MapReduce框架中最重要的操作之一
	MapTask和ReduceTask均會對資料按照key進行排序，該操作屬于Hadoop的默認行為，任何應用程式中的資料均會被排序，而不管邏輯上是否需要，(默認排序是按照字典順序排序，且實作該排序的方法是快速排序)
	對于MapTask，它會將處理的結果暫時存放到環形緩沖區中，當環形緩沖區使用率達到一定閾值后，再對緩沖區中的資料進行一次快速排序，并將這些有序資料溢寫到磁盤上，而當資料處理完畢后，它會對磁盤上所有檔案進行歸并排序
	對于ReduceTask，它每個MapTask上遠程拷貝相應的資料檔案，如果檔案大小超過一定閾值，則溢寫到磁盤上，否則存盤在記憶體中，如果磁盤上檔案數目達到一定閾值，則進行一次歸并排序生成一個更大檔案；如果記憶體中檔案大小或者數目超過一定閾值，則進行一次合并后將資料溢寫到磁盤上，當所有資料拷貝完后，ReduceTask統一對記憶體和磁盤上的所有資料進行一次歸并排序
	
排序分類：
	(1)部分排序：MapReduce根據輸入記錄的鍵對資料集排序，保證輸出的每個檔案內部有序
	(2)全排序：最終輸出結果只有一個檔案，且檔案內部有序，實作方式是只設定一個ReduceTask，但該方法在處理大型檔案時效率極低，因為一臺機器處理所有檔案，完全喪失了MapReduce所提供的并行架構，
	(3)輔助排序(GroupingComparator分組)：在Reduce端對key進行分組，在接收的key為bean物件時，想讓一個或幾個欄位相同的key進入到同一個reduce方法時，可以采用分組排序
	(4)二次排序：自定義排序程序中，如果compareTo中的判斷條件為兩個即為二次排序

6.4 Combiner合并

Combiner合并的使用案例

1.Combiner是MR程式中Mapper和Reducer之外的一種組件
2.Combiner組件的父類是Reducer
3.Combiner和Reducer的區別在于運行的位置
	Combiner是在每一個MapTask所在的節點運行
	Reducer是在接收全域所有Mapper的輸出結果
4.Combiner的意義就是對每一個MapTask的輸出進行區域匯總，以減小網路傳輸量
5.Combiner能夠應用的前提是不能影響最終的業務邏輯，而且Combiner的輸出kv與Reducer的輸入kv型別要對應
	使用場景(求和√，求均值×)
求均值不行的例子：	Mapper			Reducer
				3 5 7 ->5			3 5 7 2 6 -> 23/5 正確的
				2 6 ->4				4 5 -> 9/2 使用場景錯誤導致不同的輸出
6.自定義Combiner實作步驟
	(a)自定義一個Combiner繼承Reducer，重寫Reduce方法
	public class WordCountCombiner extends Reducer<Text, IntWritable, Text,IntWritable> {
 		private IntWritable outV = new IntWritable();
 		@Override
 		protected void reduce(Text key, Iterable<IntWritable> values, Contextcontext) throws IOException, InterruptedException {
 			int sum = 0;
 			for (IntWritable value : values) {
			 	sum += value.get();
 			}

			 outV.set(sum);
 			context.write(key,outV);
 		}
	}
	(b)在Job驅動類中設定
     job.setCombinerClass(WordCountCombiner.class);

7.OutputFormat資料輸出

OutputFormat案例實操

OutputFormat是MapReduce輸出的基類，所有實作MapReduce輸出都實作了OutputFormat介面，默認格式TextOutputFormat

FileOutputFormat(含TextOutputFormat,SequenceFileOutputFormat,MapFileOutputFormat),FilterOutputFormat,DBOutputFormat

自定義OutPutFormat
1.應用場景：輸出資料到MySQL/HBase/Elasticsearch等存盤框架中
2.自定義OutputFormat步驟
	a.自定義一個類繼承FileOutputFormat
	b.改寫RecordWriter,具體改寫輸出資料的方法write()

8.MapTask作業機制

(1)Read階段：MapTask通過InputFormat獲得的RecordReader，從輸入InputSplit中決議處一個個key/value
(2)Map階段：該節點主要是將決議出的key/value交給用戶撰寫map()函式處理，并產生一系列新的key/value
(3)Collect收集階段：在用戶撰寫map()函式中，當資料處理完成后，一般會呼叫OutputCollector.collect()輸出結果，在該函式內部，它將會生成的key/value磁區(呼叫Partition)，并寫入一個環形記憶體緩沖區中
(4)Spill(溢寫)階段：當環形緩沖區滿后，MapReducer會將資料寫到本地磁盤上，生成一個臨時檔案，需要注意的是，將資料寫入本地磁盤之前，先要對資料進行一次本地排序，并在必要時對資料進行合并、壓縮等操作

	步驟1：利用快速排序演算法對緩沖區的資料進行排序，排序方式是，先按照磁區編號Partition進行排序，然后按照key進行排序，這樣，經過排序后，資料以磁區為單位聚集在一起，且同一磁區內所有資料按照key有序
	步驟2：安裝磁區編號由小到大依次將每個磁區中的資料寫入任務作業目錄下的臨時檔案output/spillN.out(N表示當前溢寫次數)中，如果用戶設定Combiner，則寫入檔案之前，對每個磁區中資料進行一次聚集操作，
	步驟3：將磁區資料的元資訊寫到記憶體索引資料結構SpillRecord中，其中每個磁區的元資訊包括在臨時檔案中的偏移量、壓縮前資料大小后壓縮后資料大小，如果當前記憶體索引大小1MB，則將記憶體索引寫到檔案output/spillN.out.index中

(5)Merge階段：當所有資料處理完成后，MapTask對所有臨時檔案進行一次合并，以確保最終只會生成一個資料檔案，
	當所有資料處理完后，MapTask會將所有臨時檔案合并成一個大檔案，并保持到檔案output/file.out中，同時生成相應的索引檔案output/file.out.index
	在進行檔案合并程序中，MapTask以磁區為單位進行合并，對于某個磁區，它將采用多輪遞回合并的方式，每輪合并mapreduce.task.io.facyor(默認10)個檔案，并將產生的檔案重新加入待合并串列中，對檔案排序后，重復以上程序，直到最終得到一個大檔案，
	讓MapTask最終只生成一個資料檔案，可避免同時打開大量檔案和同時讀取大量小檔案產生的隨機讀取帶來的開銷

9.ReduceTask作業機制

(1)Copy階段：ReduceTask從各個MapTask上遠程拷貝一片資料，并針對某一片資料，如果其大小超過一定閾值，則寫到磁盤上，否則直接放到記憶體中，
(2)Sort階段：在遠程拷貝資料的同時，ReduceTask啟動了兩個后臺執行緒對記憶體和磁盤上的檔案進行合并，以防止記憶體使用過多或磁盤上檔案過多，按照MapReduce語意，用戶撰寫reduce()函式輸入資料時按key進行聚集的一組資料，為了將key相同的資料聚在一起，Hadoop采用了基于排序的策略，由于各個MapTask已經實作對自己的處理結果進行了區域排序，因此，ReduceTask只需對所有資料進行一次歸并排序即可
(3)Reduce階段：reduce()函式將計算結果寫到HDFS上

10 MapTask & ReduceTask原始碼決議

=================== MapTask ===================
context.write(k, NullWritable.get()); //自定義的 map 方法的寫出，進入
	output.write(key, value);
		//MapTask727 行，收集方法，進入兩次
		collector.collect(key, value,partitioner.getPartition(key, value, partitions));
			HashPartitioner(); //默認磁區器
		collect() //MapTask1082 行 map 端所有的 kv 全部寫出后會走下面的 close 方法
			close() //MapTask732 行
				collector.flush() // 溢位刷寫方法，MapTask735 行，提前打個斷點，進入
					sortAndSpill() //溢寫排序，MapTask1505 行，進入
						sorter.sort() QuickSort //溢寫排序方法，MapTask1625 行，進入
					mergeParts(); //合并檔案，MapTask1527 行，進入
				collector.close(); //MapTask739 行,收集器關閉,即將進入 ReduceTask
=================== ReduceTask ===================
if (isMapOrReduce()) //reduceTask324 行，提前打斷點
initialize() // reduceTask333 行,進入
init(shuffleContext); // reduceTask375 行,走到這需要先給下面的打斷點
	totalMaps = job.getNumMapTasks(); // ShuffleSchedulerImpl 第 120 行，提前打斷點
 		merger = createMergeManager(context); //合并方法，Shuffle 第 80 行
			// MergeManagerImpl 第 232 235 行，提前打斷點
			this.inMemoryMerger = createInMemoryMerger(); //記憶體合并
			this.onDiskMerger = new OnDiskMerger(this); //磁盤合并
rIter = shuffleConsumerPlugin.run();
		eventFetcher.start(); //開始抓取資料，Shuffle 第 107 行，提前打斷點
		eventFetcher.shutDown(); //抓取結束，Shuffle 第 141 行，提前打斷點
		copyPhase.complete(); //copy 階段完成，Shuffle 第 151 行
		taskStatus.setPhase(TaskStatus.Phase.SORT); //開始排序階段，Shuffle 第 152 行
	sortPhase.complete(); //排序階段完成，即將進入 reduce 階段 reduceTask382 行
reduce(); //reduce 階段呼叫的就是我們自定義的 reduce 方法，會被呼叫多次
	cleanup(context); //reduce 完成之前，會最后呼叫一次 Reducer 里面的 cleanup 方法

11.MapReduce開發總結

	1、InputFormat
		1）默認的是TextInputformat  kv  key偏移量，v :一行內容
		2）處理小檔案CombineTextInputFormat 把多個檔案合并到一起統一切片
	2、Mapper 
		setup()初始化；  map()用戶的業務邏輯； clearup() 關閉資源；
	3、磁區
		默認磁區HashPartitioner ，默認按照key的hash值%numreducetask個數
		自定義磁區
	4、排序
		1）部分排序  每個輸出的檔案內部有序，
		2）全排序：  一個reduce ,對所有資料大排序，
		3）二次排序：  自定義排序范疇， 實作 writableCompare介面， 重寫compareTo方法
			總流量倒序  按照上行流量 正序
	5、Combiner 
		前提：不影響最終的業務邏輯（求和 沒問題   求平均值）
		提前聚合map  => 解決資料傾斜的一個方法
	6、Reducer
		用戶的業務邏輯；
		setup()初始化；reduce()用戶的業務邏輯； clearup() 關閉資源；
	7、OutputFormat
		1）默認TextOutputFormat  按行輸出到檔案
		2）自定義