前言

統(tǒng)計(jì)科學(xué)家使用交互式的統(tǒng)計(jì)工具(比如R)來回答數(shù)據(jù)中的問題，獲得全景的認(rèn)識(shí)。與之相比，數(shù)據(jù)工程師則更像一名程序員，他們?cè)诜?wù)器上編寫代碼，創(chuàng)建和應(yīng)用機(jī)器學(xué)習(xí)模型，熟悉C++和Java等系統(tǒng)級(jí)語言，經(jīng)常需要和企業(yè)級(jí)數(shù)據(jù)中心的某些組件打交道,比如Hadoop。

而有的數(shù)據(jù)科學(xué)家專注于更細(xì)的領(lǐng)域，就像精通R但從未聽說過Python或者scikit-learn(反之亦然)，即便兩者都提供了豐富的統(tǒng)計(jì)庫。(文末附有驚喜!)

Spark相比其他工具

如果可以提供一種統(tǒng)一的工具，運(yùn)行在統(tǒng)一的架構(gòu)，用統(tǒng)一的語言編程，并可以同時(shí)滿足統(tǒng)計(jì)科學(xué)家和數(shù)據(jù)工程師的需求，那該多好啊。難道為了研究數(shù)據(jù)，我就必須去學(xué)一種像Python或R的語言?我一直使用傳統(tǒng)的數(shù)據(jù)分析工具，難道為了應(yīng)對(duì)大規(guī)模計(jì)算，就必須去懂MapReduce?正是統(tǒng)計(jì)工具的不完美造就了這種局面：

• R提供了一個(gè)豐富的統(tǒng)計(jì)分析和機(jī)器學(xué)習(xí)的解釋器。但R難以在分布式條件下執(zhí)行數(shù)據(jù)的分析和清洗，以便開展其所擅長的數(shù)據(jù)分析，也不以一種主流的開發(fā)語言為人所知。
• Python是-種通用的編程語言，也不乏出色的第三方數(shù)據(jù)分析庫(像Pandas和scikit-learn)，但Python也有和R一樣的缺陷:只能局限在處理單機(jī)能負(fù)載的數(shù)據(jù)量。
• 在經(jīng)典的MapReduce計(jì)算框架上開發(fā)分布式的機(jī)器學(xué)習(xí)算法是可行的(參考Mahout)，但程序員需要從零開始，更別說移植復(fù)雜計(jì)算的難度。
• 為降低復(fù)雜計(jì)算移植到MapReduce的難度，Crunch 提供一個(gè)簡(jiǎn)單的、傻瓜式的Java API,但MapReduce天生決定了它在迭代計(jì)算方面是低效的，盡管大多數(shù)機(jī)器學(xué)習(xí)算法都需要迭代計(jì)算。 

Spark的優(yōu)勢(shì)

Spark是一個(gè)超有潛力的通用數(shù)據(jù)計(jì)算平臺(tái)，無論是對(duì)統(tǒng)計(jì)科學(xué)家還是數(shù)據(jù)工程師。大部分人討論到Spark時(shí)，總是注意到將數(shù)據(jù)駐留內(nèi)存以提高計(jì)算效率的方面(相對(duì)MapReduce)，Spark 擁有許多的特征，使之真正成為一個(gè)融合統(tǒng)計(jì)科學(xué)和數(shù)據(jù)工程的交叉點(diǎn)：

• Spark附帶了一個(gè)機(jī)器學(xué)習(xí)庫MLib，雖然只是在初始階段。
• Spark是用Scala語言編寫的，運(yùn)行在Java虛擬機(jī)上，同時(shí)也提供像R和Python的命令行解釋器。
• 對(duì)Java程序員，Scala 的學(xué)習(xí)曲線是比較陡峭的，但所幸Scala可以兼容一切的Java庫。
• Spark的RDD(彈性分布式數(shù)據(jù)集)，是Crunch開發(fā)者熟知的--種數(shù)據(jù)結(jié)構(gòu)。
• Spark模仿了Scala 的集合計(jì)算API，對(duì)Java和Scala開發(fā)者來說耳熟能詳，而Python開發(fā)者也不難上手，而Scala對(duì)統(tǒng)計(jì)計(jì)算的支持也不錯(cuò)。
• Spark和其底層的Scala語言，并不只是為機(jī)器學(xué)習(xí)而誕生的，除此之外，像數(shù)據(jù)訪問、日志ETL和整合都可以通過API輕松搞定。就像Python,你可以把整個(gè)數(shù)據(jù)計(jì)算流程搬到Spark平臺(tái).上來，而不僅僅是模型擬合和分析。

