從WordCount看Spark大數(shù)據(jù)處理的核心機(jī)制

責(zé)任編輯:editor005

作者:YoyaProgrammer

2015-06-05 13:40:00

摘自:優(yōu)雅程序員

既然反復(fù)寫磁盤和從磁盤加載數(shù)據(jù)使得性能下降,那就把數(shù)據(jù)放到內(nèi)存中,這就是Spark基于內(nèi)存的彈性分布式數(shù)據(jù)集(RDD)的出發(fā)點(diǎn)。

在上一篇文章中,我們講了Spark大數(shù)據(jù)處理的可擴(kuò)展性和負(fù)載均衡,今天要講的是更為重點(diǎn)的容錯(cuò)處理,這涉及到Spark的應(yīng)用場(chǎng)景和RDD的設(shè)計(jì)來(lái)源。

Spark的應(yīng)用場(chǎng)景

Spark主要針對(duì)兩種場(chǎng)景:

機(jī)器學(xué)習(xí),數(shù)據(jù)挖掘,圖應(yīng)用中常用的迭代算法(每一次迭代對(duì)數(shù)據(jù)執(zhí)行相似的函數(shù))

交互式數(shù)據(jù)挖掘工具(用戶反復(fù)查詢一個(gè)數(shù)據(jù)子集)

Spark在spark-submit外,還提供了spark-shell,它就是專門用來(lái)做交互數(shù)據(jù)挖掘的工具

MapReduce等框架并不明確支持迭代中間結(jié)果/數(shù)據(jù)子集的共享,所以需要將數(shù)據(jù)輸出到磁盤,然后在每次查詢時(shí)重新加載,這帶來(lái)較大的開(kāi)銷。

既然反復(fù)寫磁盤和從磁盤加載數(shù)據(jù)使得性能下降,那就把數(shù)據(jù)放到內(nèi)存中,這就是Spark基于內(nèi)存的彈性分布式數(shù)據(jù)集(RDD)的出發(fā)點(diǎn)。

自動(dòng)容錯(cuò)

MapReduce是容錯(cuò)性非常好的系統(tǒng)。處理一步就放到磁盤,再處理一步又放到磁盤,一旦哪一步有問(wèn)題,重做就好了,真可謂是一步一個(gè)腳印。Spark為了上述場(chǎng)景下的性能,把數(shù)據(jù)放在內(nèi)存中,那整個(gè)系統(tǒng)的容錯(cuò)就成了最困難的地方。

一般來(lái)說(shuō),分布式數(shù)據(jù)集的容錯(cuò)性有兩種方式:即數(shù)據(jù)檢查點(diǎn)和記錄數(shù)據(jù)的更新。由于面向的是大規(guī)模數(shù)據(jù)分析,數(shù)據(jù)檢查點(diǎn)操作成本很高:需要通過(guò)數(shù)據(jù)中心的網(wǎng)絡(luò)連接在機(jī)器之間復(fù)制龐大的數(shù)據(jù)集,而網(wǎng)絡(luò)帶寬往往比內(nèi)存帶寬低得多,同時(shí)還需要消耗更多的存儲(chǔ)資源(在內(nèi)存中復(fù)制數(shù)據(jù)可以減少需要緩存的數(shù)據(jù)量,而存儲(chǔ)到磁盤則會(huì)拖慢應(yīng)用程序)。所以選擇記錄更新的方式。但是,如果更新太多,那么記錄更新成本也不低。因此,RDD只支持讀操作,并且只支持粗粒度轉(zhuǎn)換,即在大量記錄上執(zhí)行的單個(gè)操作。將創(chuàng)建RDD的一系列轉(zhuǎn)換記錄下來(lái)(即Lineage),以便恢復(fù)丟失的分區(qū)。

