《大數(shù)據(jù)平臺(tái)部署與運(yùn)維》課件 單元5 MapReduce實(shí)現(xiàn)電商銷售數(shù)據(jù)統(tǒng)計(jì)

《大數(shù)據(jù)平臺(tái)部署與運(yùn)維》單元5MapReduce實(shí)現(xiàn)電商銷售數(shù)據(jù)統(tǒng)計(jì)任務(wù)5.1認(rèn)識(shí)MapReduce01理解MapReduce的原理與體系架構(gòu)02掌握YARN運(yùn)行機(jī)制學(xué)習(xí)目標(biāo)任務(wù)5.1認(rèn)識(shí)MapReduce【任務(wù)場(chǎng)景】經(jīng)理：小張，我們后臺(tái)數(shù)據(jù)量越來(lái)越大，服務(wù)器計(jì)算性能不佳，計(jì)算效率是個(gè)問題，你有什么好的建議嗎？小張：Hadoop的核心組件MapReduce可以用來(lái)做大規(guī)模數(shù)據(jù)集的并行運(yùn)算，我們已經(jīng)有了Hadoop集群，不妨試試MapReduce。經(jīng)理：嗯，沒錯(cuò)，MapReduce可以并行拆分和處理TB級(jí)數(shù)據(jù)，運(yùn)行在普通服務(wù)器組成的集群上也能保證快速高效地處理海量數(shù)據(jù)，你先了解一下MapReduce的原理和運(yùn)行機(jī)制吧。小張：好的。任務(wù)5.1認(rèn)識(shí)MapReduce【任務(wù)布置】MapReduce的運(yùn)行依賴于JDK和Hadoop，因此必須將Hadoop的基礎(chǔ)環(huán)境提前安裝好，才能進(jìn)行MapReduce的運(yùn)行和操作。本任務(wù)要求在前面已經(jīng)完成安裝部署Hadoop平臺(tái)的node1節(jié)點(diǎn)上完成，要求理解MapReduce的原理和體系架構(gòu)；理解YARN運(yùn)行機(jī)制；最終在node1上運(yùn)行MapReduce自帶的單詞計(jì)數(shù)程序，查看運(yùn)行結(jié)果。任務(wù)5.1認(rèn)識(shí)MapReduce5.1.1MapReduce介紹MapReduce是一種分布式計(jì)算模型，由Google提出，起初主要用于搜索領(lǐng)域，解決海量數(shù)據(jù)的計(jì)算問題。MapReduce是Hadoop框架內(nèi)的一種編程模型，用于訪問存儲(chǔ)在Hadoop文件系統(tǒng)(HDFS)中的大數(shù)據(jù)，它是一個(gè)核心組件，是Hadoop框架功能不可或缺的一部分。MapReduce是面向大數(shù)據(jù)并行處理的計(jì)算模型、框架和平臺(tái)，它隱含了以下三層含義：（1）MapReduce是一個(gè)基于集群的高性能并行計(jì)算平臺(tái)（ClusterInfrastructure）。（2）MapReduce是一個(gè)并行計(jì)算與運(yùn)行軟件框架（SoftwareFramework）。（3）MapReduce是一個(gè)并行程序設(shè)計(jì)模型與方法（ProgrammingModel&Methodology）。任務(wù)5.1認(rèn)識(shí)MapReduceMapReduce原理（1）基本概念用一個(gè)比較形象的例子解釋MapReduce：我們要數(shù)圖書館中的所有書。你數(shù)1號(hào)書架，我數(shù)2號(hào)書架。這就是“Map”。我們?nèi)嗽蕉?，?shù)書就更快。現(xiàn)在我們到一起，把所有人的統(tǒng)計(jì)數(shù)加在一起。這就是“Reduce”。MapReduce由兩個(gè)階段組成：Map（映射）和Reduce（歸納），用戶只需要實(shí)現(xiàn)map()和reduce()兩個(gè)函數(shù)，即可實(shí)現(xiàn)分布式計(jì)算。這兩個(gè)函數(shù)的形參是key、value對(duì)，表示函數(shù)的輸入信息。任務(wù)5.1認(rèn)識(shí)MapReduceMapReduce原理（2）映射和歸納簡(jiǎn)單說(shuō)來(lái)，一個(gè)映射函數(shù)就是對(duì)一些獨(dú)立元素組成的概念上的列表的每一個(gè)元素進(jìn)行指定的操作。事實(shí)上，每個(gè)元素都是被獨(dú)立操作的，而原始列表沒有被更改，因?yàn)檫@里創(chuàng)建了一個(gè)新的列表來(lái)保存新的答案。這就是說(shuō)，Map操作是可以高度并行的，這對(duì)高性能要求的應(yīng)用以及并行計(jì)算領(lǐng)域的需求非常有用。而歸納操作指的是對(duì)一個(gè)列表的元素進(jìn)行適當(dāng)?shù)暮喜?。雖然它不如映射函數(shù)那么并行，但是因?yàn)榛?jiǎn)總是有一個(gè)簡(jiǎn)單的答案，大規(guī)模的運(yùn)算相對(duì)獨(dú)立，所以化簡(jiǎn)函數(shù)在高度并行環(huán)境下也很有用。任務(wù)5.1認(rèn)識(shí)MapReduceMapReduce優(yōu)勢(shì)（1）易于理解MapReduce通過(guò)抽象模型和計(jì)算框架把需要做什么和具體做什么分開了，為開發(fā)者提供了一個(gè)抽象和高層的編程接口和框架，開發(fā)者僅需關(guān)心其他應(yīng)用層的具體計(jì)算問題，大大降級(jí)了開發(fā)者使用時(shí)的心智負(fù)擔(dān)。

（2）良好的擴(kuò)展性基于MapReduce的計(jì)算性能可以隨節(jié)點(diǎn)數(shù)目增長(zhǎng)保持近似于線性的增長(zhǎng)，這個(gè)特點(diǎn)是MapReduce處理海量數(shù)據(jù)的關(guān)鍵，通過(guò)將計(jì)算節(jié)點(diǎn)增至幾百或者幾千可以很容易地處理數(shù)百TB甚至PB級(jí)別的離線數(shù)據(jù)。

（3）分布可靠MapReduce通過(guò)把對(duì)數(shù)據(jù)集的大規(guī)模操作分發(fā)給網(wǎng)絡(luò)上的每個(gè)節(jié)點(diǎn)實(shí)現(xiàn)可靠性；每個(gè)節(jié)點(diǎn)會(huì)周期性地返回它所完成的工作和最新狀態(tài)。如果一個(gè)節(jié)點(diǎn)保持沉默超過(guò)一個(gè)預(yù)設(shè)的時(shí)間間隔，主節(jié)點(diǎn)記錄下這個(gè)節(jié)點(diǎn)狀態(tài)為死亡，并把分配給這個(gè)節(jié)點(diǎn)的數(shù)據(jù)發(fā)到別的節(jié)點(diǎn)。

任務(wù)5.1認(rèn)識(shí)MapReduce5.1.2

MapReduce體系架構(gòu)MapReduce1.0

采用了Manager/Worker（M/W）架構(gòu)。它主要由以下幾個(gè)組件組成：Client、JobTracker、TaskTracker和Task。任務(wù)5.1認(rèn)識(shí)MapReduceMapReduce1.0體系架構(gòu)介紹（1）Client用戶編寫的MapReduce程序通過(guò)Client提交到JobTracker端