在命令行解釋器中執(zhí)行的代碼，和編譯后運(yùn)行的效果相同。而且，命令行的輸入可以得到實(shí)時(shí)反饋，你將看到數(shù)據(jù)透明地在集群間傳遞與計(jì)算。

Spark和MLib還有待完善整個(gè)項(xiàng)目有不少bug，效率也還有提升的空間，和YARN的整合也存在問題。Spark還沒辦法提供像R那樣豐富的數(shù)據(jù)分析函數(shù)。但Spark已然是世界上最好的數(shù)據(jù)平臺(tái)，足以讓來自任何背景的數(shù)據(jù)科學(xué)家側(cè)目。 

Stack Overflow問題的自動(dòng)標(biāo)注

Stack Overflow是一個(gè)著名的軟件技術(shù)問答平臺(tái)，在上面提的每個(gè)問題有可能被打上若干個(gè)短文本的標(biāo)簽，比如java或者sql，我們的目標(biāo)在于建立一.套系統(tǒng)，使用ALS推薦算法，為新問題的標(biāo)簽提供預(yù)測(cè)和建議。從推薦系統(tǒng)的角度，你可以把問題想象成user,把標(biāo)簽想象成item。

首先，從Stack Overflow下載官方提供的截至20140120的問答數(shù)據(jù)

sta ckoverflow. com-Posts. 7z 

這是一個(gè)能夠直接用于分布式計(jì)算的bzip格式文件，但在我們的場(chǎng)景下，必須先解壓并拷

貝到HDFS

bzcat stackover flow. com-Posts.7z| hdfs dfs -put一/user /srowen/ Posts. xml 

解壓后的文件大約是24.4GB，包含210萬個(gè)問題，1800 萬個(gè)回答，總共標(biāo)注了930萬個(gè)標(biāo)簽，這些標(biāo)簽排重之后大概是34000個(gè)。

確認(rèn)機(jī)器安裝了Spark 之后，輸入spark-shell即可打開Scala的REPL環(huán)境。首先,我們讀取一個(gè)存儲(chǔ)在HDFS的Posts. xm文件：

val postsXML = sC. textFile("hdfs:/ //user /srowen/Posts. xml") 

這時(shí)命令行工具會(huì)返回：

postsXML: org. apache. spark.rdd.RDD[String] = MappedRDD[1] at textFile at :12 

顯示文本文件已轉(zhuǎn)化為一個(gè)String型的RDD，你可以通過調(diào)用RDD的函數(shù)，實(shí)現(xiàn)任意的查詢運(yùn)算。比如統(tǒng)計(jì)文件的行數(shù)：

postsXML. count 

這條指令生成大量的輸出，顯示Spark正在利用分布式的環(huán)境計(jì)數(shù)，最終打印出18066983。

下一步，將XML文件的每- -行都存入形如(questionID, tag)的元組。得益于Scala的函數(shù)式編程的風(fēng)格，RDD和Scala集合-樣可以使用map等方法： 

val postIDTags = postsXML. flatMap { line =>  
// Matches Id=".. ."  
Tags="..." in line  
val idTagRegex = "Id=\"(\\d+)\". +Tags=\"([^\"]+)\"".r  
// // Finds tags like <TAG> value from above  
val tagRegex = "<([^&]+)>".r  
// Yields 0 or 1 matches:  
idTagRegex. findFirstMatchIn(line) match {  
// No match -- not a line  
case None => None  
// Match, and can extract ID and tags from m  
case Some(m) => {  
val postID = m. group(1) . toInt  
val tagsString = m. group(2)  
// Pick out just TAG matching group  
val tags = tagRegex. findAllMatchIn(tagsString)。map(_ . group  
(1)) . toList  
// Keep only question with at least 4 tags, and map to (pos  
t,tag) tuples  
if (tags.size >= 4) tags . map( (postID,_)) else None  
}  
}  
// Because of flatMap，individual lists will concatenate  
// into one collection of tuples  
} 

