hadoop--簡介及原理.ppt

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