hadoop--簡(jiǎn)介及原理.ppt

Hadoop介紹,1,Hadoop介紹,2,MapReduce簡(jiǎn)介,Mapreduce是一中用于數(shù)據(jù)處理的編程模型。例子:用Hadoop處理氣象數(shù)據(jù)，找出每年最高氣溫MapReduce工作過(guò)程分為兩個(gè)階段：map階段和reduce階段每個(gè)階段都有鍵/值對(duì)作為輸入和輸出。程序員還定義兩個(gè)函數(shù)map函數(shù)和reduce函數(shù)。用map函數(shù)來(lái)找出年份和氣溫。本例map函數(shù)只是一個(gè)數(shù)據(jù)準(zhǔn)備階段。通過(guò)這種方式建立數(shù)據(jù)，使得reducer函數(shù)能在此基礎(chǔ)上進(jìn)行工作，找出每年最高氣溫。思考下面的輸入數(shù)據(jù),3,MapReduce簡(jiǎn)介(續(xù)),4,這些行以鍵/值對(duì)的方式來(lái)表示map函數(shù),MapReduce簡(jiǎn)介(續(xù)),Map函數(shù)的功能僅僅提取年份和氣溫(粗體顯示)，并將其作為輸出被發(fā)送(氣溫值已被解釋為整數(shù))(1950,0)(1950,22)(1950,-11)(1949,211)(1949,78)Map函數(shù)的輸出先有MapReduce框架處理，然后再被發(fā)送到reduce函數(shù)。這一處理過(guò)程根據(jù)鍵來(lái)對(duì)鍵/值進(jìn)行排序和分組。因此，reduce函數(shù)會(huì)看到如下輸入：(1949,111,78)(1950,0,22,-11)每年的年份后都有一系列氣溫度數(shù)。所有reduce函數(shù)現(xiàn)在必須重復(fù)這個(gè)列表并從中找出最大的度數(shù)：(1949,111)(1950,22)這是最后的輸出：全球氣溫記錄中每年的最高氣溫,5,MapReduce簡(jiǎn)介(續(xù)),上面的數(shù)據(jù)流如下圖所示。在圖的底部是Unix的管道，模擬整個(gè)MapReduce的流程。,6,圖1MapReduce的邏輯數(shù)據(jù)流,MapReduce簡(jiǎn)介(續(xù)),MapReduce的分布式應(yīng)用MapReduce作業(yè)(job)是客戶端執(zhí)行的基本單位：包括輸入數(shù)據(jù),MapReduce程序和配置信息。Hadoop通過(guò)把作業(yè)分成若干個(gè)小任務(wù)(task)來(lái)工作，包括。兩種類(lèi)型的任務(wù):map任務(wù)和reduce任務(wù)。有兩種類(lèi)型的節(jié)點(diǎn)控制著作業(yè)執(zhí)行的過(guò)程：jobtracker和多個(gè)tasktracker。Jobtracker通過(guò)調(diào)度任務(wù)在tasktracker上運(yùn)行來(lái)協(xié)調(diào)所有運(yùn)行在系統(tǒng)上的作業(yè)。Tasktracker運(yùn)行任務(wù)的同時(shí)，把進(jìn)度報(bào)告?zhèn)魉偷絡(luò)obtracker，jobtracker則記錄著每項(xiàng)任務(wù)的整體進(jìn)展情況。如果其中一個(gè)任務(wù)失敗，jobtracker可以重新調(diào)度任務(wù)到另外一個(gè)tasktracker。Hadoop把輸入數(shù)據(jù)劃分成等長(zhǎng)的小數(shù)據(jù)發(fā)送到MapReduce，成為輸入分片(inputsplit)或分片。Hadoop為每個(gè)分片(split)創(chuàng)建一個(gè)map任務(wù)，由它來(lái)運(yùn)行用戶自定義的map函數(shù)來(lái)分析每個(gè)分片的記錄。,7,MapReduce簡(jiǎn)介(續(xù)),Map任務(wù)把輸出寫(xiě)入本地硬盤(pán)，而不是HDFS，因?yàn)閙ap的輸出作為中間輸出，中間輸出被reduce處理后產(chǎn)生最終的輸出，一旦作業(yè)完成，map的輸出就可以刪除了。如果該節(jié)點(diǎn)運(yùn)行的map任務(wù)在map輸出給reduce任務(wù)處理之前崩潰，那么hadoop將在另一個(gè)節(jié)點(diǎn)上重新運(yùn)行map任務(wù)以再次產(chǎn)生map輸出。Reduce任務(wù)并不具備數(shù)據(jù)本地讀取的優(yōu)勢(shì)。一個(gè)單一的reduce任務(wù)往往來(lái)自于所有mapper的輸出。因此，有序map的輸出必須通過(guò)網(wǎng)絡(luò)傳輸?shù)絩educe任務(wù)運(yùn)行的節(jié)點(diǎn)，在那里進(jìn)行合并，然后傳遞到用戶定義的reduce函數(shù)中。Reduce的輸出通常存儲(chǔ)在HDFS中。對(duì)于每個(gè)reduce輸出的HDFS塊，第一個(gè)副本存儲(chǔ)在本地節(jié)點(diǎn)，其他副本存儲(chǔ)在其他機(jī)架節(jié)點(diǎn)中。,8,MapReduce簡(jiǎn)介(續(xù)),一個(gè)單一的reduce任務(wù)的整個(gè)數(shù)據(jù)流如圖2所示虛線框表示節(jié)點(diǎn)，虛線箭頭表示數(shù)據(jù)傳輸?shù)揭粋€(gè)節(jié)點(diǎn)上實(shí)線箭頭表示節(jié)點(diǎn)之間的數(shù)據(jù)傳輸,9,圖2MapReduce中單一reduce任務(wù)的數(shù)據(jù)流圖,MapReduce簡(jiǎn)介(續(xù)),Reduce任務(wù)數(shù)目不是由輸入的大小決定的，而是單獨(dú)具體指定的。如果有多個(gè)reducer,map任務(wù)會(huì)對(duì)其輸出進(jìn)行分區(qū)(partition)，為每個(gè)reduce任務(wù)創(chuàng)建一個(gè)分區(qū)(partition)。每個(gè)分區(qū)包含許多鍵(及其關(guān)聯(lián)的值)，但每個(gè)鍵的記錄都在同一個(gè)分區(qū)中。分區(qū)可以通過(guò)用戶定義的partitioner來(lái)控制，通常是用默認(rèn)的分區(qū)工具，使用hash函數(shù)來(lái)形成“木桶”鍵/值。多個(gè)reduce任務(wù)數(shù)據(jù)流圖如圖3所示。,10,圖3多個(gè)reduce任務(wù)的MapReduce數(shù)據(jù)流,MapReduce簡(jiǎn)介(續(xù)),因?yàn)樘幚砜梢圆⑿羞M(jìn)行，也有可能不存在reduce任務(wù)，不需要shuffle的時(shí)候。這種情況下，唯一的非本地節(jié)點(diǎn)數(shù)據(jù)傳輸是當(dāng)map任務(wù)寫(xiě)入到HDFS中。如圖4。,11,圖4MapReduce中沒(méi)有reduce的數(shù)據(jù)流,MapReduce簡(jiǎn)介(續(xù)),CombinerFunction,12,MapReduce簡(jiǎn)介(續(xù)),CombinerFunction集群的帶寬限制了MapReduce作業(yè)的數(shù)量，因此需要減小map和reduce任務(wù)之間傳輸數(shù)據(jù)量。Hadoop允許用戶聲明一個(gè)combiner，運(yùn)行在map的輸出上，該函數(shù)輸出作為reduce函數(shù)的輸入。Combiner是一個(gè)優(yōu)化方法。例：1950年的讀數(shù)由兩個(gè)map處理(在不同的分片中)，假設(shè)第一個(gè)i餓map的輸出如下:(1950,0)(1950,20)(1950,10)第二個(gè)map輸出如下:(1950,25)(1950,15)Reduce函數(shù)再調(diào)用時(shí)被傳入以下數(shù)字:(1950,0,20,10,25,15)因?yàn)?5是輸入中的最大值，所以輸出如下:(1950,25)當(dāng)使用了combiner，像reduce函數(shù)那樣，為每個(gè)map輸出找到最高氣溫。Reduce函數(shù)被調(diào)用的時(shí)候?qū)⒈粋魅肴缦聰?shù)值:(1950,20,25)Reduce輸出結(jié)果和以前一樣。注：combiner并不能取代reduce函數(shù)，因?yàn)閞educe函數(shù)仍然需要處理來(lái)自不同的map給出的相同鍵記錄。但它可以幫助減少map和reduce之間的數(shù)據(jù)傳輸量。,13,MapReduce簡(jiǎn)介(續(xù)),MapReduce的工作流如何將一個(gè)數(shù)據(jù)處理問(wèn)題轉(zhuǎn)換成MapReduce模型？前面的數(shù)據(jù)處理處理的問(wèn)題都很簡(jiǎn)單，如何處理復(fù)雜的問(wèn)題？(這里的復(fù)雜是指有更多的MapReduce作業(yè)，而不是更復(fù)雜的map和reduce函數(shù)，即增加更多的作業(yè)而不是增加作業(yè)的復(fù)雜度。)對(duì)于復(fù)雜問(wèn)題，可以考慮使用MapReduce之上的高級(jí)語(yǔ)言，如Pig,Hive或Cascading。直接的好處是免于處理到MapReduce作業(yè)的轉(zhuǎn)換，而是將中心集中于正在進(jìn)行的分析上。例：假設(shè)我們想找到每個(gè)氣象臺(tái)年度每天平均氣溫最高的記錄。例如：要就算029070-99999氣象臺(tái)的1月1日的平均每日最高氣溫的記錄，我們將從這個(gè)氣象臺(tái)的1901年1月1日，1902年1月1日直到2000年的1月1日的氣溫中找出每日平均氣溫的最大值。該如何使用MapReduce來(lái)計(jì)算呢？,14,MapReduce簡(jiǎn)介(續(xù)),計(jì)算分為下面兩個(gè)部分1.計(jì)算每個(gè)station-data對(duì)的每日最高氣溫.2.計(jì)算每個(gè)station-day-month鍵(key)的平均每日最高氣溫Mapper從上一個(gè)作業(yè)得到輸出記錄(station-data,最高氣溫值)，并丟掉年份將其值賦予到記錄(station-day-month,最高氣溫值)。Reduce函數(shù)為每個(gè)station-day-month鍵(key)得到一個(gè)平均最高氣溫。第一部分的輸出結(jié)果為：029070-99999190101010029070-9999919020101-94前兩個(gè)字段是鍵(key),最后一列指的是從指定氣象臺(tái)和日期讀入的最高氣溫。第二部分計(jì)算這些年的最高氣溫的平均值029070-999990101-68以上是氣象臺(tái)029070-99999整個(gè)世紀(jì)1月1日平均每日最高氣溫-6.8,15,MapReduce工作原理,Hadoop如何運(yùn)行一個(gè)job？,16,MapReduce工作原理,17,圖5Hapoop運(yùn)行MapReduce作業(yè)的工作原理,在圖中上層有四個(gè)實(shí)體:客戶端，提交MapReduce作業(yè)Jobtracker，協(xié)調(diào)作業(yè)的運(yùn)行。Jobtracker是Java的一個(gè)應(yīng)用程序，主要類(lèi)是JobTracker。Tasktracker，運(yùn)行作業(yè)劃分后的任務(wù)。Tasktracker是一個(gè)Java應(yīng)用程序，主要類(lèi)是TaskTracker。分布式文件系統(tǒng)(HDFS)，用來(lái)在其他實(shí)體間共享作業(yè)文件。,Hadoop運(yùn)行作業(yè)的過(guò)程,JobSubmission(作業(yè)提交)JobClient的runjob()用于方法是產(chǎn)生JobClient實(shí)例和調(diào)用其submitJob()方法的簡(jiǎn)便方法(step1)。提交作業(yè)后，runjob()方法每秒輪訓(xùn)作業(yè)的進(jìn)度，如果發(fā)現(xiàn)與上一個(gè)記錄不同，便把報(bào)告顯示到控制臺(tái)。作業(yè)完成后，如果成功，顯示作業(yè)計(jì)數(shù)器。否則，控制臺(tái)會(huì)顯示導(dǎo)致作業(yè)失敗的錯(cuò)誤。JobClient的submitJob()方法實(shí)現(xiàn)了作業(yè)提交。過(guò)程如下：向jobtracker請(qǐng)求一個(gè)新的作業(yè)ID(通過(guò)調(diào)用JobTracker的getNewJobId()(step2)。檢查作業(yè)的輸出說(shuō)明。比如，如果沒(méi)有指定輸出目錄或者它已經(jīng)存在，作業(yè)就不會(huì)提交，所有錯(cuò)誤返回給MapReduce程序。計(jì)算作業(yè)的輸入劃分。如果劃分無(wú)法計(jì)算，比如因?yàn)檩斎肼窂讲淮嬖?，作業(yè)就不會(huì)被提交，并有錯(cuò)誤返回給MapReduce程序。將運(yùn)行作業(yè)所需要的資源包括作業(yè)JAR文件，配置文件和計(jì)算的輸入劃分(computedinputsplits)復(fù)制到j(luò)obtracker文件系統(tǒng)中一個(gè)以作業(yè)ID號(hào)命名的目錄中。由于作業(yè)JAR副本較多，在tasktracker運(yùn)行作業(yè)時(shí)，集群能夠?yàn)樗鼈兲峁┰S多副本進(jìn)行訪問(wèn)(step3)。告訴jobtracker作業(yè)準(zhǔn)備執(zhí)行(調(diào)用JobTracker的submitjob()方法)(step4)。,18,Hadoop運(yùn)行作業(yè)的過(guò)程(續(xù)),JobInitialization(作業(yè)初始化)JobTracker接收到對(duì)其submitJob()方法的調(diào)用后，會(huì)把此調(diào)用方法一個(gè)內(nèi)部對(duì)列中，交由作業(yè)調(diào)度器進(jìn)行調(diào)度，并對(duì)其進(jìn)行初始化。初始化包括創(chuàng)建一個(gè)代表該正在運(yùn)行的作業(yè)的對(duì)象，它封裝任務(wù)和記錄信息，以便跟蹤任務(wù)的狀態(tài)和進(jìn)程(step5)。要?jiǎng)?chuàng)建運(yùn)行任務(wù)列表，作業(yè)調(diào)度器首先從共享文件系統(tǒng)中獲取JobClient已計(jì)算好的輸入劃分信息(step6)。然后為每個(gè)劃分(split)創(chuàng)建一個(gè)map任務(wù)。TaskAssignment(任務(wù)分配)TaskTracker執(zhí)行一個(gè)簡(jiǎn)單的循環(huán)，定期發(fā)送心跳(heartbeat)方法調(diào)用Jobtracker。心跳方法告訴jobtracker,traktracker是否還存活，同時(shí)也充當(dāng)兩者之間的消息通道。作為心跳方法調(diào)用的一部分，tasktracker會(huì)指明他是否已經(jīng)準(zhǔn)備運(yùn)行新的任務(wù)，如果是，jobtracker會(huì)為它分配一個(gè)任務(wù)，并使用心跳方法的返回值與tasktracker進(jìn)行通信(step7)。,19,Hadoop運(yùn)行作業(yè)的過(guò)程(續(xù)),要選擇一個(gè)reduce任務(wù)，jobtracker只是簡(jiǎn)單的從尚未運(yùn)行的reduce任務(wù)列表中選取下一個(gè)來(lái)執(zhí)行，并沒(méi)有考慮數(shù)據(jù)的本地化。然而，對(duì)于一個(gè)map任務(wù)，它考慮的是tasktracker的網(wǎng)絡(luò)位置和選取一個(gè)距離其輸入劃分文件最近的tasktracker。在理想的情況下，任務(wù)是data-local(數(shù)據(jù)本地化)的，與分割文件所在的節(jié)點(diǎn)運(yùn)行在相同的節(jié)點(diǎn)上。同樣，任務(wù)也可能是rack-local(機(jī)架本地化)的：和分割文件在同一個(gè)機(jī)架，但不在同一節(jié)點(diǎn)。一些任務(wù)既不是數(shù)據(jù)本地化的，也不是機(jī)架本地化的，從與它們自身運(yùn)行的不同機(jī)架上檢索數(shù)據(jù)。可以通過(guò)查看作業(yè)的計(jì)數(shù)器得知每種類(lèi)型任務(wù)的比例。,20,Hadoop運(yùn)行作業(yè)的過(guò)程(續(xù)),TaskExecution(任務(wù)執(zhí)行)Tasktracker本地化作業(yè)的JAR文件，將它從共享文件系統(tǒng)復(fù)制到tasktracker所在的文件系統(tǒng)，同時(shí)，將應(yīng)用程序所需要的全部文件從分布式緩存復(fù)制到本地磁盤(pán)(step8)。然后，為任務(wù)新建一個(gè)本地工作目錄，并把JAR文件中的內(nèi)容解壓到這個(gè)文件夾中。第3步，新建一個(gè)taskrunner實(shí)例來(lái)運(yùn)行任務(wù)。TaskRunner啟動(dòng)一個(gè)新的Java虛擬機(jī)(step9)來(lái)運(yùn)行每個(gè)任務(wù)(step10)，使得用戶定義的map和reduce函數(shù)的任何缺陷都不會(huì)影響tasktracker。子進(jìn)程通過(guò)unbilical接口與父進(jìn)程進(jìn)行通信。它每隔幾秒便告訴父進(jìn)程它的進(jìn)度，直到任務(wù)完成。,21,Hadoop運(yùn)行作業(yè)的過(guò)程(續(xù)),ProgressandStatusUpdates(進(jìn)度和狀態(tài)更新)MapReduce作業(yè)是一個(gè)長(zhǎng)時(shí)間運(yùn)行的批量作業(yè)，能夠得知作業(yè)的進(jìn)展是很重要的。一個(gè)作業(yè)和每個(gè)任務(wù)都有一個(gè)狀態(tài)，這些狀態(tài)信息怎樣與客戶端進(jìn)行通信呢？任務(wù)在運(yùn)行時(shí)，對(duì)任務(wù)進(jìn)度(即任務(wù)完成率)保持追蹤。對(duì)于map任務(wù)，是已處理完輸入的百分比。對(duì)于reduce任務(wù)，是估計(jì)reduce輸入已處理的百分比。JobCompletion(作業(yè)完成)Jobtracker收到作業(yè)最后一個(gè)任務(wù)已完成的通知后，便把作業(yè)的狀態(tài)設(shè)置為“成功”。然后，在JobClient查詢狀態(tài)時(shí),它將得知任務(wù)已成功完成，所以便顯示一條消息告訴用戶，然后從runJob()方法返回。如果jobtracker有相應(yīng)的設(shè)置，也會(huì)發(fā)送一個(gè)HTTP作業(yè)通知，希望收到回調(diào)的客戶端可以通過(guò)job.end.notification.url屬性來(lái)設(shè)置。最后，jobtracker清空作業(yè)的工作狀態(tài)，提示tasktracker也清空作業(yè)的工作狀態(tài)。,22,作業(yè)的調(diào)度,FIFO調(diào)度算法(早期版本)FairScheduler,23,Shuffle和排序,MapReduce保證每個(gè)reduce輸入都已經(jīng)按鍵排序。系統(tǒng)執(zhí)行排序的過(guò)程map輸出傳到reduce作為后者的輸入即成為shuffle(混洗或稱(chēng)洗牌)。學(xué)習(xí)shuffle如何工作，對(duì)于需要優(yōu)化MapReduce程序大有幫助。,24,圖6MapReduce的shuffle和排序,MapReduce的類(lèi)型與格式,Hadoop如何處理格式化數(shù)據(jù)，非格式化數(shù)據(jù)，半格式化數(shù)據(jù)？Hadoop可以處理很多格式的數(shù)據(jù)，從一般的文本文件到一整個(gè)數(shù)據(jù)庫(kù)。InputFormats(輸入格式)InputSplitsandRecord(輸入分片與記錄)TextInput(文本輸入)BinaryInput(二進(jìn)制輸入)MultipleInput(多種輸入)DatabaseInput(數(shù)據(jù)庫(kù)輸入),25,MapReduce的類(lèi)型與格式,26,圖7InputFormat類(lèi)的層次結(jié)構(gòu),MapReduce的類(lèi)型與格式,OutputFormats(輸出格式)TextOutput(文本輸出)BinaryOutput(二進(jìn)制輸出)MultipleOutput(多種輸出)LazyOutput(延遲輸出)DatabaseOutput(數(shù)據(jù)庫(kù)輸出),27,MapReduce的類(lèi)型與格式,28,圖8OutputFormat類(lèi)的層次結(jié)構(gòu),Pig簡(jiǎn)介,什么是Pig?Pig是SQL-like語(yǔ)言,是在MapReduce上構(gòu)建的一種高級(jí)查詢語(yǔ)言,把一些運(yùn)算編譯進(jìn)MapReduce模型的Map和Reduce中,并且用戶可以定義自己的功能。Pig有以下兩部分構(gòu)成：表達(dá)數(shù)據(jù)流的語(yǔ)言，稱(chēng)為PigLatin。運(yùn)行PigLatin程序的執(zhí)行環(huán)境。目前有兩種環(huán)境：?jiǎn)蝹€(gè)JVM本地執(zhí)行和在Hadoop集群上分布運(yùn)行。PigLatin程序有一系列操作或者轉(zhuǎn)換組成，用于將輸入數(shù)據(jù)生成輸出。從總體上看，這些操作描述了一個(gè)數(shù)據(jù)流，Pig執(zhí)行環(huán)境將這個(gè)數(shù)據(jù)流轉(zhuǎn)換成可執(zhí)行的語(yǔ)句然后運(yùn)行。實(shí)際上，Pig將轉(zhuǎn)換變成一系列的MapReduce作業(yè)，它能是人們專(zhuān)注于數(shù)據(jù)而不是執(zhí)行本質(zhì)。,29,Pig簡(jiǎn)介,Pig的特點(diǎn)Pig是一種用于搜索大型數(shù)據(jù)集的腳本語(yǔ)言。MapReduce缺點(diǎn)之一就是開(kāi)發(fā)周期很長(zhǎng)。編寫(xiě)mapper和reducer，編譯和打包代碼，提交作業(yè)和結(jié)果檢索是一個(gè)耗時(shí)的工作。Pig的優(yōu)點(diǎn)就是它能夠通過(guò)從控制臺(tái)發(fā)出6條PigLatin的方法簡(jiǎn)單處理數(shù)TB的數(shù)據(jù)。由于Pig提供了幾條用于反復(fù)檢查程序中數(shù)據(jù)結(jié)構(gòu)的命令，所以它非常支持程序員編寫(xiě)查詢語(yǔ)句。Pig被設(shè)計(jì)成是可擴(kuò)展的。處理路徑中所有部分幾乎都是可定制的：裝載，存儲(chǔ)，過(guò)濾，分組，排序和鏈接都可以換用用戶定義函數(shù)(UDF)。這些功能在Pig內(nèi)嵌數(shù)據(jù)模型上工作，所以他們可以與Pig的平臺(tái)緊密結(jié)合在一起。,30,Pig簡(jiǎn)介,PigPig運(yùn)行的是一個(gè)客戶端應(yīng)用程序。在Hapoop集群上運(yùn)行Pig也不需要在集群