2013年年底，我第一次接觸到Spark，當(dāng)時(shí)我對(duì)Spark所使用的Scala語(yǔ)言產(chǎn)生了較大的興趣。一段時(shí)間后，我做了一個(gè)預(yù)測(cè)泰坦尼克號(hào)船上人員生存概率的數(shù)據(jù)科學(xué)項(xiàng)目。事實(shí)證明這是一個(gè)更深入了解Spark概念和編程框架的絕佳途徑。我強(qiáng)烈建議任何希望學(xué)習(xí)Spark的開(kāi)發(fā)者都尋找一個(gè)項(xiàng)目入手。

如今，諸如亞馬遜、eBay和雅虎等公司都開(kāi)始采用Spark技術(shù)。許多機(jī)構(gòu)將Spark部署在上千個(gè)節(jié)點(diǎn)的集群中。據(jù)Spark FAQ中記錄，已知的最大集群節(jié)點(diǎn)個(gè)數(shù)已超過(guò)8000。事實(shí)上，Spark是一項(xiàng)非常值得學(xué)習(xí)的技術(shù)。

本文主要介紹了Spark概念及一些實(shí)例。這些信息主要從Apache Spark網(wǎng)站和相關(guān)書籍中獲取。

什么是Apache Spark?

正如廣告所提到的，Spark是一個(gè)運(yùn)算速度快如閃電的Apache項(xiàng)目。它有一個(gè)逐漸壯大的開(kāi)源社區(qū)，同時(shí)它還是現(xiàn)今最熱門的Apache項(xiàng)目。

Spark提供了一個(gè)運(yùn)算速度快的一般化數(shù)據(jù)處理平臺(tái)。Spark可以讓你的程序提高100倍的內(nèi)存計(jì)算速度，或者10倍的磁盤計(jì)算速度(Hadoop)。去年的Daytona GraySort比賽中，Spark只用了Hadoop十分之一數(shù)量的機(jī)器就實(shí)現(xiàn)了其三倍多的速度。Spark已經(jīng)成了處理PB級(jí)別數(shù)據(jù)運(yùn)算速度最快的開(kāi)源工具。

Spark還可以更加便捷地完成你的項(xiàng)目，為了更好地說(shuō)明這個(gè)問(wèn)題，我們首先看下如何實(shí)現(xiàn)大數(shù)據(jù)中的“Hello World!”案例。用Java語(yǔ)言編寫的MapReduce過(guò)程需要大約50行的代碼，而利用Spark你只需要以下幾行代碼：

sparkContext.textFile("hdfs://...") 
    .flatMap(line => line.split(" ")) 
    map(word => (word, 1)).reduceByKey(_+_) 
    .saveAsTextFile("hdfs://...") 

學(xué)習(xí)如何使用Apache Spark時(shí)，另一個(gè)重要的東西是其提供了脫機(jī)的交互式shell(REPL)。利用REPL，我們可以逐行檢測(cè)代碼是否有誤。考慮到Spark較為簡(jiǎn)短的代碼風(fēng)格，這使得即時(shí)數(shù)據(jù)分析任務(wù)成了容易實(shí)現(xiàn)的事情。

此外，Spark還提供了其他的一些特性：

• 當(dāng)前Spark提供了Scala、Java、Python和R的API接口
• 較好地整合了Hadoop生態(tài)系統(tǒng)和數(shù)據(jù)儲(chǔ)存系統(tǒng)(HDFS, Amazon S3, HIVE, HBase, Cassandra等)
• 既可以在Hadoop YARN或者Apache Mesos等集群上運(yùn)行，也可以單機(jī)運(yùn)行。

Spark核心組件可以和其他一些高效的軟件庫(kù)無(wú)縫連接使用。這些軟件庫(kù)包括SparkSQL, Spark Streming, MLlib(機(jī)器學(xué)習(xí)專用)和GraphX，下文將詳細(xì)介紹這些組件。其他一些軟件庫(kù)和擴(kuò)展功能目前正處于開(kāi)發(fā)過(guò)程中。

Spark核心組件

Spark是處理大規(guī)模數(shù)據(jù)的并行分布式基礎(chǔ)引擎。它主要負(fù)責(zé)以下幾個(gè)功能：

• 內(nèi)存管理和故障恢復(fù)
• 制定并管理集群中的任務(wù)
• 和數(shù)據(jù)儲(chǔ)存系統(tǒng)交互

Spark引入了RDD(Resilient Distributed Dataset)的概念，RDD是一個(gè)抽象的數(shù)據(jù)集，它提供對(duì)數(shù)據(jù)并行和容錯(cuò)的處理。我們可以通過(guò)加載外部數(shù)據(jù)集或從驅(qū)動(dòng)程序集中切分得到一個(gè)可以包含任意類型項(xiàng)目的RDD。

RDD支持兩種類型的運(yùn)算：

• 數(shù)據(jù)轉(zhuǎn)換(數(shù)據(jù)映射、過(guò)濾、合并等)在一個(gè)RDD上執(zhí)行，而其結(jié)果被儲(chǔ)存到另外一個(gè)RDD中。
• 數(shù)據(jù)運(yùn)算(降維、計(jì)數(shù)等)則是通過(guò)在RDD中計(jì)算后才返回相應(yīng)的結(jié)果。