；同時(shí)，用戶可通過(guò)Client提供的一些接口查看作業(yè)運(yùn)行狀態(tài)。（2）JobTrackerJobTracker主要負(fù)責(zé)資源監(jiān)控和作業(yè)調(diào)度。（3）TaskTracker

TaskTracker會(huì)周期性地通過(guò)Heartbeat（心跳檢測(cè)）將本節(jié)點(diǎn)上資源的使用情況和任務(wù)的運(yùn)行進(jìn)度匯報(bào)給JobTracker，同時(shí)接收J(rèn)obTracker發(fā)送過(guò)來(lái)的命令并執(zhí)行相應(yīng)的操作（如啟動(dòng)新任務(wù)、殺死任務(wù)等）。（4）TaskTask分為MapTask和ReduceTask兩種，均由TaskTracker啟動(dòng)。任務(wù)5.1認(rèn)識(shí)MapReduceMapReduce2.0體系架構(gòu)Hadoop2.0新引入的資源管理系統(tǒng)，是直接從MapReducev1演化而來(lái)的；核心思想：將MapReducev1中JobTracker的資源管理和任務(wù)調(diào)度兩個(gè)功能分開，分別由ResourceManager和進(jìn)程實(shí)現(xiàn)。任務(wù)5.1認(rèn)識(shí)MapReduceMapReduce2.0體系架構(gòu)介紹（1）Client與MapReducev1的Client類似，用戶通過(guò)Client與YARN交互，提交MapReduce作業(yè)，查詢作業(yè)運(yùn)行狀態(tài)，管理作業(yè)等。（2）ResourceManagerResourceManager是一個(gè)全局的資源管理器，負(fù)責(zé)整個(gè)系統(tǒng)的資源管理和分配。它主要由兩個(gè)組件構(gòu)成：調(diào)度器和應(yīng)用程序管理器。（3）NodeManagerNodeManager是每個(gè)節(jié)點(diǎn)上的資源和任務(wù)管理器，一方面，它會(huì)定時(shí)地向ResourceManager匯報(bào)本節(jié)點(diǎn)上的資源使用情況和各個(gè)Container的運(yùn)行狀態(tài)；另一方面，它接收并處理來(lái)自ApplicationMaster的Container啟動(dòng)/停止等各種請(qǐng)求。任務(wù)5.1認(rèn)識(shí)MapReduceMapReduce2.0體系架構(gòu)介紹（4）ApplicationMaster

ApplicationMaster功能類似于1.0中的JobTracker，但不負(fù)責(zé)資源管理。功能包括：任務(wù)劃分、任務(wù)調(diào)度、任務(wù)狀態(tài)監(jiān)控和容錯(cuò)。（5）ContainerContainer是YARN中的資源抽象，它封裝了某個(gè)節(jié)點(diǎn)上的多維度資源，如內(nèi)存、CPU、磁盤、網(wǎng)絡(luò)等，當(dāng)ApplicationMaster向ResourceManager申請(qǐng)資源時(shí)，ResourceManager為ApplicationMaster返回的資源便是用Container表示的。（6）MapTask/ReduceTaskMapTask/ReduceTask周期性向ApplicationMaster匯報(bào)心跳。一旦Task掛掉，則ApplicationMaster將為之重新申請(qǐng)資源，并重新運(yùn)行。任務(wù)5.1認(rèn)識(shí)MapReduceMapReduce任務(wù)執(zhí)行過(guò)程（1）Map任務(wù)執(zhí)行過(guò)程MapTask執(zhí)行過(guò)程如下圖所示。由該圖可知，MapTask先將對(duì)應(yīng)的split迭代解析成一個(gè)個(gè)key/value對(duì)，依次調(diào)用用戶自定義的map()函數(shù)進(jìn)行處理，最終將臨時(shí)結(jié)果存放到本地磁盤上，其中臨時(shí)數(shù)據(jù)被分成若干個(gè)partition(分片)，每個(gè)partition將被一個(gè)ReduceTask處理。任務(wù)5.1認(rèn)識(shí)MapReduceMapReduce任務(wù)執(zhí)行過(guò)程（2）Reduce任務(wù)執(zhí)行過(guò)程ReduceTask執(zhí)行過(guò)程如下圖所示。該過(guò)程分為三個(gè)階段：①?gòu)倪h(yuǎn)程節(jié)點(diǎn)上讀取MapTask中間結(jié)果（稱為“Shuffle階段”）；②按照key對(duì)key/value對(duì)進(jìn)行排序（稱為“Sort階段”）；③依次讀取<key,valuelist>，調(diào)用用戶自定義的reduce()函數(shù)處理，并將最終結(jié)果存到HDFS上（稱為“Reduce階段”）。任務(wù)5.1認(rèn)識(shí)MapReduce5.1.3MapReduce發(fā)展現(xiàn)狀以前，在Google，MapReduce用在非常廣泛的應(yīng)用程序中，MapReduce實(shí)現(xiàn)以后，它被用來(lái)重新生成Google的整個(gè)索引，并取代老的adhoc程序去更新索引。Hadoop解決了有無(wú)問題。很快人們發(fā)現(xiàn)MapReduce復(fù)雜度很高，即使技術(shù)實(shí)力強(qiáng)大如Facebook都很難寫出高效正確的MapReduce程序。此外除了解決批處理問題，人們需要Hadoop能解決其遇到的交互式查詢?nèi)蝿?wù)。為此，F(xiàn)acebook開發(fā)了Hive，該項(xiàng)目快速流行起來(lái)，到現(xiàn)在還有很多用戶。Facebook當(dāng)時(shí)更是高達(dá)95%的用戶使用Hive而不是裸寫MapReduce程序。但是MapReduce的思想和技術(shù)原理還是值得我們學(xué)習(xí)。馬里蘭大學(xué)教授、2010年出版的《Data-IntensiveTextProcessingwithMapReduce》一書的作者JimmyLin在書中提出：MapReduce改變了我們組織大規(guī)模計(jì)算的方式，它代表了第一個(gè)有別于馮·諾依曼結(jié)構(gòu)的計(jì)算模型，是在集群規(guī)模而非單個(gè)機(jī)器上組織大規(guī)模計(jì)算的新的抽象模型上的第一個(gè)重大突破，是到所見到的最為成功的基于大規(guī)模計(jì)算資源的計(jì)算模型。任務(wù)5.1認(rèn)識(shí)MapReduce5.1.4YARN運(yùn)行機(jī)制Hadoop主要組件有HDFS和YARN，HDFS是分布式文件系統(tǒng)，主要是進(jìn)行文件的存儲(chǔ)，而YARN是Hadoop集群資源管理系統(tǒng)，支持分布式計(jì)算模式。YARN上運(yùn)行一個(gè)應(yīng)用的步驟：（1）客戶端聯(lián)系資源管理器，要求運(yùn)行一個(gè)applicationmaster進(jìn)程。（2）資源管理其找到一個(gè)能夠在容器啟動(dòng)applicationmaster的節(jié)點(diǎn)管理器。（3）applicationmaster運(yùn)行后，根據(jù)應(yīng)用本身向資源管理器請(qǐng)求更多容器。（4）資源管理器給applicationmaster分配需要的資源后，applicationmaster在對(duì)應(yīng)資源節(jié)點(diǎn)管理器啟動(dòng)容器，節(jié)點(diǎn)管理器獲取任務(wù)運(yùn)行需要的resources后，在該容器運(yùn)行任務(wù)。任務(wù)5.1認(rèn)識(shí)MapReduceYARN調(diào)度理想情況下YARN應(yīng)用發(fā)出資源請(qǐng)求會(huì)立刻給予滿足，但現(xiàn)實(shí)是資源是有限的，通常需要等待才能得到所需資源，因此YARN提供多種調(diào)度器和可配置策略來(lái)供選擇。（1）FIFO調(diào)度器FIFO調(diào)度器將應(yīng)用放置在一個(gè)隊(duì)列中，按照提交順序運(yùn)行應(yīng)用，首先為隊(duì)列中第一個(gè)應(yīng)用請(qǐng)求分配資源，第一個(gè)應(yīng)用請(qǐng)求被滿足后再依次為隊(duì)列下一個(gè)應(yīng)用服務(wù)。（2）容器調(diào)度器(CapacityScheduler)

