Spark Shell簡單使用

基礎

Spark的shell作為一個強大的交互式數據分析工具，提供了一個簡單的方式學習API。它可以使用Scala(在Java虛擬機上運行現有的Java庫的一個很好方式)或Python。在Spark目錄里使用下面的方式開始運行：

 

./bin/spark-shell

在Spark Shell中，有一個專有的SparkContext已經為您創建好了，變量名叫做sc。自己創建的SparkContext將無法工作。可以用--master參數來設置SparkContext要連接的集群，用--jars來設置需要添加到CLASSPATH的jar包，如果有多個jar包，可以使用逗號分隔符連接它們。例如，在一個擁有4核的環境上運行spark-shell，使用：

./bin/spark-shell --master local[4]

或在CLASSPATH中添加code.jar，使用：

./bin/spark-shell --master local[4] --jars code.jar

可以執行spark-shell --help獲取完整的選項列表。 

Spark最主要的抽象是叫Resilient Distributed Dataset(RDD)的彈性分布式集合。RDDs可以使用Hadoop InputFormats(例如HDFS文件)創建，也可以從其他的RDDs轉換。讓我們在Spark源代碼目錄里從README.md文本文件中創建一個新的RDD。

scala> val textFile = sc.textFile("file:///home/hadoop/hadoop/spark/README.md")
16/07/24 03:30:53 INFO storage.MemoryStore: ensureFreeSpace(217040) called with curMem=321016, maxMem=280248975
16/07/24 03:30:53 INFO storage.MemoryStore: Block broadcast_2 stored as values in memory (estimated size 212.0 KB, free 266.8 MB)
16/07/24 03:30:53 INFO storage.MemoryStore: ensureFreeSpace(20024) called with curMem=538056, maxMem=280248975
16/07/24 03:30:53 INFO storage.MemoryStore: Block broadcast_2_piece0 stored as bytes in memory (estimated size 19.6 KB, free 266.7 MB)
16/07/24 03:30:53 INFO storage.BlockManagerInfo: Added broadcast_2_piece0 in memory on localhost:43303 (size: 19.6 KB, free: 267.2 MB)
16/07/24 03:30:53 INFO spark.SparkContext: Created broadcast 2 from textFile at <console>:21
textFile: org.apache.spark.rdd.RDD[String] = MapPartitionsRDD[5] at textFile at <console>:21

注意：1. 其中2~7行是日志信息，暫且不必關注，主要看最后一行。之后的運行日志信息將不再貼出。用戶也可以進入到spark目錄/conf文件夾下，此時有一個log4j.properties.template文件，我們執行如下命令將其拷貝一份為log4j.properties，並對log4j.properties文件進行修改。

 

 

cp log4j.properties.template log4j.properties
vim log4j.properties

如下圖所示，將INFO改為WARN，這樣就不輸出藍色部分的日志信息：

 

2. 另外，file:///home/hadoop/hadoop/spark/README.md，首部的file代表本地目錄，注意file:后有三個斜杠(/)；中間紅色部分是我的spark安裝目錄，讀者可根據自己的情況進行替換。

 

RDD的actions從RDD中返回值，transformations可以轉換成一個新RDD並返回它的引用。下面展示幾個action：

 

scala> textFile.count()
res0: Long = 98

scala> textFile.first()
res1: String = # Apache Spark

其中，count代表RDD中的總數據條數；first代表RDD中的第一行數據。

 

下面使用一個transformation，我們將使用filter函數對textFile這個RDD進行過濾，取出包含字符串"Spark"的行，並返回一個新的RDD：

 

scala> val linesWithSpark = textFile.filter(line => line.contains("Spark"))
linesWithSpark: org.apache.spark.rdd.RDD[String] = MapPartitionsRDD[2] at filter at <console>:23

當然也可以把actions和transformations連接在一起使用：

 

 

scala> textFile.filter(line => line.contains("Spark")).count()
res2: Long = 19

上面這條語句表示有多少行包括字符串"Spark"。

 

 

更多RDD操作

RDD actions和transformations能被用在更多的復雜計算中。比如想要找到一行中最多的單詞數量：

 

scala> textFile.map(line => line.split(" ").size).reduce((a, b) => if (a > b) a else b)
res3: Int = 14

首先將行映射成一個整型數值產生一個新的RDD。在這個新的RDD上調用reduce找到行中最大的單詞數個數。map和reduce的參數是Scala的函數串(閉包)，並且可以使用任何語言特性或者Scala/Java類庫。例如，我們可以很方便地調用其他的函數聲明。我們使用Math.max()函數讓代碼更容易理解：

 

 

scala> import java.lang.Math
import java.lang.Math

scala> textFile.map(line => line.split(" ").size).reduce((a, b) => Math.max(a, b))
res4: Int = 14

大家都知道，Hadoop流行的一個通用數據流模式是MapReduce。Spark能夠很容易地實現MapReduce：

 

 

scala> val wordCounts = textFile.flatMap(line => line.split(" ")).map(word => (word, 1)).reduceByKey((a, b) => a + b)
wordCounts: org.apache.spark.rdd.RDD[(String, Int)] = ShuffledRDD[8] at reduceByKey at <console>:24

這里，我們結合了flatMap、map和reduceByKey來計算文件里每個單詞出現的數量，它的結果是包含一組(String, Int)鍵值對的RDD。我們可以使用collect操作收集單詞的數量：

 

 

scala> wordCounts.collect()
res5: Array[(String, Int)] = Array((package,1), (For,2), (Programs,1), (processing.,1), (Because,1), (The,1), (cluster.,1), (its,1), ([run,1), (APIs,1), (have,1), (Try,1), (computation,1), (through,1), (several,1), (This,2), ("yarn-cluster",1), (graph,1), (Hive,2), (storage,1), (["Specifying,1), (To,2), (page](http://spark.apache.org/documentation.html),1), (Once,1), (application,1), (prefer,1), (SparkPi,2), (engine,1), (version,1), (file,1), (documentation,,1), (processing,,2), (the,21), (are,1), (systems.,1), (params,1), (not,1), (different,1), (refer,2), (Interactive,2), (given.,1), (if,4), (build,3), (when,1), (be,2), (Tests,1), (Apache,1), (all,1), (./bin/run-example,2), (programs,,1), (including,3), (Spark.,1), (package.,1), (1000).count(),1), (Versions,1), (HDFS,1), (Data.,1), (>...

 

 

緩存

Spark支持把數據集緩存到內存之中，當要重復訪問時，這是非常有用的。舉一個簡單的例子：

scala> linesWithSpark.cache()
res6: linesWithSpark.type = MapPartitionsRDD[2] at filter at <console>:23

scala> linesWithSpark.count()
res7: Long = 19

scala> linesWithSpark.count()
res8: Long = 19

scala> linesWithSpark.count()
res9: Long = 19

首先緩存linesWithSpark數據集，然后重復訪問count函數返回的值。當然，我們並不能察覺明顯的查詢速度變化，但是當在大型的數據集中使用緩存，將會非常顯著的提升相應的迭代操作速度。