超大集群的簡(jiǎn)單數(shù)據(jù)處理

超大集群的簡(jiǎn)單數(shù)據(jù)處理MapReduce是一種編程模式，它可以實(shí)現(xiàn)處理、產(chǎn)生海量數(shù)據(jù)集。用戶(hù)指定一個(gè)map函數(shù)，通過(guò)這個(gè)map函數(shù)處理鍵值對(duì)，并且產(chǎn)生一系列的中間鍵值對(duì)，然后使用reduce函數(shù)來(lái)合并所有的具有相同key值的中間鍵值對(duì)中值的部分?，F(xiàn)實(shí)生活中的很多任務(wù)的實(shí)現(xiàn)都是基于這個(gè)模式的。使用這樣的函數(shù)形式實(shí)現(xiàn)的程序可以自動(dòng)分布到一個(gè)由普通機(jī)器組成的超大集群上并發(fā)執(zhí)行。run-time系統(tǒng)會(huì)自動(dòng)解決輸入數(shù)據(jù)的分布細(xì)節(jié)、跨越集群的程序執(zhí)行調(diào)度、處理機(jī)器的失效，并且管理機(jī)器之間的通訊請(qǐng)求。這樣的模式允許程序員可以不需要有什么并發(fā)處理或者分布式系統(tǒng)的經(jīng)驗(yàn)，就可以處理超大的分布式系統(tǒng)的資源。MapReduce系統(tǒng)可以在一個(gè)由普通機(jī)器組成的大型集群上實(shí)現(xiàn)運(yùn)行，并且有著很高的擴(kuò)展性：一個(gè)典型的MapReduce計(jì)算處理通常分布到上千臺(tái)機(jī)器上來(lái)處理上TB的數(shù)據(jù)。程序員會(huì)發(fā)現(xiàn)這樣的系統(tǒng)很容易使用：已經(jīng)開(kāi)發(fā)出來(lái)了上百個(gè)MapReduce程序，并且每天在Google的集群上有上千個(gè)MapReducejob正在執(zhí)行。在過(guò)去的5年中，Google實(shí)現(xiàn)了上百個(gè)用于特別計(jì)算目的的程序來(lái)處理海量的原始數(shù)據(jù)，比如蠕蟲(chóng)文檔，web請(qǐng)求log，等等。用于計(jì)算出不同的數(shù)據(jù)，比如降序索引，不同的圖示展示的web文檔，蠕蟲(chóng)采集的每個(gè)host的page數(shù)量摘要，給定日期內(nèi)最常用的查詢(xún)等等。絕大部分計(jì)算都是概念上很簡(jiǎn)潔的。不過(guò)，輸入的數(shù)據(jù)通常是非常巨大的，并且為了能在合理時(shí)間內(nèi)執(zhí)行完畢，其上的計(jì)算必須分布到上百個(gè)或者上千個(gè)計(jì)算機(jī)上去執(zhí)行。如何并發(fā)計(jì)算，如何分布數(shù)據(jù)，如何處理失敗等相關(guān)問(wèn)題合并在一起就會(huì)導(dǎo)致原本簡(jiǎn)單的計(jì)算都掩埋在為了解決這些問(wèn)題而引入的很復(fù)雜的代碼中。因?yàn)檫@種復(fù)雜度，我們?cè)O(shè)計(jì)了一種新的東西來(lái)讓我們能夠方便處理這樣的簡(jiǎn)單計(jì)算。這些簡(jiǎn)單計(jì)算原本很簡(jiǎn)單，但是由于考慮到并發(fā)處理細(xì)節(jié)，容錯(cuò)細(xì)節(jié)，以及數(shù)據(jù)分布細(xì)節(jié)，負(fù)載均衡等等細(xì)節(jié)問(wèn)題，而導(dǎo)致代碼非常復(fù)雜。所以我們抽象這些公共的細(xì)節(jié)到一個(gè)lib中，這種抽象是源自L(fǎng)isp以及其他很多面向功能的語(yǔ)言的map和reduce概念。MapReduce的主要貢獻(xiàn)在于提供了一個(gè)簡(jiǎn)單強(qiáng)大的接口，通過(guò)這個(gè)接口，可以把大尺度的計(jì)算自動(dòng)的并發(fā)和分布執(zhí)行。使用這個(gè)接口，可以通過(guò)普通PC的巨大集群，來(lái)達(dá)到極高的性能。對(duì)于MapReduce的編程模式，運(yùn)算處理一組輸入的（input）鍵值對(duì)（key/valuepairs），就會(huì)產(chǎn)生一組輸出的（output）鍵值對(duì)。MapReduce函數(shù)庫(kù)由用戶(hù)用兩個(gè)函數(shù)來(lái)表達(dá)這樣的計(jì)算：Map和ReduceoMap函數(shù)，是用戶(hù)自定義的的函數(shù)，處理輸入的鍵值對(duì)，并且產(chǎn)生一組中間的鍵值對(duì)。MapReduce函數(shù)庫(kù)整合所有相同的中間鍵值，發(fā)送給Reduce函數(shù)進(jìn)行處理。Reduce函數(shù)同樣也是用戶(hù)提供的，它處理中間鍵值以及這個(gè)中間鍵值相關(guān)的值集合。這個(gè)函數(shù)合并這些值，最后形成一個(gè)相對(duì)較小的值集合。通常一個(gè)單次Reduce執(zhí)行會(huì)產(chǎn)生0個(gè)或者1個(gè)輸出值。提供給Reduce函數(shù)的中間值是通過(guò)一個(gè)iterator來(lái)提供的。這就讓我們可以處理超過(guò)內(nèi)存容量的值列表。從概念上講，使用者提供的map和reduce函數(shù)有著如下相關(guān)類(lèi)型：map(k1,v1)->list(k2,v2)reduce(k2,list(v2))->list(v2)也就是，輸入的鍵值和輸出的鍵值是屬于不同的域的。進(jìn)一步說(shuō)，中間的鍵值是和輸出的鍵值屬于相同的域的。另外還有一些可以簡(jiǎn)單的通過(guò)MapReduce計(jì)算模型來(lái)展示；分布式Grep、URL訪(fǎng)問(wèn)頻率統(tǒng)計(jì)、逆向Web-Link圖、主機(jī)關(guān)鍵向量指標(biāo)、逆序索引、分布式排序等。接下來(lái)將說(shuō)明MapReduce的實(shí)現(xiàn)。MapReduce接口可以有很多種不同的實(shí)現(xiàn)。應(yīng)當(dāng)根據(jù)不同的環(huán)境選擇不同的實(shí)現(xiàn)。比如，一個(gè)實(shí)現(xiàn)可以適用于小型的共享內(nèi)存的機(jī)器，另一個(gè)實(shí)現(xiàn)可能是基于大型NUMA多處理器系統(tǒng)，還可能有為大規(guī)模計(jì)算機(jī)集群的實(shí)現(xiàn)。Map操作通過(guò)把輸入數(shù)據(jù)進(jìn)行分區(qū)(比如分為M塊)，就可以分布到不同的機(jī)器上執(zhí)行了。Reduce操作是通過(guò)對(duì)中間產(chǎn)生的key的分布來(lái)進(jìn)行分布的，中間產(chǎn)生的key可以根據(jù)某種分區(qū)函數(shù)進(jìn)行分布(比如hash(key)modR)，分布成為R塊。分區(qū)(R)的數(shù)量和分區(qū)函數(shù)都是由用戶(hù)指定的。下圖將說(shuō)明MapReduce操作的整體數(shù)據(jù)流。

