Spark核心編程

RDD抽象通過(guò)語(yǔ)言集成API公開(kāi)。這簡(jiǎn)化了編程的復(fù)雜性，因?yàn)閼?yīng)用程序的處理RDDS方式類似于操縱的本地集合數(shù)據(jù)。

Spark Shell

Spark提供了一個(gè)交互的shell ? 一個(gè)強(qiáng)大的工具，以交互方式分析數(shù)據(jù)。 這是在 Scala或Python語(yǔ)言。Spark主要抽象稱為彈性分布式數(shù)據(jù)集(RDD)項(xiàng)目的分布式采集。RDDS可以從Hadoop的輸入格式來(lái)創(chuàng)建(如HDFS文件)或通過(guò)轉(zhuǎn)化其他RDDS。

打開(kāi) Spark Shell

$ spark-shell

創(chuàng)建簡(jiǎn)單RDD

scala> val inputfile = sc.textFile(“input.txt”)

inputfile: org.apache.spark.rdd.RDD[String] = input.txt MappedRDD[1] at textFile at <console>:12

RDD 轉(zhuǎn)換

RDD轉(zhuǎn)換返回指向新的RDD，并允許創(chuàng)建RDDS之間的依賴關(guān)系。 在依賴關(guān)系鏈中的每個(gè)RDD(依賴關(guān)系的字串)具有這樣的功能，用于計(jì)算其數(shù)據(jù)并具有一個(gè)指針(依賴性)到其父RDD。

Spark是懶惰的，所以什么都不會(huì)被執(zhí)行，除非調(diào)用一些改造或行動(dòng)將觸發(fā)作業(yè)創(chuàng)建和執(zhí)行?？磫卧~計(jì)數(shù)示例，如下面的代碼片段。

動(dòng)作

RDD編程

示例

考慮一個(gè)單詞計(jì)數(shù)的例子 ? 它計(jì)算出現(xiàn)在文檔中的每個(gè)單詞。請(qǐng)看下面的文字為輸入并保存在主目錄中的 input.txt 文件。

input.txt ? 作為輸入文件

people are not as beautiful as they look, 
as they walk or as they talk.
they are only as beautiful  as they love, 
as they care as they share.

打開(kāi)Spark-Shell

下面的命令用來(lái)打開(kāi)spark shell. 通常情況下，spark 使用Scala構(gòu)建。因此，Spark 程序需要在 Scala 環(huán)境中運(yùn)行。

$ spark-shell 

如果Spark shell 成功打開(kāi)，會(huì)發(fā)現(xiàn)下面的輸出?？纯摧敵觥癝park 上下文可作為sc” 的最后一行表示Spark容器會(huì)自動(dòng)創(chuàng)建Spark 上下文對(duì)象名為sc。啟動(dòng)程序的第一步驟之前，SparkContext 對(duì)象應(yīng)該被創(chuàng)建。

Spark assembly has been built with Hive, including Datanucleus jars on classpath 
Using Spark's default log4j profile: org/apache/spark/log4j-defaults.properties 
15/06/04 15:25:22 INFO SecurityManager: Changing view acls to: hadoop 
15/06/04 15:25:22 INFO SecurityManager: Changing modify acls to: hadoop 
15/06/04 15:25:22 INFO SecurityManager: SecurityManager: authentication disabled;
   ui acls disabled; users with view permissions: Set(hadoop); users with modify permissions: Set(hadoop) 