使用容量調(diào)度器時(shí)，一個(gè)獨(dú)立的專門隊(duì)列保證小作業(yè)一提交就啟動(dòng)，由于隊(duì)列容量是為固定隊(duì)列中作業(yè)所保留的，這種策略是以整個(gè)集群利用率為代價(jià)，相比于FIFO調(diào)度器相比，大作業(yè)執(zhí)行時(shí)間更長(zhǎng)。（3）公平調(diào)度器(FairScheduler)

使用公平調(diào)度器時(shí)，不需要預(yù)留一定資源，調(diào)度器會(huì)在所有運(yùn)行的作業(yè)之間動(dòng)態(tài)平衡資源。任務(wù)5.1認(rèn)識(shí)MapReduce【工作流程】1.新建本地文件，編輯文件內(nèi)容。2.上傳文件到hdfs。3.運(yùn)行自帶jar包，查看運(yùn)行結(jié)果。任務(wù)5.1認(rèn)識(shí)MapReduce【操作步驟】1.

在本地/home/hdfs目錄下新建文件hello。文件中輸入如下內(nèi)容2.

將hello文件上傳至HDFS根目錄下：3.運(yùn)行jar包。在Hadoop的安裝目錄下有一個(gè)示例jar包，地址為：/usr/local/hadoop/share/hadoop/mapreduce/hadoop-mapreduce-examples-3.1.1.jar，里面有很多框架自帶的例子。使用以下命令查看jar包中包含的例子：使用以下命令運(yùn)行自帶的wordcount程序：[root@master01/]#mkdir/home/hdfs[root@master01/]#cd/home/hdfs[root@master01/]#vimhelloHelloHDFSHelloMapReduce[root@master01/]#hdfsdfs-put/home/hdfs/hello/[root@master01/]#hadoopjarhadoop-mapreduce-examples-3.1.1.jar[root@master01/]#hadoopjarhadoop-mapreduce-examples-3.1.1.jarwordcount/hello/out任務(wù)5.1認(rèn)識(shí)MapReduce【操作步驟】4.常見問題初次運(yùn)行hadoopjar命令時(shí)可能會(huì)出現(xiàn)找不到或無(wú)法加載主類的錯(cuò)誤。需要到y(tǒng)arn-site.xml配置文件修改配置，添加以下內(nèi)容。<property><name>yarn.application.classpath</name><value>/opt/hadoop/etc/hadoop:/opt/hadoop/share/hadoop/common/lib/*:/opt/hadoop/share/hadoop/common/*:/opt/hadoop/share/hadoop/hdfs:/opt/hadoop/share/hadoop/hdfs/lib/*:/opt/hadoop/share/hadoop/hdfs/*:/opt/hadoop/share/hadoop/mapreduce/lib/*:/opt/hadoop/share/hadoop/mapreduce/*:/opt/hadoop/share/hadoop/yarn:/opt/hadoop/share/hadoop/yarn/lib/*:/opt/hadoop/share/hadoop/yarn/*</value></property>任務(wù)總結(jié)理解MapReduce的原理與體系架構(gòu)。掌握YARN運(yùn)行機(jī)制。北京匯智科教《大數(shù)據(jù)平臺(tái)部署與運(yùn)維》單元5MapReduce實(shí)現(xiàn)電商銷售數(shù)據(jù)統(tǒng)計(jì)任務(wù)5.2MapReduce實(shí)現(xiàn)詞頻統(tǒng)計(jì)01理解MapReduce數(shù)據(jù)處理流程02掌握MapReduce相關(guān)Java

API學(xué)習(xí)目標(biāo)03掌握MapReduce驅(qū)動(dòng)類的編寫任務(wù)5.2MapReduce實(shí)現(xiàn)詞頻統(tǒng)計(jì)【任務(wù)場(chǎng)景】經(jīng)理：小張，我們后續(xù)也要使用Java編寫MapReduce程序，你對(duì)MapReduce的JavaAPI了解嗎？小張：Hadoop框架底層就是用Java實(shí)現(xiàn)的，所以Hadoop提供了許多用來(lái)調(diào)用MapReduce的JavaAPI，我們可以借助MapReduce的JavaAPI可以完成很多項(xiàng)自定義操作。經(jīng)理：是的，使用MapReduce的JavaAPI時(shí)也要注意數(shù)據(jù)類型轉(zhuǎn)換，要使用Hadoop特有的Writable數(shù)據(jù)類型，你先熟悉一下MapReduce的JavaAPI，然后自己實(shí)現(xiàn)一個(gè)詞頻統(tǒng)計(jì)程序，在服務(wù)器上運(yùn)行一下試試吧。小張：好的。任務(wù)5.2MapReduce實(shí)現(xiàn)詞頻統(tǒng)計(jì)【任務(wù)布置】MapReduce的運(yùn)行依賴與JDK和Hadoop，因此必須將Hadoop的基礎(chǔ)環(huán)境提前安裝好，才能進(jìn)行MapReduce的運(yùn)行和操作。本任務(wù)要求在前面已經(jīng)完成安裝部署Hadoop平臺(tái)的node1節(jié)點(diǎn)上完成，要求理解MapReduce數(shù)據(jù)處理流程，掌握MapReduce相關(guān)JavaAPI及驅(qū)動(dòng)類的創(chuàng)建；基于IDEA進(jìn)行開發(fā)，最終在node1上運(yùn)行打包后的程序，實(shí)現(xiàn)詞頻統(tǒng)計(jì)。任務(wù)5.2MapReduce實(shí)現(xiàn)詞頻統(tǒng)計(jì)5.2.1MapReduce數(shù)據(jù)處理流程MapReduce處理數(shù)據(jù)過(guò)程主要分成Map和Reduce兩個(gè)階段。數(shù)據(jù)處理流程如下圖所示：任務(wù)5.2MapReduce實(shí)現(xiàn)詞頻統(tǒng)計(jì)Map任務(wù)處理（共五步）（1）讀取輸入文件內(nèi)容，解析成key、value對(duì)。對(duì)輸入文件的每一行，解析成key、value對(duì)。每一個(gè)鍵值對(duì)調(diào)用一次map函數(shù)。（2）寫自己的邏輯，對(duì)輸入的key、value處理，轉(zhuǎn)換成新的key、value輸出。（3）對(duì)輸出的key、value進(jìn)行分區(qū)。（4）對(duì)不同分區(qū)的數(shù)據(jù)，按照key進(jìn)行排序、分組。相同key的value放到一個(gè)集合中。（5）(可選)分組后的數(shù)據(jù)進(jìn)行歸約。任務(wù)5.2MapReduce實(shí)現(xiàn)詞頻統(tǒng)計(jì)Reduce任務(wù)處理（共三步）（1）對(duì)多個(gè)map任務(wù)的輸出，按照不同的分區(qū)，通過(guò)網(wǎng)絡(luò)copy到不同的reduce節(jié)點(diǎn)。（2）對(duì)多個(gè)map任務(wù)的輸出進(jìn)行合并、排序。寫reduce函數(shù)自己的邏輯，對(duì)輸入的key、value處理，轉(zhuǎn)換成新的key、value輸出。（3）把reduce的輸出保存到文件中。任務(wù)5.2MapReduce實(shí)現(xiàn)詞頻統(tǒng)計(jì)5.2.2MapReduce相關(guān)Java