雖然只支持粗粒度轉(zhuǎn)換限制了編程模型,但是RDD仍然可以很好地適用于很多應(yīng)用,特別是支持?jǐn)?shù)據(jù)并行的批量分析應(yīng)用,包括數(shù)據(jù)挖掘、機(jī)器學(xué)習(xí)、圖算法等,因?yàn)檫@些程序通常都會(huì)在很多記錄上執(zhí)行相同的操作。

RDD抽象

RDD是只讀的、分區(qū)記錄的集合。RDD只能基于在穩(wěn)定物理存儲(chǔ)中的數(shù)據(jù)集和其他已有的RDD上執(zhí)行確定性操作來(lái)創(chuàng)建。這些確定性操作稱之為轉(zhuǎn)換,如map、filter、groupBy、join(轉(zhuǎn)換不是程開(kāi)發(fā)人員在RDD上執(zhí)行的操作)。

RDD含有如何從其他RDD計(jì)算出本RDD的相關(guān)信息(即Lineage),據(jù)此可以從物理存儲(chǔ)的數(shù)據(jù)計(jì)算出相應(yīng)的RDD分區(qū)。

在需要反復(fù)使用的某個(gè)數(shù)據(jù)集時(shí),使用RDD的持久化,即persist,這個(gè)持久化優(yōu)先是放在內(nèi)存中的。

再來(lái)看看WordCount

說(shuō)了這么多,我們依然拿WordCount來(lái)說(shuō)說(shuō),幫忙小伙伴們理解,還沒(méi)有看本系列前兩篇文章的童鞋抓緊去看看哈。

val file = "hdfs://127.0.0.1:9000/file.txt"

val lines = sc.textFile(file)

val words = lines.flatMap(line => line.split("\s+"))

val partialCountMap = words

.mapPartitions(convertWordsInPartitionToWordCountMap)

val wordCount = distCountMap.reduce(mergeMaps)

WordCount一共涉及到三個(gè)RDD,用于承載文本行的lines,用于承載單詞的words,用于承載每個(gè)文件塊上部分單詞計(jì)數(shù)的 partialCountMap。Lineage關(guān)系:partialCountMap的父RDD為words,words的父RDD為lines,如下圖:

Word Count Lineage

  有了Lineage和RDD的只讀特性,就可以輕松完成容錯(cuò)了。

如果words在slave1上的一個(gè)分區(qū)出問(wèn)題了,那么我們只需要加載slave1上對(duì)應(yīng)的文件塊,并重新計(jì)算其lines對(duì)應(yīng)的分區(qū),進(jìn)而計(jì)算得到words的這個(gè)分區(qū)。

圖中每個(gè)slave中只畫了一個(gè)文件塊,實(shí)際上可能有多個(gè)文件塊。一定要注意的是哪個(gè)分區(qū)出問(wèn)題了,只會(huì)重算這一個(gè)分區(qū),也就只會(huì)重新加載這個(gè)分區(qū)關(guān)聯(lián)的文件塊。

上面討論的是窄依賴的情況,如果像groupBy這種轉(zhuǎn)換,一個(gè)RDD分區(qū)需要依賴父RDD的多個(gè)分區(qū),那么一個(gè)分區(qū)掛了,就需要計(jì)算父RDD中的多個(gè)分區(qū)。

分布式系統(tǒng)的三個(gè)問(wèn)題:可擴(kuò)展性,負(fù)載均衡,容錯(cuò)處理,都解決了吧。

不知道看到這里的小伙伴,心里是否有個(gè)疑問(wèn),既然RDD的API只支持粗粒度的轉(zhuǎn)換,它真的能夠支持這么多千奇百怪的應(yīng)用場(chǎng)景嗎?下一篇,我們一起看RDD的API,以及它對(duì)其它大數(shù)據(jù)處理框架能夠處理的應(yīng)用場(chǎng)景的等效解決方案。

鏈接已復(fù)制,快去分享吧

企業(yè)網(wǎng)版權(quán)所有?2010-2024 京ICP備09108050號(hào)-6京公網(wǎng)安備 11010502049343號(hào)