ProgramMaster.assign(6)writeworkerworkeroutputworkerworkeroutputfileO對(duì)于每一個(gè)map和reduce任務(wù)來(lái)說(shuō)，都需要保存它的狀態(tài)（idle,U)fork.*ProgramMaster.assign(6)writeworkerworkeroutputworkerworkeroutputfileO對(duì)于每一個(gè)map和reduce任務(wù)來(lái)說(shuō)，都需要保存它的狀態(tài)（idle,U)fork.*-' (1)fork(l)forkIntermediatefiles(onlocaldisks)ReducephaseOutputfiles⑵，assignreduce.(5)remolen?ad(4)localwritein-progress或者completed），并且識(shí)別不同的workerM器（對(duì)于非idel的任務(wù)狀態(tài)）。master是一個(gè)由map任務(wù)產(chǎn)生的中間區(qū)域文件位置信息到reduce任務(wù)的一個(gè)通道。因此，對(duì)于每一個(gè)完成得map任務(wù)，master保存下來(lái)這個(gè)map任務(wù)產(chǎn)生的R中間區(qū)域文件信息的位置和大小。對(duì)于這個(gè)位置和大小信息是當(dāng)接收到map任務(wù)完成得時(shí)候做的。這些信息是增量推送到處于in-progress狀態(tài)的reduce任務(wù)的worker上的。由于MapReduce函數(shù)庫(kù)是設(shè)計(jì)用于在成百上千臺(tái)機(jī)器上處理海量數(shù)據(jù)的，所以這個(gè)函數(shù)庫(kù)必須考慮到機(jī)器故障的容錯(cuò)處理。master會(huì)定期ping每一個(gè)worker機(jī)器。如果在一定時(shí)間內(nèi)沒(méi)有worker機(jī)器的返回，master就認(rèn)為這個(gè)worker失效了。所有這臺(tái)worker完成的map任務(wù)都被設(shè)置成為他們的初始idel狀態(tài)，并且因此可以被其他worker所調(diào)度執(zhí)行。類(lèi)似的，所有這個(gè)機(jī)器上正在處理的map任務(wù)或者reduce任務(wù)都被設(shè)置成為idle狀態(tài)，可以被其他worker所重新執(zhí)行。在失效機(jī)器上的已經(jīng)完成的map任務(wù)還需要再次重新執(zhí)行，這是因?yàn)橹虚g結(jié)果存放在這個(gè)失效的機(jī)器上，所以導(dǎo)致中間結(jié)果無(wú)法訪(fǎng)問(wèn)。已經(jīng)完成的recude任務(wù)無(wú)需再次執(zhí)行，因?yàn)樗麄兊慕Y(jié)果已經(jīng)保存在全局的文件系統(tǒng)中了。當(dāng)map任務(wù)首先由Aworker執(zhí)行，隨后被Bworker執(zhí)行的時(shí)候(因?yàn)槿胧Я?，所有執(zhí)行reduce任務(wù)的worker都會(huì)被通知。所有還沒(méi)有來(lái)得及從A上讀取數(shù)據(jù)的worker都會(huì)從B上讀取數(shù)據(jù)。MapReduce可以有效地支持到很大尺度的worker失效的情況。在master中，定期會(huì)設(shè)定checkpoint，寫(xiě)出master的數(shù)據(jù)結(jié)構(gòu)。如果master任務(wù)失效了，可以從上次最后一個(gè)checkpoint開(kāi)始啟動(dòng)另一個(gè)master進(jìn)程。不過(guò)，由于只有一個(gè)master在運(yùn)行，所以他如果失效就比較麻煩，因此我們當(dāng)前的實(shí)現(xiàn)上，是如果master失效了，就終止MapReduce執(zhí)行?？蛻?hù)端可以檢測(cè)這種失效并且如果需要就重新嘗試MapReduce操作。如果上邊我們講的，我們把map階段拆分到M小塊，并且reduce階段拆分到R小塊執(zhí)行。在理想狀態(tài)下，M和日應(yīng)當(dāng)比worker機(jī)器數(shù)量要多得多。每一個(gè)worker機(jī)器都通過(guò)執(zhí)行大量的任務(wù)來(lái)提高動(dòng)態(tài)的負(fù)載均衡能力，并且能夠加快故障恢復(fù)的速度：這個(gè)失效機(jī)器上執(zhí)行的大量map任務(wù)都可以分布到所有其他worker機(jī)器上執(zhí)行。但是我們的實(shí)現(xiàn)中，實(shí)際上對(duì)于M和R的取值有一定的限制，因?yàn)閙aster必須執(zhí)行O(M+R)次調(diào)度，并且在內(nèi)存中保存O(M*R)個(gè)狀態(tài)。進(jìn)一步來(lái)說(shuō)，用戶(hù)通常會(huì)指定R的值，因?yàn)槊恳粋€(gè)reduce任務(wù)最終都是一個(gè)獨(dú)立的輸出文件。在實(shí)際中，我們傾向于調(diào)輸?shù)闹?，使得每一個(gè)獨(dú)立任務(wù)都是處理大約16M到64M的輸入數(shù)據(jù)（這樣，上面描寫(xiě)的本地優(yōu)化策略會(huì)最有效），另外，我們使R比較小，這樣使得R占用不多的worker機(jī)器。我們通常會(huì)用這樣的比例來(lái)執(zhí)行MapReduce:M=200，000，R=5，000，使用2，000臺(tái)worker機(jī)器。通常情況下，一個(gè)MapReduce的總執(zhí)行時(shí)間會(huì)受到最后的幾個(gè)”拖后腿”的任務(wù)影響：在計(jì)算過(guò)程中，會(huì)有一個(gè)機(jī)器過(guò)了比正常執(zhí)行時(shí)間長(zhǎng)得多的時(shí)間還沒(méi)有執(zhí)行完map或者reduce任務(wù)，導(dǎo)致MapReduce總?cè)蝿?wù)不能按時(shí)完成。出現(xiàn)拖后腿的情況有很多原因。比如：一個(gè)機(jī)器的硬盤(pán)有點(diǎn)問(wèn)題，經(jīng)常需要反復(fù)讀取糾錯(cuò)，然后把讀取輸入數(shù)據(jù)的性能從30M/s降低到1M/s。cluster調(diào)度系統(tǒng)已經(jīng)在某臺(tái)機(jī)器上調(diào)度了其他的任務(wù)，所以因?yàn)镃PU/內(nèi)存/本地硬盤(pán)/網(wǎng)絡(luò)帶寬等競(jìng)爭(zhēng)的關(guān)系，導(dǎo)致執(zhí)行MapReduce的代碼性能比較慢。我們最近出現(xiàn)的一個(gè)問(wèn)題是機(jī)器的啟動(dòng)代碼有問(wèn)題，導(dǎo)致關(guān)閉了cpu的cache：在這些機(jī)器上的任務(wù)性能有上百倍的影響。我們有一個(gè)通用的機(jī)制來(lái)減少拖后腿的情況。當(dāng)MapReduce操作接近完成的時(shí)候，master調(diào)度備用進(jìn)程來(lái)執(zhí)行那些剩下的in-progress狀態(tài)的任務(wù)。無(wú)論當(dāng)最初的任務(wù)還是backup任務(wù)執(zhí)行完成的時(shí)候，都把這個(gè)任務(wù)標(biāo)記成為已經(jīng)完成。我們調(diào)優(yōu)了這個(gè)機(jī)制，通常只會(huì)占用多幾個(gè)百分點(diǎn)的機(jī)器資源。但是我們發(fā)現(xiàn)這樣做以后對(duì)于減少超大MapReduce操作的總處理時(shí)間來(lái)說(shuō)非常有效。到目前為止，最成功的MapReduce的應(yīng)用就是重寫(xiě)了Googleweb搜索服務(wù)所使用到的index系統(tǒng)。索引系統(tǒng)處理蠕蟲(chóng)系統(tǒng)抓回來(lái)的超大量的數(shù)據(jù)，這些數(shù)據(jù)保存在GFS文件里。普通這些文檔的大小是超過(guò)了20TB的數(shù)據(jù)。索引程序是通過(guò)一系列的，大概5到10次MapReduce操作來(lái)建立索引。通過(guò)利用MapReduce有這樣一些好處:?索引代碼很簡(jiǎn)單，很小，很容易理解。因?yàn)閷?duì)于容錯(cuò)的處理代碼，分布以及并行處理代碼都通過(guò)MapReduce函數(shù)庫(kù)封裝了，所以索引代碼很簡(jiǎn)單，很小，很容易理解。例如，當(dāng)使用MapReduce函數(shù)庫(kù)的時(shí)候，計(jì)算的代碼行數(shù)從原來(lái)的3800行C++代碼一下減少到大概700行代碼。?MapReduce的函數(shù)庫(kù)的性能已經(jīng)非常好，所以我們可以把概念上不相關(guān)的計(jì)算步驟分開(kāi)處理，而不是混在一起以期減少處理次數(shù)。這使得我們?nèi)菀赘淖兯饕幚矸绞健?索引系統(tǒng)的操作更容易了，這是因?yàn)闄C(jī)器的失效，速度慢的機(jī)器，以及網(wǎng)絡(luò)風(fēng)暴都已經(jīng)由MapReduce自己解決了，而不需要操作人員的交互。此外，我們可以簡(jiǎn)單的通過(guò)對(duì)索引系統(tǒng)增加機(jī)器的方式提高處理性能。MapReduce的編程模式在Google成功應(yīng)用于許多方面。我們把這種成功應(yīng)用歸結(jié)為幾個(gè)方面：首先，這個(gè)編程模式易于使用，即使程序員沒(méi)有并行或者分布式系統(tǒng)經(jīng)驗(yàn)，由于Ma