API及應(yīng)用1.

MapReduce中數(shù)據(jù)類型介紹MapReduce中所有的數(shù)據(jù)類型都要實(shí)現(xiàn)Writable接口，以便于這些類型定義的數(shù)據(jù)可以被序列化進(jìn)行網(wǎng)絡(luò)傳輸和文件存儲(chǔ)。Hadoop數(shù)據(jù)類型都放在org.apache.hadoop.io包下。Hadoop數(shù)據(jù)類型和Java數(shù)據(jù)類型對(duì)照表數(shù)據(jù)類型Hadoop數(shù)據(jù)類型Java數(shù)據(jù)類型布爾型BooleanWritableboolean整型ByteWritablebyteShortWritableshortIntWritableintLongWritablelong浮點(diǎn)型FloatWritablefloatDoubleWritabledouble字符串（文本）Textstring數(shù)組ArrayWritableArraymap集合MapWritablemap任務(wù)5.2MapReduce實(shí)現(xiàn)詞頻統(tǒng)計(jì)2.

MapReduce

Mapper類在MapReduce中，Mapper類的作用是將輸入的鍵值對(duì)映射到一組中間鍵值對(duì)。它將輸入記錄轉(zhuǎn)換為中間記錄。這些中間記錄與給定的輸出鍵相關(guān)聯(lián)，并傳遞給Reducer以獲得最終輸出。Mapper類的常用方法如下表所示：方法描述voidcleanup(Contextcontext)此方法僅在任務(wù)結(jié)束時(shí)調(diào)用一次。voidmap(KEYINkey,VALUEINvalue,Contextcontext) 對(duì)于輸入Mapper中的每個(gè)鍵值，只能調(diào)用此方法一次。voidrun(Contextcontext)可以重寫此方法以控制Mapper的執(zhí)行。voidsetup(Contextcontext)此方法僅在任務(wù)開始時(shí)調(diào)用一次。任務(wù)5.2MapReduce實(shí)現(xiàn)詞頻統(tǒng)計(jì)3.

MapReduce

Reducer類在MapReduce中，Reducer類的作用是減少中間值的集合。它的實(shí)現(xiàn)可以通過(guò)JobContext.getConfiguration()方法訪問作業(yè)的配置。Reducer類的常用方法如下表所示：方法描述voidcleanup(Contextcontext)此方法僅在任務(wù)結(jié)束時(shí)調(diào)用一次。voidreduce(KEYINkey,Iterable<VALUEIN>values,Contextcontext) 此方法只為每個(gè)鍵調(diào)用一次。voidrun(Contextcontext)這個(gè)方法可以用來(lái)控制Reducer的任務(wù)。voidsetup(Contextcontext)此方法僅在任務(wù)開始時(shí)調(diào)用一次。任務(wù)5.2MapReduce實(shí)現(xiàn)詞頻統(tǒng)計(jì)4.

MapReduce

Job類Job類用于配置作業(yè)和提交作業(yè)，它還控制執(zhí)行和查詢狀態(tài)。Job類的常用方法如下表所示：方法描述voidsetJarByClass(Class<>class)核心接口，指定執(zhí)行類所在的jar包本地位置。java通過(guò)class文件找到執(zhí)行jar包，該jar包被上傳到HDFS。voidsetMapOutputKeyClass(Class<>class)該方法用于設(shè)置map輸出數(shù)據(jù)的key類型。voidsetMapOutputValueClass(Class<>class)該方法用于設(shè)置map輸出數(shù)據(jù)的value類型。setOutputKeyClass(Class<>theClass)核心接口，指定MapReduce作業(yè)的輸出key的類型setOutputValueClass(Class<>theClass)核心接口，指定MapReduce作業(yè)的輸出value的類型voidsetMapperClass(Class<extendsMapper>class)核心接口，指定MapReduce作業(yè)的Mapper類，默認(rèn)為空。voidsetNumReduceTasks(inttasks)該方法用于設(shè)置job的reduce任務(wù)數(shù)voidsetReducerClass(Class<extendsReducer>class)核心接口，指定MapReduce作業(yè)的Reducer類，默認(rèn)為空。setPartitionerClass(Class<extendsPartitioner>class)指定MapReduce作業(yè)的Partitioner類。該方法用來(lái)分配map的輸出結(jié)果到哪個(gè)reduce類，默認(rèn)使用HashPartitioner，均勻分配map的每條鍵值對(duì)記錄。任務(wù)5.2MapReduce實(shí)現(xiàn)詞頻統(tǒng)計(jì)5.2.3MapReduce驅(qū)動(dòng)類驅(qū)動(dòng)類主要用于關(guān)聯(lián)Mapper和Reducer以及提交整個(gè)程序。驅(qū)動(dòng)類的開發(fā)有著固定的格式。編寫驅(qū)動(dòng)類總分七大步驟：1.獲取job對(duì)象。新建一個(gè)配置對(duì)象實(shí)例，用于獲取job實(shí)例方法傳遞參數(shù)。2.設(shè)置Jar的路徑(就是當(dāng)前驅(qū)動(dòng)類)。設(shè)置當(dāng)前驅(qū)動(dòng)類路徑，參數(shù)即為當(dāng)前驅(qū)動(dòng)類的類名。Configurationconfiguration=newConfiguration();Jobjob=Job.getInstance(configuration);job.setJarByClass(WordCountDriver.class);任務(wù)5.2MapReduce實(shí)現(xiàn)詞頻統(tǒng)計(jì)3.關(guān)聯(lián)mapper和reducer。設(shè)置mapper和reducer類，參數(shù)為自定義的mapper類類名及reducer類的類名。4.設(shè)置mapper輸出的key和value類型。設(shè)置mapper輸出的key和value類型時(shí)可根據(jù)業(yè)務(wù)靈活調(diào)整，類型需使用已實(shí)現(xiàn)Writable的數(shù)據(jù)類型。5.設(shè)置最終輸出的key和value設(shè)置最終輸出的key和value類型時(shí)可根據(jù)業(yè)務(wù)靈活調(diào)整，需使用已實(shí)現(xiàn)Writable的Hadoop數(shù)據(jù)類型。job.setMapperClass(WordcountMapper.class);job.setReducerClass(WordcountReducer.class);job.setMapOutputKeyClass(Text.class);job.setMapOutputValueClass(IntWritable.class);job.setOutputKeyClass(Text.class);job.setOutputValueClass(IntWritable.class);

