2.1資料科學家的Scala

spark是用scala語言編寫的，使用scala語言進行大資料開發的好處有

1、效能開銷小

減少不同環境下傳遞程式碼和資料的錯誤和效能開銷

2、能用上最新的版本和最好的功能

Spark的新功能毫無疑問是首先適配scala語言，但使用spark的所有功能可不是那麼容易

3、有助於瞭解spark的原理

2.2小試牛刀

接下來是通過一個數據清理的例子來熟悉scala和spark，可以從http://bit.ly/1Aoywaq下載本節的資料集（需要翻牆下載），推薦一個VPN網址一小時VPN。

資料的預覽

第一步是讀取資料

val conf = new SparkConf().setAppName("WordCount").setMaster("local") //先生成RDD的配置物件conf，配置名稱WordCount，執行在本地模式
val sc = new SparkContext(conf)                                                                           
val rawblock1 = sc.textFile("/home/sam/下載/linkage/donation/block_1.csv")//讀取一個表格的資料到RDD
val rawblock2 = sc.textFile("/home/sam/下載/linkage/donation/block_2.csv")
val raws = rawblock1.union(rawblock2)                                                                 //合併RDD
 

然後去除表格的頭部

def  isHeader(line:String) = line.contains("id_1")            //包含字元“id_1”行就是表格頭部
val noheader = raws.filter(!isHeader(_))                           //RDD是隻讀的，把不包含頭部的資料篩選出來生成一個新的RDD

資料轉換異常資料處理

為了讓資料更方便使用，重新整理資料的結構，把資料分為4部分

case class MatchData(id1: Int, id2: Int,scores: Array[Double], matched: Boolean)
 

定義一個類，再通過下面的parse()方法把每行資料整合到MatchData類裡面

然後把字串資料轉換為資料本來的型別。

id_1	int
id_2	int
cmp_fname_c1到cmp_plz	double
is_match	boolean

可以看到表格中有很多？，代表著資料缺失，在資料型別轉換的同時需要把？轉為NAN

def toDouble(s: String) = {if ("?".equals(s)) Double.NaN else s.toDouble}      

定義把每行資料轉重新結構化為MatchData類的方法

def parse(line: String) = {
      val pieces = line.split(',')
      val id1 = pieces(0).toInt
      val id2 = pieces(1).toInt
      val scores = pieces.slice(2, 11).map(toDouble)
      val matched = pieces(11).toBoolean
      MatchData(id1, id2, scores, matched)
    }
 

把noheader   RDD轉為結構化RDD

val parsed = noheader.map(line => parse(line))

可以把結構化的資料分組，按照is_match的值進行分組

val grouped = parsed.groupBy(_.matched)

把每個分組的資料個數打印出來

grouped.mapValues(_.size).foreach(println)

關閉spark

sc.stop()