Spark的數(shù)據(jù)轉(zhuǎn)換過(guò)程并不是實(shí)時(shí)返回運(yùn)算結(jié)果。實(shí)際上，該過(guò)程知識(shí)記錄下需要執(zhí)行的操作過(guò)程和相應(yīng)的數(shù)據(jù)集。只有當(dāng)執(zhí)行數(shù)據(jù)運(yùn)算過(guò)程且結(jié)果已經(jīng)返回到驅(qū)動(dòng)程序中時(shí)，Spark才執(zhí)行數(shù)據(jù)轉(zhuǎn)換進(jìn)程。該設(shè)計(jì)使得Spark可以更高效地執(zhí)行任務(wù)。例如，如果一個(gè)大型數(shù)據(jù)集被轉(zhuǎn)換成許多子集并被傳輸?shù)降谝徊降臄?shù)據(jù)運(yùn)算過(guò)程中，那么此時(shí)Spark只能處理并返回第一步的運(yùn)算結(jié)果，并無(wú)法處理整個(gè)數(shù)據(jù)集的運(yùn)算過(guò)程。

默認(rèn)設(shè)定下，任何一個(gè)處理數(shù)據(jù)轉(zhuǎn)換過(guò)程的RDD將會(huì)在每次處理完數(shù)據(jù)運(yùn)算后被還原。然而，你也可以使用高速緩存的方法將RDD保存下來(lái)，此時(shí)Spark會(huì)將內(nèi)容儲(chǔ)存在集群中以便于下次更快捷地調(diào)用。

Spark SQL

SparkSQL是Spark的一個(gè)組件，它可以利用SQL或者Hive查詢語(yǔ)法來(lái)查詢數(shù)據(jù)。它起先被視為MapReduce的替代方案，現(xiàn)今SparkSQL已被整合到Spark堆棧中。為了提供對(duì)更多數(shù)據(jù)類型的支持，它將SQL語(yǔ)句納入系統(tǒng)中，這使其成為一個(gè)非常強(qiáng)大的工具。以下是Hive兼容查詢語(yǔ)句的實(shí)例：

// sc is an existing SparkContext 
val sqlContext = new org.apache.spark.sql.hive.HiveCONTEXT(sc) 
 
sqlContext.sql("CREATE TABLE IF NOT EXISTS src (key INT, value STRING)") 
sqlContext.sql("LOAD DATA LOCAL INPATH 'examoles/src/main/resources/kvl.txt' INTO TABLE src") 
 
// Queries are expressed in HiveQL 
sqlContext.sql("FROM src SELECT key, value").collect().foreach(println) 

Spark Streaming

Spark Streaming支持實(shí)時(shí)流式數(shù)據(jù)處理，比如Web服務(wù)器日志文件、Twitter等社交網(wǎng)絡(luò)數(shù)據(jù)和類似Kafka的信息數(shù)據(jù)。Spark中，Spark Streaming接收輸入流數(shù)據(jù)并將其劃分成小子集。接下來(lái)，如下圖所示，這些數(shù)據(jù)被Spark引擎所處理并被整合成最終的結(jié)果。

Spark Streaming的API接口和Spark核心組件非常匹配，因此所有的編程人員可以輕易地處理流式數(shù)據(jù)。

MLlibMLlib是一個(gè)機(jī)器學(xué)習(xí)庫(kù)，它提供了為大規(guī)模集群計(jì)算所設(shè)計(jì)的分類、回歸、聚類和協(xié)同過(guò)濾等機(jī)器學(xué)習(xí)算法。其中一部分算法也適用于處理流式數(shù)據(jù)，比如普通線性二乘回歸估計(jì)和k均值聚類算法。值得注意的是，Apache Mahout(Hadoop的機(jī)器學(xué)習(xí)算法軟件庫(kù))已經(jīng)脫離MapReduce陣營(yíng)轉(zhuǎn)而投向Spark MLlib中。

GraphX

[[162837]]

GraphX是用于繪圖和執(zhí)行繪圖并行計(jì)算的軟件庫(kù)，它為ETL(探索性分析和反復(fù)的繪圖計(jì)算)提供了一套統(tǒng)一的工具。除了繪圖操作技巧，它還提供了類似于PageRank的一般性繪圖算法。

如何使用Apache Spark?

TwitterUtils.createStream(...) 
    .filter(_.getText.contains("earthquake") || _.getText.contains("shaking")) 

現(xiàn)在我們已經(jīng)回答了“什么是Apache Spark?”這個(gè)問(wèn)題，接下來(lái)讓我們思考下Spark可以用來(lái)處理啥樣的問(wèn)題呢?

最近，我偶然看到一篇文章。文章中提到利用Twitter流式數(shù)據(jù)來(lái)檢測(cè)地震。有趣的是，實(shí)驗(yàn)表明該技術(shù)可以比日本氣象局更快地告知百姓地震的情況。即使他們?cè)谖恼率褂昧瞬煌姆椒?，但我認(rèn)為這是一個(gè)很好的例子，它可以用來(lái)檢驗(yàn)我們?nèi)绾卫煤?jiǎn)化的片段代碼而不使用粘接代碼將Spark付諸實(shí)踐。