任務(wù)5.2MapReduce實(shí)現(xiàn)詞頻統(tǒng)計(jì)6.設(shè)置輸入輸出路徑(要讀取的數(shù)據(jù)輸入路徑和輸出數(shù)據(jù)的路徑)設(shè)置路徑有兩種方式，一種是通過(guò)傳入?yún)?shù)動(dòng)態(tài)讀取，另一種是在程序中寫成固定路徑，但是會(huì)降低擴(kuò)展性，推薦使用第一種方式。第一種方式在運(yùn)行時(shí)必須傳入輸入輸出路徑，示例：hadoopjarWordCountApp/hello/out需要注意這里的路徑都是HDFS中的路徑，也就是讀取和輸出的文件都是在HDFS中的。7.提交job將作業(yè)提交到群集并等待它完成，參數(shù)設(shè)置為true代表打印顯示對(duì)應(yīng)的進(jìn)度，根據(jù)作業(yè)結(jié)果，終止當(dāng)前運(yùn)行的Java虛擬機(jī)，退出程序。FileInputFormat.setInputPaths(job,newPath(args[1]));FileOutputFormat.setOutputPath(job,newPath(args[2]));//或FileInputFormat.setInputPaths(job,newPath("/map/input"));FileOutputFormat.setOutputPath(job,newPath("/map/output"));

Booleanresult=job.waitForCompletion(true);System.exit(result?0:-1);

任務(wù)5.2MapReduce實(shí)現(xiàn)詞頻統(tǒng)計(jì)【工作流程】創(chuàng)建新的項(xiàng)目，編寫Mapper類，繼承父類并重寫map方法，編寫Reducer類，繼承父類并重寫reduce方法，編寫驅(qū)動(dòng)類，程序?qū)С鰹閖ar包運(yùn)行，查看結(jié)果。任務(wù)5.2MapReduce實(shí)現(xiàn)詞頻統(tǒng)計(jì)【操作步驟】創(chuàng)建新的項(xiàng)目，項(xiàng)目類型選擇Maven，項(xiàng)目名稱為MRWordCount，pom依賴為以下內(nèi)容。<dependencies><dependency><groupId>junit</groupId><artifactId>junit</artifactId><version>RELEASE</version></dependency><dependency><groupId>org.apache.hadoop</groupId><artifactId>hadoop-hdfs</artifactId><version>3.1.1</version></dependency><dependency><groupId>org.apache.hadoop</groupId><artifactId>hadoop-client</artifactId><version>3.1.1</version></dependency><dependency><groupId>org.apache.hadoop</groupId><artifactId>hadoop-common</artifactId><version>3.1.1</version></dependency><dependency><groupId>org.apache.logging.log4j</groupId><artifactId>log4j-core</artifactId><version>2.8.2</version></dependency></dependencies>