15/06/04 15:25:22 INFO HttpServer: Starting HTTP Server 
15/06/04 15:25:23 INFO Utils: Successfully started service 'HTTP class server' on port 43292. 
Welcome to 
      ____              __ 
     / __/__  ___ _____/ /__ 
    _\ \/ _ \/ _ `/ __/  '_/ 
   /___/ .__/\_,_/_/ /_/\_\   version 1.4.0 
      /_/  
		
Using Scala version 2.10.4 (Java HotSpot(TM) 64-Bit Server VM, Java 1.7.0_71) 
Type in expressions to have them evaluated. 
Spark context available as sc 
scala>

創(chuàng)建一個(gè)RDD

下面的命令被用于從給定位置讀出的文件。這里，新的 RDD 使用輸入文件名創(chuàng)建。這是在 textFile(“”)方法的參數(shù)字符串是用于輸入文件名的絕對(duì)路徑。然而，如果僅給出文件名，那么它輸入文件則在當(dāng)前位置。

scala> val inputfile = sc.textFile("input.txt")

執(zhí)行字?jǐn)?shù)轉(zhuǎn)換

我們的目標(biāo)是計(jì)算一個(gè)文件中的字?jǐn)?shù)。分裂每一行成詞創(chuàng)建一個(gè)平面地圖(flatMap(line ? line.split(“ ”)).

接下來(lái)，讀每個(gè)詞作為一個(gè)鍵和值 ‘1’ (<key, value> = <word,1>) 使用映射函數(shù) (map(word ? (word, 1)).

最后，加入類似的鍵值降低這些鍵 (reduceByKey(_+_)).

下面的命令用于執(zhí)行字?jǐn)?shù)統(tǒng)計(jì)邏輯。執(zhí)行此操作后，不會(huì)有任何輸出，因?yàn)檫@不是一個(gè)動(dòng)作，這是一個(gè)轉(zhuǎn)換; 指向一個(gè)新的RDD或告訴spark，用給定的數(shù)據(jù)來(lái)做什么)。

scala> val counts = inputfile.flatMap(line => line.split(" ")).map(word => (word, 1)).reduceByKey(_+_);

當(dāng)前RDD

同時(shí)用RDD工作，如果想了解當(dāng)前的RDD，那么可使用下面的命令。 它會(huì)告訴你關(guān)于當(dāng)前RDD及其依賴調(diào)試的描述。

scala> counts.toDebugString

緩存轉(zhuǎn)換

可以使用 persist() 或 cache() 方法標(biāo)記一個(gè)RDD。在第一次計(jì)算的操作，這將被保存在存儲(chǔ)器中的節(jié)點(diǎn)上。使用下面的命令來(lái)存儲(chǔ)中間轉(zhuǎn)換在內(nèi)存中。

scala> counts.cache()

應(yīng)用動(dòng)作

應(yīng)用動(dòng)作(操作)，比如存儲(chǔ)所有的轉(zhuǎn)換結(jié)果到一個(gè)文本文件中。saveAsTextFile(“”)方法字符串參數(shù)是輸出文件夾的絕對(duì)路徑。試試下面的命令來(lái)保存輸出文本文件。在下面的例子中， ‘output’ 的文件夾為當(dāng)前位置。

scala> counts.saveAsTextFile("output")

檢查輸出

打開(kāi)另一個(gè)終端進(jìn)入主目錄(其中spark 在其他終端中執(zhí)行)。下面的命令用于檢查輸出目錄。

[hadoop@localhost ~]$ cd output/ 
[hadoop@localhost output]$ ls -1 
 
part-00000 
part-00001 
_SUCCESS

下面的命令是用來(lái)查看輸出的 Part-00001 文件。

[hadoop@localhost output]$ cat part-00000

輸出

(people,1) 
(are,2) 
(not,1) 
(as,8) 
(beautiful,2) 
(they, 7) 
(look,1) 

[hadoop@localhost output]$ cat part-00001 

輸出

(walk, 1) 
(or, 1) 
(talk, 1) 
(only, 1) 
(love, 1) 
(care, 1) 
(share, 1) 

UN持久存儲(chǔ)

http://localhost:4040

Scala> counts.unpersist() 

將看到如下輸出 ?

15/06/27 00:57:33 INFO ShuffledRDD: Removing RDD 9 from persistence list 
15/06/27 00:57:33 INFO BlockManager: Removing RDD 9 
15/06/27 00:57:33 INFO BlockManager: Removing block rdd_9_1 
15/06/27 00:57:33 INFO MemoryStore: Block rdd_9_1 of size 480 dropped from memory (free 280061810) 
15/06/27 00:57:33 INFO BlockManager: Removing block rdd_9_0 
15/06/27 00:57:33 INFO MemoryStore: Block rdd_9_0 of size 296 dropped from memory (free 280062106) 
res7: cou.type = ShuffledRDD[9] at reduceByKey at <console>:14

http://localhost:4040/