大數(shù)據(jù)技術(shù)及應(yīng)用-基于Python語(yǔ)言 課件 第7章 分布式計(jì)算框架MapReduce

第7章分布式計(jì)算框架MapReduce本章學(xué)習(xí)目標(biāo)掌握MapReduce的基本工作原理；掌握MapReduce各步驟輸入/輸出的銜接關(guān)系；掌握利用HadoopStreaming進(jìn)行代碼測(cè)試的基本思路與方法；掌握利用Python設(shè)計(jì)MapReduce程序的思路和方法；掌握MapReduce程序設(shè)計(jì)模式的基本概念，以及理解常見(jiàn)的幾種設(shè)計(jì)模式；理解MRJob的工作原理，并能夠進(jìn)行簡(jiǎn)單的MRJob程序設(shè)計(jì)應(yīng)用。01MapReduce概述02WordCount實(shí)例詳解目錄Contents03Python代碼測(cè)試04利用Python的迭代器和生成器優(yōu)化wordCount程序05MapReduce程序設(shè)計(jì)模式06用MRJob庫(kù)編寫MapReduce程序07本章小結(jié)MapReduce概述01MapReduce概述HadoopMapReduce是一個(gè)計(jì)算框架，目的是為了方便編寫具有高可靠性、高容錯(cuò)性的，能在大型集群（數(shù)千個(gè)節(jié)點(diǎn)）上并行地處理大量數(shù)據(jù)（數(shù)TB的數(shù)據(jù)集）的分布式應(yīng)用程序。該計(jì)算框架管理數(shù)據(jù)傳遞的所有細(xì)節(jié)，如發(fā)起任務(wù)、驗(yàn)證任務(wù)完成情況以及在節(jié)點(diǎn)之間復(fù)制數(shù)據(jù)等。第一個(gè)MapReduce的Python程序Hadoop自帶有計(jì)算圓周率PI的示例jar包，程序用隨機(jī)數(shù)的方式計(jì)算PI值，該程序是一個(gè)java程序，通過(guò)執(zhí)行以下命令調(diào)用集群進(jìn)行計(jì)算：hadoopjar$HADOOP_HOME/share/hadoop/mapreduce/hadoop-mapreduce-examples-3.3.0.jarpi1010輸出結(jié)果如下：第一個(gè)MapReduce的Python程序第一個(gè)Python程序也是實(shí)現(xiàn)PI的計(jì)算。第一個(gè)MapReduce的Python程序第一個(gè)MapReduce的Python程序在HDFS上創(chuàng)建一個(gè)空的文件，之后以該空文件作為輸入文件便可以運(yùn)行計(jì)算PI的程序了。運(yùn)行完成后，可以在HDFS文件系統(tǒng)中查看輸出文件內(nèi)容。假設(shè)把程序?qū)?yīng)的mapper和reducer存放在“/codes/Pi”文件夾中，HDFS系統(tǒng)上用作輸入的數(shù)據(jù)存放于HDFS的“/data”文件夾，輸出存放于“/data_rlt”。其操作步驟具體如下：第一個(gè)MapReduce的Python程序chmod+x/codes/Pi/*.*第一個(gè)MapReduce的Python程序利用如下命令運(yùn)行該程序運(yùn)行結(jié)果如下：mapredstreaming-input/data/tmp.txt-output/data_rlt-mapper/codes/pi/pi_mapper.py-reducer/codes/pi/pi_reducer.py第一個(gè)MapReduce的Python程序MapReduce工作原理MapReduce程序分三個(gè)階段執(zhí)行，即map階段、shuffle階段和reduce階段其中shuffle可以由MapReduce框架自動(dòng)完成。因此，多數(shù)情況下，MapReduce算法主要包括兩個(gè)任務(wù)：Map和Reduce。MapReduce工作原理1．Map階段Map的任務(wù)是對(duì)輸入數(shù)據(jù)進(jìn)行處理，通常情況下，MapReduce作業(yè)的輸入和輸出都存儲(chǔ)在文件系統(tǒng)中，MapReduce作業(yè)將輸入數(shù)據(jù)集拆分為獨(dú)立的塊，這些塊由Map任務(wù)以完全并行的方式處理，并轉(zhuǎn)換為一種（key,value）的數(shù)據(jù)模式。MapReduce工作原理2．Reduce階段該階段包括shuffle和reduce，Reducer的任務(wù)是把Mapper生成的（key,value）集整合處理，并輸出最終結(jié)果到HDFS中。系統(tǒng)對(duì)map任務(wù)的輸出執(zhí)行排序和轉(zhuǎn)換，并映射為reduce任務(wù)的輸入，此過(guò)程就是Shuffle。兩階段任務(wù)完成后，集群收集并縮減數(shù)據(jù)以形成最后結(jié)果，并將其發(fā)送回Hadoop服務(wù)器。MapReduce工作原理CombinerMapReduce程序還可以包含combiner，combiner在Map之后執(zhí)行，通常情況下實(shí)現(xiàn)對(duì)Map輸出數(shù)據(jù)的簡(jiǎn)單歸納，從而減少數(shù)據(jù)傳輸。在多數(shù)應(yīng)用中，combiner都是能夠大幅提高程序效率的關(guān)鍵步驟。MapReduce工作原理Hadoop框架是用Java實(shí)現(xiàn)的，MapReduce應(yīng)用程序卻可以不用Java編寫。Hadoop流（HadoopStreaming）是一個(gè)工具應(yīng)用程序（Utility），它允許用戶使用任何可執(zhí)行文件作為mapper或reducer來(lái)創(chuàng)建和運(yùn)行作業(yè)；此外，Hadoop管道（HadoopPipes）是一種SWIG兼容的C++API，他可以支持編寫非JNI（JavaNativeInterface）的MapReduce應(yīng)用程序。MapReduce作業(yè)的工作流程MapReduce作業(yè)的工作流程用戶程序中的MapReduce庫(kù)首先將輸入文件劃分為M片，每片大小一般在16M到64M之間（可由參數(shù)指定），然后，它在集群多臺(tái)機(jī)器上啟動(dòng)相同的程序復(fù)制。其中一個(gè)復(fù)制程序?yàn)閙aster(主模塊)，其余的都是worker(分模塊)。worker接收master分配的任務(wù)，其中有M個(gè)Map任務(wù)和R個(gè)Reduce任務(wù)要分配。master挑選一個(gè)空閑的worker并為其分配一個(gè)map任務(wù)或者reduce任務(wù)。MapReduce作業(yè)的工作流程被分配到Map任務(wù)的worker會(huì)去讀取相應(yīng)的輸入塊的內(nèi)容。它從輸入文件中解析出鍵值對(duì)并且將每個(gè)鍵值對(duì)傳送給用戶定義的Map函數(shù)。由Map函數(shù)產(chǎn)生的中間鍵值對(duì)則緩存在內(nèi)存中。被緩存的鍵值對(duì)會(huì)階段性地寫回本地磁盤，并且被劃分函數(shù)分割成R份。這些緩存在磁盤上的數(shù)據(jù)位置會(huì)被回傳給master(主模塊)，master再負(fù)責(zé)將這些位置轉(zhuǎn)發(fā)給Reduceworker。MapReduce作業(yè)的工作流程當(dāng)Reduceworker(Reduce分模塊)從master(主模塊)那里接收到這些位置信息時(shí)，它會(huì)使用遠(yuǎn)程過(guò)程調(diào)用從Mapworker的本地磁盤中獲取緩存的數(shù)據(jù)。當(dāng)Reduceworker讀入全部的中間數(shù)據(jù)之后，將根據(jù)中間鍵對(duì)進(jìn)行排序，這樣所有具有相同鍵的鍵值對(duì)就都聚集在一起。因?yàn)橐丫哂邢嗤I的鍵值對(duì)映射到同一個(gè)reducetask中，所以需要對(duì)中間鍵值對(duì)進(jìn)行排序。MapReduce作業(yè)的工作流程Reduceworker遍歷已經(jīng)排好序的中間數(shù)據(jù)，每當(dāng)遇到一個(gè)新的中間鍵，它會(huì)將key和相應(yīng)的中間值傳遞給Reduce函數(shù)。Reduce函數(shù)的輸出會(huì)被添加到這個(gè)Reduce部分的輸出文件中。當(dāng)所有的Maptasks和Reducetasks都完成時(shí)，master將喚醒用戶程序。至此，用戶代碼中的MapReduce調(diào)用返回。MapReduce作業(yè)的工作流程執(zhí)行完后，MapReduce的執(zhí)行結(jié)果被存放在R個(gè)輸出文件中（每個(gè)Reducetask對(duì)應(yīng)一個(gè)，文件名由用戶指定）。因?yàn)檫@些輸出文件經(jīng)常作為另一個(gè)MapReduce調(diào)用的輸入，或者將它們用于另外一個(gè)能夠以多個(gè)文件作為輸入的分布式應(yīng)用，所以，用戶并不需要將R個(gè)輸出文件歸并成一個(gè)。MapReduce作業(yè)的工作流程MapReduce的輸入/輸出MapReduce框架只對(duì)<key，value>（鍵值對(duì)）進(jìn)行操作，該框架將作業(yè)的輸入視為一組<key，value>，并生成<key，value>作為作業(yè)的輸出。與Java程序不同，Python的Map程序輸入/輸出流為標(biāo)準(zhǔn)輸入/輸出，STDIN和STDOUT，mapper輸入從STDIN中接收數(shù)據(jù)，reducer輸出結(jié)果到STDOUT；在mapper和reducer之間的數(shù)據(jù)也是通過(guò)標(biāo)準(zhǔn)輸入/輸出進(jìn)行關(guān)聯(lián)。至于他們之間的數(shù)據(jù)流關(guān)系則交由hadoopStreaming處理。WordCount實(shí)例詳解02WordCount程序源碼使用Python編寫MapReduce的要點(diǎn)在于利用Hadoop流的API，通過(guò)STDIN(標(biāo)準(zhǔn)輸入)、STDOUT(標(biāo)準(zhǔn)輸出)在Map函數(shù)和Reduce函數(shù)之間傳遞數(shù)據(jù)。我們需要做的是利用Python的sys.stdin讀取輸入數(shù)據(jù)，并把輸出傳送給sys.stdout。Hadoop流將會(huì)幫助我們處理中間的過(guò)程。WordCount程序一共包含兩個(gè)部分，分布式map階段程序wordCount_mapper.py和reduce階段程序wordCount_reducer.py，分別如下所示。WordCount程序源碼WordCount程序源碼WordCount程序源碼WordCount程序源碼編寫完代碼，并把他們分別存儲(chǔ)在文件wordCount_mapper.py和wordCount_reducer.py。檢查上述文件是否具有可執(zhí)行權(quán)限，若沒(méi)有，可以通過(guò)chmod命令設(shè)置兩個(gè)文件的執(zhí)行權(quán)限，如“chmod+x./wordCounter/mapper.py”。WordCount程序執(zhí)行利用HadoopStreaming執(zhí)行一個(gè)MapReduce任務(wù)的調(diào)用方法如下所示。hadoop-streaming.jar通常命名為hadoop-streaming-x.x.x.jar，其中x.x.x為具體版本號(hào)（例如對(duì)應(yīng)hadoop3.3.0版的文件名為hadoop-streaming-3.3.0.jar）。WordCount程序執(zhí)行（1）數(shù)據(jù)準(zhǔn)備（路徑可能會(huì)因個(gè)人不同）后在HDFS上創(chuàng)建“/wordCountData”文件夾，并把數(shù)據(jù)文件（文本文件）上傳到該文件夾；WordCount程序執(zhí)行（2）執(zhí)行命令如下（當(dāng)命令在一行中寫不下時(shí)，可以使用“\”換行）mapredstreaming\-input/wordCountData\-output/wordCount_rlt\-mapper"/codes/wordCount/wordCount_mapper.py"\-reducer"/codes/wordCount/wordCount_reducer.py"WordCount程序執(zhí)行（3）運(yùn)行完成后，查看文件夾“/wordCount_rlt”中的文件，可以通過(guò)cat直接查看WordCount程序原理將文件拆分成多片段（splits），由于測(cè)試用的文件較小，所以每個(gè)文件為一個(gè)split，并將文件形成標(biāo)準(zhǔn)輸入流，這一步由MapReduce框架自動(dòng)完成；在mapper中按行讀取標(biāo)準(zhǔn)輸入數(shù)據(jù)，并對(duì)單詞進(jìn)行分割形成<key,value>，其中key為單詞，value為該單詞出現(xiàn)次數(shù)，這里沒(méi)有對(duì)單詞進(jìn)行統(tǒng)計(jì)，所以value都是1；得到map方法輸出的<key,value>集后，shuffle將它們按照key值進(jìn)行排序，并作為reduce的輸入數(shù)據(jù)；得到排過(guò)序的鍵值對(duì)后，利用自定義的reduce方法進(jìn)行處理，得到新的<key,value>對(duì)，并作為WordCount的輸出結(jié)果。WordCount程序原理MapReducePython代碼的測(cè)試03MapReducePython代碼的測(cè)試使用Python編寫完成的MapReduce程序，通常需要進(jìn)行調(diào)試、測(cè)試，以排除隱藏的邏輯問(wèn)題。對(duì)運(yùn)行于分布式環(huán)境的MapReduce程序而言，其運(yùn)行的整個(gè)過(guò)程由mapReduce框架自動(dòng)管理，在這種情況下要通過(guò)一遍遍運(yùn)行查找程序，再在輸出的日志中發(fā)現(xiàn)邏輯錯(cuò)誤、甚至語(yǔ)法問(wèn)題將是非常耗時(shí)且煩瑣的。因此，我們需要通過(guò)使用一些工具和方法來(lái)協(xié)助程序代碼的調(diào)試。MapReducePython代碼的測(cè)試Python程序的運(yùn)行是通過(guò)HadoopStreaming工具進(jìn)行數(shù)據(jù)流的控制，并且Python的Map程序是通過(guò)標(biāo)準(zhǔn)輸入STDIN接收輸入數(shù)據(jù)的，所以可以通過(guò)構(gòu)造標(biāo)準(zhǔn)輸出作為map程序的輸入，然后利用linux的管道把map輸出到sort工具實(shí)現(xiàn)排序；排序完成后的數(shù)據(jù)再通過(guò)管道輸出到reduce程序，從而實(shí)現(xiàn)簡(jiǎn)單測(cè)試環(huán)境的構(gòu)建。而reduce的最終輸出結(jié)果，則可以通過(guò)重定向標(biāo)準(zhǔn)輸出到文件的方式實(shí)現(xiàn)結(jié)果的輸出和存儲(chǔ)。MapReducePython代碼的測(cè)試MapReducePython代碼的測(cè)試以前面的wordCount程序?yàn)槔?，可以先把輸入文件之一?fù)制到與mapper和reducer同一文件夾中，此時(shí)，該文件夾包含wordCount_mapper.py、wordCount_reducer.py兩個(gè)程序文件，運(yùn)行如下命令：MapReducePython代碼的測(cè)試還可以把上述節(jié)點(diǎn)分開(kāi)。例如，可以首先把mapper的輸出重定向到tmp.txt文件中，以便查看mapper階段的運(yùn)行結(jié)果。然后再把中間結(jié)果作為reducer的輸入便可以分步查看兩個(gè)階段的運(yùn)行結(jié)果。運(yùn)行命令如下：catinput.txt|./wordCount_mapper.py>tmp.txtcattmp.txt|sort-k1,1|./wordCount_reducer.py>rlt.txt利用迭代器和生成器優(yōu)化wordCount程序04Python中的迭代器迭代器是訪問(wèn)集合元素的一種方式，它提供了一個(gè)統(tǒng)一的訪問(wèn)集合的接口。迭代器對(duì)象從集合的第一個(gè)元素開(kāi)始訪問(wèn)，直到訪問(wèn)完所有的元素結(jié)束。迭代器只提供了next方法，所以只能往前不會(huì)后退。Python中的迭代器一個(gè)簡(jiǎn)單的迭代器及使用Python中的迭代器把一個(gè)類作為一個(gè)迭代器使用需要在類中實(shí)現(xiàn)兩個(gè)方法__iter__()與__next__()Python中的生成器帶有yield的函數(shù)在Python中被稱為generator（生成器），生成器是迭代器的一種。例如：Python中的生成器簡(jiǎn)單而言，yield的作用就是把一個(gè)函數(shù)變成一個(gè)generator生成器常被用于數(shù)據(jù)文件讀取，生成器可以根據(jù)需要讀取數(shù)據(jù)。如果直接對(duì)文件對(duì)象調(diào)用read()方法，會(huì)導(dǎo)致不可預(yù)測(cè)的內(nèi)存占用。使用yield利用固定長(zhǎng)度的緩沖區(qū)來(lái)不斷讀取文件內(nèi)容，我們無(wú)須編寫讀文件的迭代類，就可以輕松實(shí)現(xiàn)文件讀取。Python中的生成器讀取文件的生成器，每次讀取BLOCK_SIZE大小itertools模塊itertools是Python內(nèi)置的模塊，itertools包提供了更加靈活的生成循環(huán)器的工具，其使用簡(jiǎn)單且功能強(qiáng)大。該模塊中的部分內(nèi)容，可以參考教材7.6.3.其中，gropby()迭代器能夠把可迭代的輸入數(shù)據(jù)按關(guān)鍵字進(jìn)行分組后輸出，例如：itertools模塊fromitertoolsimportgroupbyfromoperatorimportitemgettermapout=[('Hello',1),('hadoop',1),('Bye',1),('hadoop',1),('Hello',1),('world',1),('Bye',1),('world',1)]mapout=sorted(mapout,key=itemgetter(0))#排序rlt_groupby=groupby(mapout,key=itemgetter(0))forkey,itemsinrlt_groupby:print("key:",key)forsubiteminitems:print("subitem:",subitem)優(yōu)化wordCount程序利用上述Python的生成器以及itertools模塊工具，我們對(duì)wordCount程序進(jìn)行修改，形成更具有Python風(fēng)格的wordCount。優(yōu)化wordCount程序優(yōu)化wordCount程序#!/usr/bin/envpython#-*-coding:UTF-8-*-fromoperatorimportitemgetterfromitertoolsimportgroupbyimportsysdefread_mapper_output(file,separator='\t'):forlineinfile:yieldline.rstrip().split(separator,1)defmain(separator='\t'):data=read_mapper_output(sys.stdin,separator=separator)#data=sorted(data,key=itemgetter(0))forcurrent_word,groupingroupby(data,itemgetter(0)):try:total_count=sum(int(count)forcurrent_word,countingroup)print("%s%s%d"%(current_word,separator,total_count))exceptvalueError:passif__name__=="__main__":main()例7.6.4測(cè)試結(jié)果MapReduce程序設(shè)計(jì)模式05MR程序設(shè)計(jì)模式MapReduce程序設(shè)計(jì)模式的目的是通過(guò)總結(jié)MapReduce任務(wù)的模式，形成具有一定可替代性的程序設(shè)計(jì)模式，以便幫助程序員提高學(xué)習(xí)、實(shí)現(xiàn)MapReduce框架上的程序開(kāi)發(fā)的效率。數(shù)據(jù)集介紹為了能夠更好的示范程序的數(shù)據(jù)處理與應(yīng)用，我們采用空氣質(zhì)量數(shù)據(jù)作為本章處理的目標(biāo)數(shù)據(jù)，全國(guó)空氣質(zhì)量數(shù)據(jù)來(lái)自中國(guó)環(huán)境監(jiān)測(cè)總站的全國(guó)城市空氣質(zhì)量實(shí)時(shí)發(fā)布平臺(tái)。數(shù)據(jù)格式：城市空氣指標(biāo)值、站點(diǎn)空氣指標(biāo)、站點(diǎn)詳細(xì)信息20140513,0,AQI,北京,8120140513,0,AQI,天津,6920140513,0,AQI,石家莊,12520140513,0,PM.5,唐山,82(1)聚合查詢模式聚合查詢是指對(duì)所要處理的數(shù)據(jù)進(jìn)行一定的統(tǒng)計(jì)聚合處理，以從概要、全局的層面上掌握數(shù)據(jù)的特征和狀況。例如，今年北京地區(qū)的空氣質(zhì)量總體情況，包括空氣質(zhì)量各個(gè)指標(biāo)的平均值、最大最小值，或者空氣質(zhì)量評(píng)價(jià)為優(yōu)的天數(shù)等。聚合查詢模式的核心思想是按照一定的關(guān)鍵字對(duì)數(shù)據(jù)進(jìn)行分組，然后對(duì)分組數(shù)據(jù)進(jìn)行數(shù)據(jù)的聚合、整理、處理。該模式又可以具體分為數(shù)值概要模式、倒排序索引模式、計(jì)數(shù)器模式等。聚合查詢模式—數(shù)值概要模式數(shù)值概要模式的目的是按照某些鍵值對(duì)數(shù)據(jù)進(jìn)行分組，然后計(jì)算特定分組下數(shù)據(jù)的聚合值，如平均值、最大最小值、求和等。任務(wù)1：統(tǒng)計(jì)每個(gè)城市的PM2.5的最大值、最小值和平均值聚合查詢模式--數(shù)值概要模式聚合查詢模式—倒排序索引模式任務(wù)1的源代碼請(qǐng)參考：教材P238-239mapredstreaming\-input/CityAQ/*\-output/AQI_rlt\-mapper/codes/AQI/-mapper.py\-reducer/codes/AQI/-reducer.py聚合查詢模式—數(shù)值概要模式任務(wù)2：增加combiner，實(shí)現(xiàn)更加高效的數(shù)據(jù)統(tǒng)計(jì)工作；聚合查詢模式—倒排序索引模式任務(wù)3的源代碼請(qǐng)參考：教材P240但該版教材的源代碼有點(diǎn)小bug聚合查詢模式—倒排序索引模式倒排索引源于實(shí)際應(yīng)用中需要根據(jù)屬性的值來(lái)查找記錄，這種索引表中的每一項(xiàng)都包括一個(gè)屬性值和具有該屬性值的記錄的地址。由于不是由記錄確定屬性值，而是由屬性值確定記錄的位置，因而稱為倒排索引(invertedindex)。帶有倒排索引的文件稱為倒排索引文件，簡(jiǎn)稱倒排文件(invertedfile)。聚合查詢模式—倒排序索引模式任務(wù)1：構(gòu)建所有城市指定年份空氣質(zhì)量為優(yōu)的日期倒排序索引。輸入：城市空氣質(zhì)量監(jiān)測(cè)記錄輸出：一個(gè)倒排序文件，每一條記錄格式為<城市,年份,空氣質(zhì)量為優(yōu)的日期列表>為了完成該任務(wù)，在Mapper中需要把城市空氣質(zhì)量符合條件的對(duì)應(yīng)的日期和時(shí)刻加以記錄，并輸出。在Reducer中，其接受到的數(shù)據(jù)序列中，具有相同城市的數(shù)據(jù)相鄰，所以只要把具有相同關(guān)鍵字的鍵值對(duì)的值取出，并形成列表便可以得到空氣質(zhì)量為優(yōu)的日期序列。過(guò)濾模式過(guò)濾模式（FilterPattern）或稱標(biāo)準(zhǔn)模式（CriteriaPattern）是一種設(shè)計(jì)模式，這種模式允許開(kāi)發(fā)人員使用不同的標(biāo)準(zhǔn)來(lái)過(guò)濾一組對(duì)象，通過(guò)邏輯運(yùn)算以解耦的方式把它們連接起來(lái)。這種類型的設(shè)計(jì)模式屬于結(jié)構(gòu)型模式，它結(jié)合多個(gè)標(biāo)準(zhǔn)來(lái)獲得單一標(biāo)準(zhǔn)。過(guò)濾模式最重要的特點(diǎn)是保持?jǐn)?shù)據(jù)的完整性，只是對(duì)記錄保留與否進(jìn)行判斷，輸出滿足條件的記錄。過(guò)濾模式--過(guò)濾器模式過(guò)濾器的功能就是通過(guò)設(shè)置一定的條件篩選并輸出滿足條件的記錄，其作用類似于SQL語(yǔ)句中的過(guò)濾條件。在過(guò)濾器模式的實(shí)現(xiàn)中，由于讀取的記錄中已經(jīng)包含了需要的全部信息，所以簡(jiǎn)單的過(guò)濾模式一般情況下都只需要mapper程序。很多情況下，若最終得到的數(shù)據(jù)是原來(lái)數(shù)據(jù)的子集，并且其體量也大大縮小，那么我們希望這些數(shù)據(jù)能夠輸出到一個(gè)文件中，而不是被分為多個(gè)part進(jìn)行存儲(chǔ)，這時(shí)可以利用reduce階段的排序，再把所有數(shù)據(jù)通過(guò)一個(gè)reducer輸出到一個(gè)文件中。若是一些綜合性任務(wù)，這里的reducer還可能會(huì)在過(guò)濾的基礎(chǔ)上完成其他任務(wù)。過(guò)濾模式--過(guò)濾器模式任務(wù)1：空氣質(zhì)量?jī)?yōu)秀的數(shù)據(jù)檢索。小王是個(gè)運(yùn)動(dòng)愛(ài)好者，他幾乎每天早上6-7點(diǎn)之間都進(jìn)行戶外運(yùn)動(dòng)，為此，他想看看北京市這幾年在他運(yùn)動(dòng)期間的空氣質(zhì)量為“優(yōu)”的詳細(xì)數(shù)據(jù)，請(qǐng)幫他從文件中找出這些數(shù)據(jù)。輸入：所有城市所有日期的空氣質(zhì)量狀況及其各項(xiàng)指標(biāo)；輸出：北京市早上六點(diǎn)時(shí)空氣質(zhì)量為優(yōu)的數(shù)據(jù)記錄。過(guò)濾模式--過(guò)濾器模式過(guò)濾模式--TopK模式TopK模式是指按照一定的排序條件，從所有符合條件的數(shù)據(jù)中挑選出排在最前的K個(gè)元素并輸出。對(duì)于該模式，其基本處理思路是從所有mapper中輸出該節(jié)點(diǎn)數(shù)據(jù)的TopK，之后再reducer中對(duì)所有mapper的TopK進(jìn)行匯總，輸出最終的K的元素。過(guò)濾模式--TopK模式任務(wù)1：空氣質(zhì)量的比較。每年，大眾都會(huì)對(duì)哪些城市是空氣質(zhì)量最好的城市爭(zhēng)論不休，小劉老師提了個(gè)衡量城市質(zhì)量的好方法，他的方法是從全國(guó)所有城市中找出空氣質(zhì)量最好的1000條記錄，再看看這1000條記錄都屬于哪些城市。很明顯，如果一個(gè)城市在前1000名最好記錄里占比最多，那么可以認(rèn)為該城市的空氣質(zhì)量是最好的?，F(xiàn)在請(qǐng)你幫幫小劉老師，從所有城市空氣質(zhì)量數(shù)據(jù)中找到Top1000的空氣質(zhì)量數(shù)據(jù)。由于某一城市某天的空氣質(zhì)量有24條監(jiān)測(cè)數(shù)據(jù)，為了簡(jiǎn)化問(wèn)題，我們假定以正午，即12時(shí)的空氣質(zhì)量作為當(dāng)天的空氣質(zhì)量衡量標(biāo)準(zhǔn)。輸入：所有城市所有的空氣質(zhì)量狀況及其各項(xiàng)指標(biāo)；輸出：所有城市12時(shí)的空氣質(zhì)量Top1000的記錄。過(guò)濾模式--TopK模式分析因?yàn)閿?shù)據(jù)分布在多個(gè)節(jié)點(diǎn)中，要選取TopK數(shù)據(jù)就必須分別從各個(gè)節(jié)點(diǎn)中獲取TopK數(shù)據(jù)，然后在Reduce階段再進(jìn)行多個(gè)K數(shù)據(jù)集的再次選擇。如果map輸出仍舊是以城市、時(shí)間為關(guān)鍵字，那么最終的數(shù)據(jù)有可能被兩個(gè)或者更多的reducer進(jìn)行匯總處理，很顯然這情況下是沒(méi)有辦法找到TopK數(shù)據(jù)的。針對(duì)這一問(wèn)題，有三種方法可以解決：（1）設(shè)置所有mapper的輸出具有相同的鍵值，那么所有數(shù)據(jù)都會(huì)匯總到一個(gè)reducer進(jìn)行處理；（2）設(shè)置只有一個(gè)reducer，這種情況也自然可以把所有數(shù)據(jù)匯總到一個(gè)站點(diǎn)；（3）設(shè)置兩個(gè)reducer階段，其中第二個(gè)reducer階段設(shè)置只有一個(gè)reducer。對(duì)于數(shù)據(jù)量不大，mapper數(shù)量較少的情況，前兩種方法不會(huì)影響整體性能；若數(shù)據(jù)量大，數(shù)據(jù)的站點(diǎn)分布廣，那么第三種方法則會(huì)有更好的表現(xiàn)。TopK模式的應(yīng)用過(guò)程中通常會(huì)用到過(guò)濾模式，或者通過(guò)分組的方式找到每組數(shù)據(jù)的TopK數(shù)據(jù)，該任務(wù)的mapper階段需要先進(jìn)行數(shù)據(jù)的過(guò)濾，去除非正午時(shí)間數(shù)據(jù)，然后再進(jìn)行各個(gè)數(shù)據(jù)節(jié)點(diǎn)上Top1000數(shù)據(jù)的選擇和輸出。因此，該任務(wù)結(jié)合了過(guò)濾器模式與TopK模式兩種模式的綜合應(yīng)用。過(guò)濾模式--去重模式去重模式是指對(duì)輸入的數(shù)據(jù)進(jìn)行對(duì)比，把其中重復(fù)的數(shù)據(jù)去除，從而保證每個(gè)關(guān)鍵字確定的記錄都只有一個(gè)。去重模式是一個(gè)運(yùn)算量較大的操作，若操作方法選擇不恰當(dāng)，則容易導(dǎo)致整個(gè)計(jì)算過(guò)程變得非常緩慢。過(guò)濾模式--去重模式任務(wù)：空氣質(zhì)量上報(bào)數(shù)據(jù)的去重空氣質(zhì)量監(jiān)測(cè)是一個(gè)全國(guó)性的工程，全國(guó)各地都需要定時(shí)上報(bào)具體的空氣質(zhì)量監(jiān)測(cè)數(shù)據(jù)。但是在系統(tǒng)的運(yùn)行初期，由于上報(bào)系統(tǒng)還不夠完善，以及網(wǎng)絡(luò)傳輸?shù)葐?wèn)題，各地經(jīng)常會(huì)出現(xiàn)重復(fù)上報(bào)空氣質(zhì)量數(shù)據(jù)的情況。例如上傳數(shù)據(jù)完成了，可是當(dāng)?shù)刈罱K頁(yè)面沒(méi)有收到確認(rèn)信息，工作人員將再次上傳數(shù)據(jù)。這一問(wèn)題在初期并沒(méi)有受到重視，但是隨著數(shù)據(jù)規(guī)范和要求越來(lái)越嚴(yán)格，現(xiàn)在希望清除歷史數(shù)據(jù)中重復(fù)上報(bào)的記錄，整理出一套不包含重復(fù)數(shù)據(jù)的完整數(shù)據(jù)集。請(qǐng)你幫忙完成這一工作。輸入：所有地方上報(bào)的空氣質(zhì)量數(shù)據(jù)，其中可能包含部分重復(fù)數(shù)據(jù)；輸出：清除所有重復(fù)數(shù)據(jù)后的空氣質(zhì)量數(shù)據(jù)。過(guò)濾模式--去重模式分析：數(shù)據(jù)分布存儲(chǔ)在各個(gè)節(jié)點(diǎn)上，在一個(gè)節(jié)點(diǎn)上運(yùn)行的mapper無(wú)法獲知其包含的數(shù)據(jù)是否也存在于其他節(jié)點(diǎn)，只有當(dāng)數(shù)據(jù)匯總到一起時(shí)才能進(jìn)行唯一性判斷，也才能進(jìn)行重復(fù)數(shù)據(jù)的消除。在實(shí)現(xiàn)過(guò)程中，對(duì)于map操作，若要進(jìn)行數(shù)據(jù)的重復(fù)性檢驗(yàn)則需要進(jìn)行數(shù)據(jù)排序，然后再進(jìn)行數(shù)據(jù)的遍歷，之后把數(shù)據(jù)輸出作為Reducer的輸入；在Reducer之前，MapReduce系統(tǒng)總是需要在進(jìn)行一次數(shù)據(jù)的排序，所以我們可以弱化mapper的功能，讓mapper中把（日期、時(shí)刻、類型、城市、數(shù)據(jù)值）直接輸出，把排序工作交由MapReduce的Shuffle和sort。在Reduce階段，在遍歷所有數(shù)據(jù)，進(jìn)行重復(fù)元素的查找和去除。數(shù)據(jù)連接模式--連接操作定義所謂的數(shù)據(jù)連接操作是指一個(gè)特定數(shù)據(jù)集中的記錄只包含了所需要數(shù)據(jù)的一部分，與其相關(guān)聯(lián)的其余信息則需要利用該記錄中的一些元素值到另外一個(gè)數(shù)據(jù)集中進(jìn)行檢索，并把兩部分?jǐn)?shù)據(jù)合并起來(lái)組成完整的記錄。例如站點(diǎn)表示的空氣質(zhì)量數(shù)據(jù)集中有數(shù)據(jù)“20140513,0,AQI,1013A,77”，我們需要根據(jù)其站點(diǎn)編號(hào)1013A從站點(diǎn)數(shù)據(jù)集中查找得到具體的站點(diǎn)信息，從而得到完整的數(shù)據(jù)“20140513,0,AQI,1013A,77,市監(jiān)測(cè)中心,天津,117.151,39.097”。數(shù)據(jù)連接模式--連接操作定義連接操作是MapReduce數(shù)據(jù)處理中最常用的操作之一，因此，本小節(jié)主要針對(duì)連接操作總結(jié)出該類程序的設(shè)計(jì)模式--連接模式。連接模式是MapReduce設(shè)計(jì)模式中最主要的模式之一，通過(guò)連接操作可以實(shí)現(xiàn)多個(gè)數(shù)據(jù)集的關(guān)聯(lián)查詢。具體而言，連接又可以分為內(nèi)連接、外連接（左外連接、右外連接，全外連接）和笛卡爾積等。數(shù)據(jù)連接模式--連接操作定義數(shù)據(jù)連接模式--連接操作定義內(nèi)連接利用站點(diǎn)編號(hào)進(jìn)行內(nèi)連接操作時(shí)，需要根據(jù)監(jiān)測(cè)數(shù)據(jù)集中的站點(diǎn)編號(hào)在站點(diǎn)信息集中進(jìn)行站點(diǎn)的查找，并把兩條對(duì)應(yīng)的記錄整合，形成包含完整信息的監(jiān)測(cè)數(shù)據(jù)。對(duì)于監(jiān)測(cè)數(shù)據(jù)集中未能在站點(diǎn)信息集中找到站點(diǎn)信息的數(shù)據(jù)，則不加入結(jié)果集中；當(dāng)然，對(duì)站點(diǎn)信息集中存在的站點(diǎn)，若其未能被某一監(jiān)測(cè)數(shù)據(jù)利用上，則在結(jié)果集中也不會(huì)體現(xiàn)該數(shù)據(jù)。數(shù)據(jù)連接模式--連接操作定義外連接使用內(nèi)連接時(shí)，當(dāng)碰到無(wú)法匹配的數(shù)據(jù)時(shí)便把數(shù)據(jù)丟棄，而使用外連接，則按照規(guī)則保留相關(guān)數(shù)據(jù)項(xiàng)。左外連接操作時(shí)，若連接操作為“A連接B”，則保留A中無(wú)法匹配的元素；右外連接操作時(shí)，若連接操作為“A連接B”，則保留B中無(wú)法匹配的元素；全外連接則保留A、B中所有無(wú)法匹配的記錄。數(shù)據(jù)連接模式--連接操作定義數(shù)據(jù)連接模式--連接操作定義數(shù)據(jù)連接模式--連接操作定義笛卡爾積笛卡爾積操作A×B則是把數(shù)據(jù)集A中的所有記錄與數(shù)據(jù)集B中的所有記錄進(jìn)行匹配，從而得到新的數(shù)據(jù)集。數(shù)據(jù)連接模式--內(nèi)連接操作在進(jìn)行連接操作的過(guò)程中，程序會(huì)讀取所要連接的數(shù)據(jù)，并以連接操作鍵值為關(guān)鍵字輸出數(shù)據(jù)；mapper輸出數(shù)據(jù)經(jīng)過(guò)reducer前的排序，可以使得具有相同連接操作鍵值的數(shù)據(jù)形成連續(xù)的輸入序列；最后，在該連續(xù)輸入序列的基礎(chǔ)上進(jìn)行連接操作并輸出最后結(jié)果。需要注意的是，為了能夠在reducer階段區(qū)分?jǐn)?shù)據(jù)記錄的來(lái)源，在mapper輸出數(shù)據(jù)時(shí)需要為兩個(gè)數(shù)據(jù)集來(lái)源的數(shù)據(jù)分別加上標(biāo)識(shí)以便于區(qū)分。數(shù)據(jù)連接模式--內(nèi)連接操作數(shù)據(jù)連接模式--內(nèi)連接操作數(shù)據(jù)連接模式--內(nèi)連接操作數(shù)據(jù)連接模式--內(nèi)連接操作實(shí)現(xiàn)任務(wù)：空氣質(zhì)量數(shù)據(jù)完整信息構(gòu)造與輸出在上述站點(diǎn)數(shù)據(jù)中進(jìn)行查詢，輸出福州地區(qū)所有站點(diǎn)的空氣質(zhì)量數(shù)據(jù)，以作為對(duì)福州地區(qū)空氣質(zhì)量進(jìn)行針對(duì)性研究的數(shù)據(jù)基礎(chǔ)。后續(xù)查詢以兩個(gè)數(shù)據(jù)集連接操作結(jié)果為基礎(chǔ)。輸入：站點(diǎn)監(jiān)測(cè)數(shù)據(jù)集以及站點(diǎn)列表。輸出：空氣質(zhì)量站點(diǎn)檢測(cè)數(shù)據(jù)，其中站點(diǎn)信息完整。需要注意的是，在該任務(wù)中，只需要輸出福州地區(qū)的站點(diǎn)監(jiān)測(cè)數(shù)據(jù)，所以，還需要在mapper中進(jìn)行數(shù)據(jù)過(guò)濾，以減少需要傳輸和連接的數(shù)據(jù)量，即把投影提到連接前完成。用MRJob庫(kù)編寫MapReduce程序06用MRJob庫(kù)編寫MapReduce程序MRJob是一個(gè)Python庫(kù)，實(shí)現(xiàn)了Hadoop的MapReduce操作。它封裝了Hadoopstreaming，可以用全Python腳本實(shí)現(xiàn)HadoopMapReduce計(jì)算，甚至可以在沒(méi)有Hadoop的環(huán)境下完成測(cè)試，并允許用戶將mapper和reducer寫進(jìn)一個(gè)類里。以下是MRJob的一些特性，這些特性使得利用MRJob編寫MapReduce作業(yè)更容易：將一個(gè)作業(yè)的所有MapReduce代碼保存在一個(gè)類中；易于上載和安裝程序代碼和數(shù)據(jù)依賴項(xiàng)；簡(jiǎn)單的一行代碼便可實(shí)現(xiàn)輸入和輸出格式的切換；自動(dòng)下載并分析Python執(zhí)行的錯(cuò)誤日志；允許在Python代碼之前或之后放置命令行過(guò)濾器。第一個(gè)MRJob程序步驟1：安裝MRJobMRJob的安裝由兩種方式，一種是利用pip進(jìn)行MRJob庫(kù)的安裝：$pipinstallmrjob或pip3installmrjob或者從git上下載源碼再進(jìn)行安裝：$pythonsetup.pytest&&pythonsetup.pyinstall安裝完成后，啟動(dòng)Python，并輸入importmrjob，若可以成功引入，則說(shuō)明Python環(huán)境下已經(jīng)成功引入MRJob。設(shè)置好MRJob后，接下來(lái)便可以編寫第一個(gè)MRJob程序了。第一個(gè)MRJob程序步驟2：編寫MRJob程序作為第一個(gè)MRJob程序，我們還是以wordCount為例。首先，建立文件mr_word_count.py，并輸入如下Python代碼。第一個(gè)MRJob程序步驟3運(yùn)行第一個(gè)MRJob程序在mr_word_count.py文件所在文件夾運(yùn)行該程序，其中input.txt可以是自己編寫的一個(gè)文本文件，也可以是系統(tǒng)中任意一個(gè)文本文件（輸出會(huì)因?yàn)檩斎氲膇nput.txt內(nèi)容不同而不同）。第一個(gè)MRJob程序步驟4理解第一個(gè)MRJob程序利用MRJob編寫Python的MapReduce程序時(shí)，一個(gè)job就是一個(gè)繼承自MRJob的一個(gè)類，同時(shí)在該類中定義了job的具體步驟。一個(gè)job可以包括mapper、combiner、reducer三個(gè)步驟，他們分別對(duì)應(yīng)mapper()，combiner()和reducer()函數(shù)。當(dāng)然在具體的MRJob程序中，這些步驟都是可選項(xiàng)，用戶可以根據(jù)需要只定義其中的一個(gè)或幾個(gè)步驟。第一個(gè)MRJob程序步驟4理解第一個(gè)MRJob程序mapper()方法輸入為<key,value>，在這里，key被忽略了，所以key的位置用一個(gè)占位符“_”替代。mapper()的輸出yield產(chǎn)生的<key,value>數(shù)據(jù)對(duì)。輸出是利用split()分割得到的每個(gè)單詞與其值1所形成的元組序列。reducer()方法的輸入為<key,values>，其中的key是mapper輸出的key，而values則不是單個(gè)value，而是一個(gè)可迭代對(duì)象。在上述第一個(gè)MRJob程序中，其輸出為<key,sum(values)>，其中sum(values)是計(jì)算values迭代對(duì)象中數(shù)據(jù)之和。從這里可以看到，MRJob的reducer輸入與Hadoop-streaming的reducer輸入是不同的，后者的輸入是按照key排序后的<key,value>序列，不是一個(gè)可迭代的value集合。最后是把該程序作為一個(gè)job運(yùn)行的啟動(dòng)語(yǔ)句”MRWordCount.run()”。第一個(gè)MRJob程序步驟5運(yùn)行MRJob：假設(shè)在同級(jí)文件夾下存在一個(gè)input.txt文件，那么其運(yùn)行可以是如下幾種方式之一。方式1：$pythonmy_job.pyinput.txt方式2：$pythonmy_job.py<input.txt若有多個(gè)文件都是該MRJob的數(shù)據(jù)來(lái)源，那么可以在程序后列出所有輸入文件，其形式如下所示：方式3：$pythonmy_job.pyinput1.txtinput2.txt還可以結(jié)合文件輸入和STDIN輸入的方式，例如方式4中input1.txt和input2.txt是文件輸入，而input3.txt則是STDIN的方式輸入。方式4：$pythonmy_job.pyinput1.txtinput2.txt-<input3.txt第一個(gè)MRJob程序步驟6運(yùn)行模式：缺省情況下，MRJob是以單Python線程的方式運(yùn)行，這種運(yùn)行方式提供了友好的程序調(diào)試環(huán)境，但這并不是分布式的程序執(zhí)行方式。MRJob程序執(zhí)行方法可以通過(guò)“-r”或者“--runner”來(lái)指定，具體執(zhí)行方式包括：“-rinline”：缺省方式為單線程方式，通常用在程序的調(diào)試階段；“-rlocal”：以多處理器方式運(yùn)行MRJob，這種方式可以模擬部分Hadoop特性；“-rhadoop”：在Hadoop集群上運(yùn)行該MRJob，在這種方式下，其數(shù)據(jù)文件的來(lái)源可以指定為hdfs文件系統(tǒng)中的數(shù)據(jù)文件；如：python3max_word_count.py-rhadoophdfs:///data-ohdfs:///user注：hdfs中不能預(yù)先有/user文件夾第一個(gè)MRJob程序步驟7為了能夠輸出出現(xiàn)次數(shù)最多的單詞，同時(shí)避免標(biāo)點(diǎn)符號(hào)對(duì)單詞分割的影響，需要對(duì)上述程序進(jìn)行三個(gè)方面的改進(jìn)。首先，利用正則表達(dá)式的方式進(jìn)行單詞的匹配，從而消除標(biāo)點(diǎn)符號(hào)等因素對(duì)單詞提取的影響；其次，在第一個(gè)reducer輸出單詞和出現(xiàn)次數(shù)的基礎(chǔ)上又添加了一個(gè)reducer，從而形成多次規(guī)約的程序結(jié)構(gòu)；最后，利用combiner進(jìn)行程序的優(yōu)化，使得mapper所在節(jié)點(diǎn)能夠?qū)?shù)據(jù)進(jìn)行初步的統(tǒng)計(jì)，從而減少數(shù)據(jù)的通信量。具體實(shí)現(xiàn)請(qǐng)參考教材7.8.1MRJob應(yīng)用詳解通過(guò)編寫第一個(gè)MRJob程序可知，為了定義一個(gè)MapReduce程序，只需要重載MRJob中的mapper、combiner和reducer，并且重載steps函數(shù)來(lái)定義MRJob的具體處理流程。在MRJob中，除了上述四個(gè)可被重載的函數(shù)外，還有如表7.14的函數(shù)可以被重載以定義用戶需要的處理流程MRJob應(yīng)用詳解MRJob應(yīng)用詳解任務(wù)的初始化與釋放（SetupandTeardown）Python腳本在每個(gè)任務(wù)中都會(huì)被HadoopStreaming調(diào)用，開(kāi)始Python程序后，從STDIN輸入數(shù)據(jù)給Python。Mrjob將會(huì)在每個(gè)任務(wù)初始化時(shí)調(diào)用*_init()方法，在每個(gè)任務(wù)結(jié)束時(shí)調(diào)用*_final()方法。與之對(duì)應(yīng)的方法有：mapper_init()combiner_init()reducer_init()mapper_final()combiner_final()reducer_final()如果在一個(gè)任務(wù)中需要預(yù)先加載一些文件（例如用到sqlite數(shù)據(jù)庫(kù)）或者創(chuàng)建臨時(shí)文件等操作，就可以使用這些方法的重載來(lái)完成。MRJob應(yīng)用詳解Shell命令作為任務(wù)（SetupandTeardown）Linux的shell提供了大量實(shí)用的小工具，例如grep、wc等，當(dāng)某個(gè)任務(wù)的工作正好與這些工具的功能相同時(shí)，便可以通過(guò)將某個(gè)步驟指定為shell命令，從而完全放棄該步驟的腳本，而利用shell命令代替之。Mrjob提供了支持這種實(shí)現(xiàn)的接口，如果需要這樣做，可以用mapper_cmd、combiner_cmd或reducer_cmd作為MRStep的參數(shù)，或者重載MRJob中對(duì)應(yīng)的方法mapper_cmd()、combiner_cmd()、和reducer_cmd()。需要注意的是：缺省的”-rinline”運(yùn)行Python方式不支持”*_cmd()”模式，若要在本地使用該命令，需要指定運(yùn)行方式為”-rlocal”。MRJob應(yīng)用詳解利用shell命令指定過(guò)濾器MRStep允許用戶為任務(wù)的輸入指定一個(gè)過(guò)濾器，在數(shù)據(jù)送達(dá)具體任務(wù)前對(duì)數(shù)據(jù)進(jìn)行過(guò)濾。其設(shè)定方法可以是為MRStep設(shè)定mapper_pre_filter和reducer_pre_filter的參數(shù)，也可以通過(guò)在MRJob中重載mapper_pre_filter()方法和reducer_pre_filter()方