任務(wù)5.2MapReduce實(shí)現(xiàn)詞頻統(tǒng)計(jì)1.編寫Mapper類創(chuàng)建Java類MyMapper，繼承父類Mapper，重寫父類map方法，實(shí)現(xiàn)如下代碼：importorg.apache.hadoop.io.LongWritable;importorg.apache.hadoop.io.Text;importorg.apache.hadoop.mapreduce.Mapper;publicclassMyMapperextendsMapper<LongWritable,Text,Text,LongWritable>{ protectedvoidmap(LongWritablek1,Textv1,Contextcontext) throwsjava.io.IOException,InterruptedException{//對(duì)v1用空格進(jìn)行分割，返回單詞數(shù)組 Stringwords[]=v1.toString().split(""); Textk2=newText(); LongWritablev2=newLongWritable(1); for(Stringword:words) { k2.set(word); context.write(k2,v2);//將k2/v2寫入磁盤 } };}

任務(wù)5.2MapReduce實(shí)現(xiàn)詞頻統(tǒng)計(jì)2.編寫Reducer類創(chuàng)建Java類MyReducer，繼承父類Reducer，重寫父類reduce方法，實(shí)現(xiàn)如下代碼：importorg.apache.hadoop.io.LongWritable;importorg.apache.hadoop.io.Text;importorg.apache.hadoop.mapreduce.Reducer;

publicclassMyReducerextendsReducer<Text,LongWritable,Text,LongWritable>{ protectedvoidreduce(Textk2,Iterable<LongWritable>vs2,Contextcontext) throwsjava.io.IOException,InterruptedException{

Textk3=k2;//k3和k2一樣，表示每個(gè)單詞 longsum=0; for(LongWritablei:vs2) { sum+=i.get(); } LongWritablev3=newLongWritable(sum);

context.write(k3,v3);//最終結(jié)果

};}

任務(wù)5.2MapReduce實(shí)現(xiàn)詞頻統(tǒng)計(jì)3.創(chuàng)建驅(qū)動(dòng)類WordCountApp創(chuàng)建驅(qū)動(dòng)類WordCountApp，并編碼如下代碼：importorg.apache.hadoop.fs.Path;importorg.apache.hadoop.io.LongWritable;importorg.apache.hadoop.io.Text;importorg.apache.hadoop.mapreduce.Job;importorg.apache.hadoop.mapreduce.lib.input.FileInputFormat;importorg.apache.hadoop.mapreduce.lib.output.FileOutputFormat;importorg.apache.hadoop.conf.Configuration;publicclassWordCountApp{ publicstaticvoidmain(String[]args)throwsIOException,ClassNotFoundException,InterruptedException{ Configurationconf=newConfiguration(); Jobjob=Job.getInstance(conf,"myjob"); job.setJarByClass(WordCountApp.class); PathfileIn=newPath(args[1]);

PathfileOut=newPath(args[2]);

job.setMapperClass(MyMapper.class);//設(shè)置使用的Mapper類

job.setMapOutputKeyClass(Text.class);//設(shè)置k2/v2類型

job.setMapOutputValueClass(LongWritable.class); FileInputFormat.setInputPaths(job,fileIn);//設(shè)置任務(wù)的輸入文件路徑

job.setReducerClass(MyReducer.class);//設(shè)置使用的Reducer類

//設(shè)置k3/v3的類型，如果k3/v3和k2/v2類型一致，此步驟可以省略

//job.setOutputKeyClass(Text.class); //job.setOutputValueClass(LongWritable.class); FileOutputFormat.setOutputPath(job,fileOut);//設(shè)置最終結(jié)果輸出路徑

Booleanresult=job.waitForCompletion(true);

System.exit(result?0:-1); }}任務(wù)5.2MapReduce實(shí)現(xiàn)詞頻統(tǒng)計(jì)4.將程序?qū)С鰹閖ar包運(yùn)行，查看結(jié)果在IDEA中打開Maven面板，選擇package進(jìn)行打包，將打包后的jar包上傳到服務(wù)器上。運(yùn)行jar包運(yùn)行完成后查看運(yùn)行結(jié)果，首先查看/out1下是否有生成_SUCCESS文件查看結(jié)果文件[root@master01~]#hadoopjarwordcount.jar

WordCountApp/hello/out1[root@master01~]#hdfs

dfs-ls/out1Found2items-rw-r--r--2rootsupergroup02021-11-1111:01/out1/_SUCCESS-rw-r--r--2rootsupergroup252021-11-1111:01/out1/part-r-00000[root@master01~]#hdfsdfs-cat/out1/part-t-00000HDFS1Hello2MapReduce1

任務(wù)總結(jié)掌握MapReduce數(shù)據(jù)處理流程掌握MapReduce相關(guān)Java

API使用方法掌握MapReduce驅(qū)動(dòng)類編寫方法北京匯智科教《大數(shù)據(jù)平臺(tái)部署與運(yùn)維》單元5MapReduce實(shí)現(xiàn)電商銷售數(shù)據(jù)統(tǒng)計(jì)任務(wù)5.3MapReduce完成電商銷售數(shù)據(jù)統(tǒng)計(jì)01掌握MapReduce自定義分區(qū)的概念與用法02掌握MapReduce自定義數(shù)據(jù)類型的概念與用法學(xué)習(xí)目標(biāo)任務(wù)5.3MapReduce完成電商銷售數(shù)據(jù)統(tǒng)計(jì)【任務(wù)場(chǎng)景】經(jīng)理：小張，接下來(lái)我們要使用MapReduce對(duì)電商銷售數(shù)據(jù)進(jìn)行統(tǒng)計(jì)，可能要用到自定義分區(qū)和自定義數(shù)據(jù)類型，你了解MapReduce中的自定義分區(qū)和自定義數(shù)據(jù)類型嗎？小張：日常業(yè)務(wù)中Hadoop提供的數(shù)據(jù)類型有時(shí)不滿足使用，我們需要根據(jù)業(yè)務(wù)創(chuàng)建合適的自定義數(shù)據(jù)類型，也可以通過(guò)自定義分區(qū)對(duì)數(shù)據(jù)進(jìn)行分區(qū)。經(jīng)理：是的，自定義數(shù)據(jù)類型根據(jù)場(chǎng)景不同需要實(shí)現(xiàn)不同的接口，自定義分區(qū)可以把數(shù)據(jù)分到不同的reducer中。我給你一份后臺(tái)導(dǎo)出的數(shù)據(jù)，你用MapReduce統(tǒng)計(jì)一下每個(gè)買家收藏商品的數(shù)量，根據(jù)收藏日期做一下自定義分區(qū)。小張：好的。任務(wù)5.3MapReduce完成電商銷售數(shù)據(jù)統(tǒng)計(jì)【任務(wù)布置】MapReduce的運(yùn)行依賴與JDK和Hadoop，因此必須將Hadoop的基礎(chǔ)環(huán)境提前安裝好，才能進(jìn)行MapReduce的運(yùn)行和操作。本任務(wù)要求在前面已經(jīng)完成安裝部署Hadoop平臺(tái)的node1節(jié)點(diǎn)上完成。要求掌握MapReduce自定義分區(qū)與自定義數(shù)據(jù)類型的創(chuàng)建；基于IDEA進(jìn)行開發(fā)，每個(gè)賣家收藏商品數(shù)量的統(tǒng)計(jì)，將收藏統(tǒng)計(jì)結(jié)果分為兩個(gè)分區(qū)，2020-04-14日之前及14日當(dāng)天的數(shù)據(jù)為一個(gè)分區(qū)，2020-04-14日之后的數(shù)據(jù)為一個(gè)分區(qū)。任務(wù)5.3MapReduce完成電商銷售數(shù)據(jù)統(tǒng)計(jì)5.3.1MapReduce完成電商銷售數(shù)據(jù)統(tǒng)計(jì)的流程數(shù)據(jù)介紹現(xiàn)有某電商網(wǎng)站用戶對(duì)商品的收藏?cái)?shù)據(jù)，記錄了用戶收藏的商品id以及收藏日期，名為buyer_favorite1。buyer_favorite1包含：買家id，商品id，收藏日期這三個(gè)字段，數(shù)據(jù)以空格分割，樣本數(shù)據(jù)及格式如下：1018110004812020-04-0416:54:312000110015972020-04-0715:07:522000110015602020-04-0715:08:272004210013682020-04-0808:20:302006710020612020-04-0816:45:332005610032892020-04-1210:50:552005610032902020-04-1211:57:352005610032922020-04-1212:05:292005410024202020-04-1415:24:12……任務(wù)5.3MapReduce完成電商銷售數(shù)據(jù)統(tǒng)計(jì)5.3.2自定義分區(qū)1.

MapReduce

Partitioner類通過(guò)前面的學(xué)習(xí)我們知道Mapper最終處理的鍵值對(duì)<key,value>，是需要送到Reducer去合并的，合并的時(shí)候，有相同key的鍵/值對(duì)會(huì)送到同一個(gè)Reducer節(jié)點(diǎn)中進(jìn)行歸并。哪個(gè)key到哪個(gè)Reducer的分配過(guò)程，是由Partitioner規(guī)定的。分區(qū)的目的是把具有相同key的值集合在一起，確保key相同的值都會(huì)在同一個(gè)reducer里面。這樣才能保證map的輸出數(shù)據(jù)被均勻的分發(fā)到reducer。HadoopMapReduce默認(rèn)的HadoopPartitioner是哈希

Partitioner（HashPartitioner），它會(huì)對(duì)key計(jì)算哈希值，并基于該哈希值對(duì)鍵值對(duì)數(shù)據(jù)進(jìn)行分區(qū)。任務(wù)5.3MapReduce完成電商銷售數(shù)據(jù)統(tǒng)計(jì)Partitioner的數(shù)量等于reducer的數(shù)量，Partitioner會(huì)根據(jù)reducer的數(shù)量來(lái)劃分?jǐn)?shù)據(jù)，reducer數(shù)量可以通過(guò)下面的方法進(jìn)行設(shè)置：JobConf.setNumReduceTasks()因此，來(lái)自同一個(gè)分區(qū)的數(shù)據(jù)會(huì)被一個(gè)reducer任務(wù)處理。需要注意的是，只有作業(yè)具有多個(gè)reducer任務(wù)時(shí)，分區(qū)才會(huì)被創(chuàng)建。也就是說(shuō)，如果作業(yè)只有1個(gè)reducer任務(wù)，分區(qū)階段是不會(huì)發(fā)生的。低效的分區(qū)意味著，某些reducer將比其他reducer任務(wù)處理更多的數(shù)據(jù)。那么，整個(gè)作業(yè)的運(yùn)行時(shí)間將取決于這些需要處理更多數(shù)據(jù)的reducer，也就是說(shuō)，作業(yè)的運(yùn)行時(shí)間會(huì)更長(zhǎng)。為了克服低效分區(qū)的問題，我們可以自定義分區(qū)器（partitioner），這樣我們就可以根據(jù)具體業(yè)務(wù)修改分區(qū)邏輯，把數(shù)據(jù)均分的分發(fā)到不同的reducer任務(wù)里。任務(wù)5.3MapReduce完成電商銷售數(shù)據(jù)統(tǒng)計(jì)2.

Partitioner實(shí)現(xiàn)過(guò)程（1）先分析一下具體的業(yè)務(wù)邏輯，確定大概有多少個(gè)分區(qū)（2）首先書寫一個(gè)類，它要繼承