首先，我們必須將與“地震”或者“抖動(dòng)”有關(guān)的推文過(guò)濾出來(lái)。我們可以非常輕易地利用Spark Streaming來(lái)實(shí)現(xiàn)該目標(biāo)：

TwitterUtils.createStream(...) 
    .filter(_.getText.contains("earthquake") || _.getText.contains("shaking")) 

接下來(lái)我們可以對(duì)處理完的推文數(shù)據(jù)做一些語(yǔ)義分析，并判斷是否能反映出當(dāng)前的地震情況。比如，類似于“地震”或者“現(xiàn)在正在晃動(dòng)!”的推文將被視為具有正效應(yīng)，而類似于“參加地震會(huì)議。”或者“昨天的地震真恐怖。”則被視為無(wú)影響效應(yīng)。文章作者利用支持向量機(jī)模型來(lái)實(shí)現(xiàn)該目標(biāo)，我們將在 此基礎(chǔ)上利用流式數(shù)據(jù)版本的模型來(lái)實(shí)現(xiàn)。下文是利用MLlib生成的代碼示例：

// We would prepare some earthquake tweet data and load it in LIBSVM format. 
val data = MLUtils.loadLibSVMFile(sc, "sample_earthquate_tweets.txt") 
 
// Split data into training (60%) and test (40%). 
val splits = data.randomSplit(Array(0.6,0.4),seed = 11L) 
val training = split(0).cache() 
val test = splits(1) 
 
// Run training algorithm to build the model 
val numIterations = 100 
val model = SVMWithSGD.train(training, numIterations) 
 
// Clear the default threshold. 
model.clearThreshold() 
 
// Compute raw scores on the test set. 
val scoreAndLabels = test.map{point =>  
    val score = model.predict(point.features) 
    (score, point.label)} 
 
// Get evaluation metrics. 
val metric = new BinaryClassificationMetrics(scoreAndLabels) 
val auROC = metrics.areaUnderROC() 
 
println("Area under ROC =" + auROC) 

如果我們關(guān)注模型的預(yù)測(cè)準(zhǔn)確率，那么我們可以進(jìn)一步對(duì)檢測(cè)到地震做出反應(yīng)。需要注意的是，對(duì)于包含地理信息的推文，我們還可以獲取震源位置。利用這個(gè)信息，我們可以通過(guò)SparkSQL從現(xiàn)有的Hive table(儲(chǔ)存需要接收地震提醒的用戶信息)中提取出他們的郵箱地址并發(fā)送一封私人電子郵件：

//sc is an existing SparkContext. 
val sqlContext = new org.apache.spark.sql.hive.HiveContext(sc) 
// sendEmail is a custom function 
sqlContext.sql("FROM earthquake_warning_users SELECT firstName, lastName, city, email") 
    .collect().foreach(sendEmail) 

Apache Spark的其他應(yīng)用

除檢測(cè)地震情況外，Spark還有許多潛在的應(yīng)用。

以下是Spark在大數(shù)據(jù)中的部分應(yīng)用：

1.在游戲領(lǐng)域中，從實(shí)時(shí)的潛在游戲事件中迅速地挖掘出有價(jià)值的模式可以創(chuàng)造出巨大的商業(yè)利益，比如用戶返回率情況、如何制定定向廣告以及如何自動(dòng)調(diào)整游戲的復(fù)雜度等。

2.在電子商務(wù)領(lǐng)域中，實(shí)時(shí)交易數(shù)據(jù)將被傳遞到k均值算法或者ALS等協(xié)同過(guò)濾流算法中。這些運(yùn)算結(jié)果將和顧客評(píng)論等非結(jié)構(gòu)化數(shù)據(jù)結(jié)合起來(lái)，用于不斷改進(jìn)交易模式以適應(yīng)新趨勢(shì)的發(fā)展。

3.在金融或證券領(lǐng)域中，Spark堆棧技術(shù)可以被應(yīng)用到信用詐騙和風(fēng)險(xiǎn)管控系統(tǒng)中。通過(guò)獲取大量的歷史數(shù)據(jù)和其他一些外泄數(shù)據(jù)以及一些連接/請(qǐng)求信息(IP地理信息或時(shí)間信息)，我們可以取得非常好的模型結(jié)果。

結(jié)論

總而言之，Spark幫助人們簡(jiǎn)化了處理大規(guī)模數(shù)據(jù)的步驟流程。不管是處理結(jié)構(gòu)化還是非結(jié)構(gòu)化數(shù)據(jù)，Spark將許多復(fù)雜的功能(比如機(jī)器學(xué)習(xí)算法和圖算法)無(wú)縫地結(jié)合起來(lái)。Spark使得大量的從業(yè)者都可以進(jìn)行大數(shù)據(jù)分析，讓我們一探究竟吧!

原文作者：RADEK OSTROWSKI

譯者：Fibears