你會(huì)發(fā)現(xiàn)這條指令的執(zhí)行是立即返回的，而不像count一樣需要等待，因?yàn)榈侥壳盀橹梗琒park并未啟動(dòng)任何主機(jī)間的數(shù)據(jù)變換。

ALS的MLib實(shí)現(xiàn)必須使用數(shù)值ID而非字符串作為惟一標(biāo)識(shí)，而問題的標(biāo)簽數(shù)據(jù)是字符串格式的，所以需要把字符串哈希成一個(gè)非負(fù)整數(shù)，同時(shí)保留非負(fù)整數(shù)到字符串的映射。這里我們先定義一個(gè)哈希函數(shù)以便復(fù)用。 

def nnHash(tag: String) = tag.hashCode & 0x7FFFFF 
var tagHashes = postIDTags .map(_._2) .distinct. map(tag => (nnHash 
(tag) , tag)) 

現(xiàn)在把元組轉(zhuǎn)換為ALS計(jì)算所需的輸入： 

import org. apache. spark. mllib. recommendation._  
// Convert to Rating(Int ,Int,Double) objects  
val alsInput = postIDTags.map(t => Rating(t. _1， nnHash(t._2)， 1.  
0) )  
// Train model with 40 features， 10 iterations of ALS  
val model = ALS. trainImplicit(alsInput, 40，10) 

這一步生成特征矩陣，可以被用來預(yù)測(cè)問題與標(biāo)簽之間的關(guān)聯(lián)。由于目前MLib還處于不完善的狀態(tài)，沒有提供一個(gè)recommend的接口來獲取建議的標(biāo)簽，我們可以簡(jiǎn)單定義一個(gè)： 

def  
recommend (questionID: Int， howMany: Int = 5): Array[(String ，  
Double)] = {  
// Build list of one question and all items and predict value f  
or all of them  
val predictions = model. predict(tagHashes.map(t => (questionI  
D,t._1)))  
// Get top howMany recommendations ordered by prediction value  
val topN = predictions. top ( howMany) (Ordering . by [Rating，Doub1  
e](_.rating))  
// Translate back to tags from IDs  
topN . map(r => (tagHashes. lookup(r . product)(0)，r .rating)) 

通過上述函數(shù)，我們可以獲得任意一個(gè)問題比如ID為7122697的How to make substring-matching query work fast on a large table?的至少4個(gè)標(biāo)簽：

recommend ( 7122697)。foreach(println) 

推薦結(jié)果如下所示： 

(sqL,0.17745152481166354)  
(database,0.13526622226672633)  
(oracle , 0.1079428707621154)  
(ruby-on-rails, 0.06067207312463499)  
(postgresql,0.050933613169706474) 

注意:

• -每次運(yùn)行得到的結(jié)果不盡相同，是因?yàn)锳LS是從隨機(jī)解開始迭代的
• -如果你希望獲得實(shí)時(shí)性更高的結(jié)果，可以在recommend前輸入tagHashes=tagHas hes. cache

真實(shí)的問題標(biāo)簽是postgresql、query-optimi zation、substring 和text-search。不過，預(yù)測(cè)結(jié)果也有一定的合理性(postgresq 經(jīng)常和ruby-on-rails一起出現(xiàn))

當(dāng)然，以上的示例還不夠優(yōu)雅和高效，但是，我希望所有來自R的分析師、鼓搗Python 的黑客和熟悉Hadoop的開發(fā)者，都能從中找到你們熟悉的部分，從而找到一條適合你們的路徑去探索Spark，并從中獲益。