org.apache.hadoop.mapreduce.Partitioner這個(gè)抽象類（3）重寫publicintgetPartition這個(gè)方法，根據(jù)具體邏輯，讀數(shù)據(jù)庫(kù)或者配置返回相同的數(shù)字（4）在main方法中設(shè)置Partioner的類，job.setPartitionerClass(DataPartitioner.class);（5）設(shè)置Reducer的數(shù)量，job.setNumReduceTasks(2);任務(wù)5.3MapReduce完成電商銷售數(shù)據(jù)統(tǒng)計(jì)3.總結(jié)分區(qū)Partitioner主要作用在于以下兩點(diǎn)（1）根據(jù)業(yè)務(wù)需要，產(chǎn)生多個(gè)輸出文件；（2）多個(gè)reduce任務(wù)并發(fā)運(yùn)行，提高整體job的運(yùn)行效率任務(wù)5.3MapReduce完成電商銷售數(shù)據(jù)統(tǒng)計(jì)5.3.3自定義數(shù)據(jù)類型Hadoop使用了自己寫的序列化格式

Writable，它格式緊湊，速度快，但是它很難用Java以外的語(yǔ)言進(jìn)行拓展或使用，因?yàn)閃ritable是Hadoop的核心，大多數(shù)MapReduce程序都會(huì)為鍵和值使用它，Hadoop中的數(shù)據(jù)類型都要實(shí)現(xiàn)Writable接口，以便用這些類型定義的數(shù)據(jù)可以被網(wǎng)絡(luò)傳輸和文件存儲(chǔ)。任務(wù)5.3MapReduce完成電商銷售數(shù)據(jù)統(tǒng)計(jì)自定義數(shù)據(jù)類型實(shí)現(xiàn)過(guò)程（1）繼承接口Writable,實(shí)現(xiàn)其方法write()和readFields(),以便該數(shù)據(jù)能被序列化后完成網(wǎng)絡(luò)傳輸或文件輸入/輸出。（2）如果該數(shù)據(jù)需要作為主鍵key使用，或需要比較數(shù)值大小時(shí)，則需要實(shí)現(xiàn)WritalbeComparable接口,實(shí)現(xiàn)其方法write(),readFields(),CompareTo()。（3）數(shù)據(jù)類型，為了方便反射，必須要有一個(gè)無(wú)參的構(gòu)造方法用來(lái)創(chuàng)建對(duì)象。（4）在自定義數(shù)據(jù)類型中，建議使用java的原生數(shù)據(jù)類型，最好不要使用Hadoop對(duì)原生類型進(jìn)行封裝的數(shù)據(jù)類型。比如intx;//IntWritable和Strings;//Text等等。任務(wù)5.3MapReduce完成電商銷售數(shù)據(jù)統(tǒng)計(jì)【工作流程】創(chuàng)建新的項(xiàng)目；編寫自定義數(shù)據(jù)類型類，繼承Writable接口并重寫write和readFields方法；編寫Mapper類，繼承父類并重寫map方法；編寫自定義分區(qū)類，繼承父類并重寫getPartition方法；編寫Reducer類，繼承父類并重寫reduce方法；編寫驅(qū)動(dòng)類，程序?qū)С鰹閖ar包運(yùn)行，查看結(jié)果。任務(wù)5.3MapReduce完成電商銷售數(shù)據(jù)統(tǒng)計(jì)【操作步驟】創(chuàng)建新的項(xiàng)目，項(xiàng)目類型選擇Maven，項(xiàng)目名稱為MRCollectionCount。1.首先編寫自定義數(shù)據(jù)類型類：CollectionWritable，實(shí)現(xiàn)Writable接口并重寫write和readFields方法。importorg.apache.hadoop.io.Writable;importjava.io.DataInput;importjava.io.DataOutput;importjava.io.IOException;publicclassCollectionWritableimplementsWritable{privateintcount;//收藏次數(shù)privateStringcollectTime;//收藏時(shí)間publicintgetCount(){returncount;}publicvoidsetCount(intcount){this.count=count;}publicStringgetCollectTime(){returncollectTime;}publicvoidsetCollectTime(StringcollectTime){this.collectTime=collectTime;}publicCollectionWritable(intcount,StringcollectTime){super();this.collectTime=collectTime;this.count=count;}

publicCollectionWritable(){

}@Overridepublicvoidwrite(DataOutputdataOutput)throwsIOException{dataOutput.writeInt(this.count);dataOutput.writeUTF(this.collectTime);}@OverridepublicvoidreadFields(DataInputdataInput)throwsIOException{this.count=dataInput.readInt();this.collectTime=dataInput.readUTF();}@OverridepublicStringtoString(){return"CollectionWritable{"+"count="+count+",collectTime='"+collectTime+'\''+'}';}}

任務(wù)5.3MapReduce完成電商銷售數(shù)據(jù)統(tǒng)計(jì)2.編寫Mapper類，繼承父類并重寫map方法importorg.apache.hadoop.io.LongWritable;importorg.apache.hadoop.io.Text;importorg.apache.hadoop.mapreduce.Mapper;

importjava.io.IOException;

