1 Split階段

?????首先，接到hdf檔案輸入,在mapreduce中的map task開始之前，將檔案按照指定的大小切割成若干個部分，每一部分稱為一個split，默認是split的大小與block的大小相等，均為128MB，split大小由minSize、maxSize、blocksize決定，以wordcount代碼為例，以下是main()方法
在這里插入圖片描述
進入waitForCompletion(true)方法，進入submit()方法

找到 return submitter .submitJobInternal(Job.this, cluster);
進入，找到 int maps = writeSplits(job, submitJobDir);

進入writeNewSplits()方法

?????進入writeNewSplits()方法，可以看出該方法首先獲取splits陣列資訊后，排序，將會優先處理大檔案，最侄訓傳mapper數量，這其中又分為兩部分：確定切片數量和寫入切片資訊，確定切片數量的任務交由FileInputFormat的getSplits(job)完成，寫入切片資訊的任務交由JobSplitWriter.createSplitFiles(jobSubmitDir, conf, jobSubmitDir.getFileSystem(conf), array)方法，該方法會將切片資訊和SplitMetaInfo都寫入HDFS中，return array.length;回傳的是map任務數，默認map的數量是： default_num = total_size / block_size;
?????實際的mapper數量就是輸入切片的數量，而切片的數量又由使用的輸入格式決定，默認為TextInputFormat，該類為FileInputFormat的子類，確定切片數量的任務交由FileInputFormat的getSplits(job)完成，FileInputFormat繼承自抽象類InputFormat，該類定義了MapReduce作業的輸入規范，其中的抽象方法List getSplits(JobContext context)定義了如何將輸入分割為InputSplit，不同的輸入有不同的分隔邏輯，而分隔得到的每個InputSplit交由不同的mapper處理，因此該方法的回傳值確定了mapper的數量，

2 Map階段

?????每個map task都有一個記憶體緩沖區， map的輸出結果先寫到記憶體中的環形緩沖區，緩沖區為100M,不斷的向緩沖區力寫資料，當達到80M時，需要將緩沖區中的資料以一個臨時檔案的方式存到磁盤，當整個map task結束后再對磁盤中這個map task所產生的所有臨時檔案做合并，生成最終的輸出檔案，最后，等待reduce task來拉取資料，當然，如果map task的結果不大，能夠完全存盤到記憶體緩沖區，且未達到記憶體緩沖區的閥值，那么就不會有寫臨時檔案到磁盤的操作，也不會有后面的合并，在寫入的程序中會進行磁區、排序、combine操作，
?????環形緩沖區：是使用指標機制把記憶體中的地址首尾相接形成一個存盤中間資料的快取區域，默認100MB;80M閾值,20M緩沖區,是為了解決寫入環形緩沖區資料的速度大于寫出到spill檔案的速度是資料的不丟失;Spill檔案：spill檔案是環形緩沖區到達閾值后寫入到磁盤的單個檔案.這些檔案在map階段計算結束時,會合成分好區的一個merge檔案供給給reduce任務抓取;spill檔案過小的時候,就不會浪費io資源合并merge;默認情況下3個以下spill檔案不合并;對于在環形緩沖區中的資料,最終達不到80m但是資料已經計算完畢的情況,map任務將會呼叫flush將緩沖區中的資料強行寫出spill檔案，

?????經過map類處理后，輸出到記憶體緩沖區(默認大小100M)，超過一定大小后，檔案溢寫到磁盤上，按照key分類
在這里插入圖片描述
按照key合并成大檔案，減少網路開銷

2.1磁區

看一下MapReduce自帶的磁區器HashPartitioner
在這里插入圖片描述
假設有聽個reduce任務，則磁區的計算如下：

2.2排序

在對map結果進行磁區之后，對于落在相同的磁區中的鍵值對，要進行排序，

3 Shuffle階段

?????Shuffle程序是MapReduce的核心，描述著資料從map task輸出到reduce task輸入的這段程序，reducetask根據自己的磁區號，去各個maptask磁區機器上取相應的結果磁區資料，reducetask會將這些檔案再進行合并（歸并排序)，
?????所有相同key的資料匯集到一個partition
在這里插入圖片描述
?????將相同的key value匯聚到一起，但不計算

4 Reduce階段

reduce階段分三個步驟：
抓取，合并，排序
?????1 reduce 任務會創建并行的抓取執行緒（fetcher）負責從完成的map任務中獲取結果檔案，是否完成是通過rpc心跳監聽，通過http協議抓取；默認是5個抓取執行緒，可調，為了是整體并行，在map任務量大，磁區多的時候，抓取執行緒調大；
?????2 抓取過來的資料會先保存在記憶體中，如果記憶體過大也溢位，不可見，不可調，但是單位是每個merge檔案，不會切分資料；每個merge檔案都會被封裝成一個segment的物件，這個物件控制著這個merge檔案的讀取記錄操作，有兩種情況出現：在記憶體中有merge資料 ?在溢寫之后存到磁盤上的資料 ?通過建構式的區分，來分別創建對應的segment物件
?????3 這種segment物件會放到一個記憶體佇列中MergerQueue，對記憶體和磁盤上的資料分別進行合并，記憶體中的merge對應的segment直接合并，磁盤中的合并與一個叫做合并因子的factor有關（默認是10）
?????4 排序問題，MergerQueue繼承輪換排序的介面，每一個segment 是排好序的，而且按照key的值大小邏輯（和真的大小沒關系）；每一個segment的第一個key都是邏輯最小，而所有的segment的排序是按照第一個key大小排序的，最小的在前面，這種邏輯總能保證第一個segment的第一個key值是所有key的邏輯最小檔案合并之后，最終交給reduce函式計算的，是MergeQueue佇列，每次計算的提取資料邏輯都是提取第一個segment的第一個key和value資料，一旦segment被呼叫了提取key的方法，MergeQueue佇列將會整體重新按照最小key對segment排序，最終形成整體有序的計算結果；
在這里插入圖片描述

partition 、Reduce、輸出檔案數量相等

Reduce任務數量
在大資料量的情況下，如果只設定1個Reduce任務，其他節點將被閑置，效率底下所以將Reduce設定成一個較大的值（max：72）.調節Reduce任務數量的方法一個節點的Reduce任務數并不像Map任務數那樣受多個因素制約

通過引數調節mapred.reduce.tasks（在組態檔中）
在代碼中呼叫job.setNumReduceTasks(int n)方法（在code中）

轉載請註明出處，本文鏈接：https://www.uj5u.com/shujuku/39983.html

標籤：大數據

上一篇：svaeas問題

下一篇：pb 如何判斷excel 檔案是否打開