publicclassMyMapperextendsMapper<LongWritable,Text,Text,CollectionWritable>{@Overrideprotectedvoidmap(LongWritablekey,Textvalue,Mapper<LongWritable,Text,Text,CollectionWritable>.Contextcontext)throwsIOException,InterruptedException{//通過(guò)空格分割finalString[]splited=value.toString().split("");//第一列為用戶IDfinalStringuserId=splited[0];finalTextk2=newText(userId);//收藏次數(shù)記為1finalintcount=1;//第三列為收藏時(shí)間finalStringcollectTime=splited[2];finalCollectionWritablev2=newCollectionWritable(count,collectTime);context.write(k2,v2);}}

任務(wù)5.3MapReduce完成電商銷售數(shù)據(jù)統(tǒng)計(jì)3.編寫自定義分區(qū)類，繼承父類并重寫getPartition方法importorg.apache.hadoop.io.Text;importorg.apache.hadoop.mapreduce.lib.partition.HashPartitioner;

publicclassMyPartitionextendsHashPartitioner<Text,CollectionWritable>{@OverridepublicintgetPartition(Textk2,CollectionWritablev2,intnumReduceTasks){Stringdate=v2.getCollectTime();//通過(guò)比對(duì)收藏時(shí)間進(jìn)行分區(qū)

if(pareTo("2020-04-14")>0){return1;}else{return0;}}}任務(wù)5.3MapReduce完成電商銷售數(shù)據(jù)統(tǒng)計(jì)4.編寫Reducer類，繼承父類并重寫reduce方法importorg.apache.hadoop.io.LongWritable;importorg.apache.hadoop.io.Text;importorg.apache.hadoop.mapreduce.Reducer;

importjava.io.IOException;

publicclassMyReducerextendsReducer<Text,CollectionWritable,Text,LongWritable>{@Overrideprotectedvoidreduce(Textk2,Iterable<CollectionWritable>values,Reducer<Text,CollectionWritable,Text,LongWritable>.Contextcontext)throwsIOException,InterruptedException{Textk3=k2;longsum=0;//收藏次數(shù)進(jìn)行累加for(CollectionWritablecollect:values){sum+=collect.getCount();}

LongWritablev3=newLongWritable(sum);

context.write(k3,v3);}}

任務(wù)5.3MapReduce完成電商銷售數(shù)據(jù)統(tǒng)計(jì)5.編寫驅(qū)動(dòng)類importorg.apache.hadoop.conf.Configuration;importorg.apache.hadoop.fs.Path;importorg.apache.hadoop.io.LongWritable;importorg.apache.hadoop.io.Text;importorg.apache.hadoop.mapreduce.Job;importorg.apache.hadoop.mapreduce.lib.input.FileInputFormat;importorg.apache.hadoop.mapreduce.lib.output.FileOutputFormat;importjava.io.IOException;publicclassCollectionApp{publicstaticvoidmain(String[]args)throwsIOException,ClassNotFoundException,InterruptedException{Configurationconf=newConfiguration();Jobjob=Job.getInstance(conf,"collectionJob");job.setJarByClass(CollectionApp.class);PathfileIn=newPath(args[0]);PathfileOut=newPath(args[1]);job.setMapperClass(MyMapper.class);job.setMapOutputKeyClass(Text.class);job.setMapOutputValueClass(CollectionWritable.class);FileInputFormat.setInputPaths(job,fileIn);//設(shè)置自定義分區(qū)類job.setPartitionerClass(MyPartition.class);//將reduce任務(wù)數(shù)量設(shè)置為2job.setNumReduceTasks(2);job.setReducerClass(MyReducer.class);job.setOutputKeyClass(Text.class);job.setOutputValueClass(LongWritable.class);FileOutputFormat.setOutputPath(job,fileOut);Booleanresult=job.waitForCompletion(true);System.exit(result?0:-1);}}任務(wù)5.3MapReduce完成電商銷售數(shù)據(jù)統(tǒng)計(jì)6.將程序?qū)С鰹閖ar包運(yùn)行，查看結(jié)果運(yùn)行完成后查看運(yùn)行結(jié)果，首先查看/co_out下是否有生成_SUCCESS文件，并且有兩個(gè)分區(qū)文件。查看結(jié)果文件[root@master01~]#hadoopjarCollectionApp.jar

CollectionApp/collection.txt/co_out

[root@master01~]#hdfsdfs-ls/co_outFound3items-rw-r--r--2rootsupergroup02021-11-1914:45/co_out/_SUCCESS-rw-r--r--2rootsupergroup732021-11-1914:45/co_out/part-r-00000-rw-r--r--2rootsupergroup02021-11-1914:45/co_out/part-r-00001

[root@master01~]#hdfsdfs-cat/co_out/part-r-00000101811200012200421200543200551200563200671200761[root@master01~]#hdfsdfs-cat/co_out/part-r-00001200543200569200641200764

任務(wù)總結(jié)理解MapReduce自定義分區(qū)和自定義數(shù)據(jù)類型的概念掌握MapReduce自定義分區(qū)和自定義數(shù)據(jù)類型的寫法能夠使用自定義分區(qū)與自定義數(shù)據(jù)類型正確完成電商數(shù)據(jù)統(tǒng)計(jì)程序北京匯智科教《大數(shù)據(jù)平臺(tái)部署與運(yùn)維》單元5MapReduce實(shí)現(xiàn)電商銷售數(shù)據(jù)統(tǒng)計(jì)任務(wù)5.4MapReduce任務(wù)監(jiān)控01掌握MapReduce任務(wù)監(jiān)控方式02理解MapReduce任務(wù)失敗常見原因?qū)W習(xí)目標(biāo)任務(wù)5.4MapReduce任務(wù)監(jiān)控【任務(wù)場(chǎng)景】經(jīng)理：小張，現(xiàn)在你已經(jīng)了解MapReduce的原理，也能通過(guò)JavaAPI實(shí)現(xiàn)一些功能了，但是你知道MapReduce執(zhí)行過(guò)程中如何監(jiān)控嗎？小張：可以在任務(wù)執(zhí)行時(shí)命令行看到一些信息，也可以通過(guò)瀏覽器和Hadoop命令去進(jìn)行監(jiān)控。經(jīng)理：嗯，不錯(cuò)，掌握這幾種監(jiān)控方式就可以對(duì)大部分的MapReduce場(chǎng)景進(jìn)行有效監(jiān)控了，你再去了解一下MapReduce任務(wù)執(zhí)行失敗常見原因。小張：好的。任務(wù)5.4MapReduce任務(wù)監(jiān)控【任務(wù)布置】掌握MapReduce任務(wù)監(jiān)控方式，理解MapReduce任務(wù)執(zhí)行失敗常見原因，再次運(yùn)行電商數(shù)據(jù)分析程序，監(jiān)控任務(wù)執(zhí)行過(guò)程。任務(wù)5.4MapReduce任務(wù)監(jiān)控5.4.1MapReduce任務(wù)監(jiān)控方式當(dāng)我們執(zhí)行一個(gè)MapReduce任務(wù)后，程序被提交給了集群。對(duì)于開發(fā)測(cè)試集群運(yùn)行程序的開發(fā)人員來(lái)說(shuō)，剛剛接觸MapReduce的初學(xué)者往往是在命令行前等著程序執(zhí)行完成，遇到任務(wù)運(yùn)行緩慢或報(bào)錯(cuò)的情況時(shí)很難做出有效響應(yīng)。下面介紹MapReduce任務(wù)監(jiān)控的幾種形式，讓大家能對(duì)MapReduce程序進(jìn)行有效監(jiān)控。任務(wù)5.4MapReduce任務(wù)監(jiān)控1.執(zhí)行時(shí)監(jiān)控執(zhí)行Hadoopjar命令后控制臺(tái)會(huì)輸出任務(wù)信息。執(zhí)行時(shí)需重點(diǎn)關(guān)注標(biāo)紅的這幾行信息。上面的輸出信息可以查看當(dāng)前任務(wù)的ID，以及map階段、reduce階段的任務(wù)進(jìn)度，當(dāng)輸出Jobjob_idcompletedsuccessfully時(shí)表明任務(wù)已成功執(zhí)行完畢。[root@master01~]#hadoopjar/opt/hadoop/share/hadoop/mapreduce/hadoop-mapreduce-examples-3.1.1.jarwordcount/collection.txt/out2021-11-3003:55:46,111INFOclient.RMProxy:ConnectingtoResourceManageratmaster01/14:80322021-11-3003:55:46,374INFOmapreduce.JobResourceUploader:DisablingErasureCodingforpath:/tmp/hadoop-yarn/staging/root/.staging/job_1638262379053_00012021-11-3003:55:47,138INFOinput.FileInputFormat:Totalinputfilestoprocess:1……2021-11-3003:55:53,710INFOmapreduce.Job:map0%reduce0%2021-11-3003:55:58,774INFOmapreduce.Job:map100%reduce0%2021-11-3003:56:02,799INFOmapreduce.Job:map100%reduce100%2021-11-3003:56:02,807INFOmapreduce.Job:Jobjob_1638262379053_0001completedsuccessfully2021-11-3003:56:02,863INFOmapreduce.Job:Counters:53……任務(wù)5.4MapReduce任務(wù)監(jiān)控2.瀏覽器監(jiān)控執(zhí)行Hadoop任務(wù)后在瀏覽器打開管理節(jié)點(diǎn)MapReduceWebUI界面