Spark随谈
Spark隨談（一）---總體架構(gòu)
Spark是一個(gè)小巧玲瓏的項(xiàng)目，由Berkeley大學(xué)的Matei為主的小團(tuán)隊(duì)所開發(fā)。使用的語言是Scala，項(xiàng)目的core部分的代碼只有63個(gè)Scala文件，充分體現(xiàn)了精簡(jiǎn)之美。
系列文章見： Spark隨談 http://www.linuxidc.com/Linux/2013-08/88592.htm
Spark之依賴
（1）Map Reduce模型
作為一個(gè)分布式計(jì)算框架，Spark采用了MapReduce模型。在它身上，Google的Map Reduce和Hadoop的痕跡很重，很明顯，它并非一個(gè)大的創(chuàng)新，而是微創(chuàng)新。在基礎(chǔ)理念不變的前提下，它借鑒，模仿并依賴了先輩，加入了一點(diǎn)改進(jìn)，極大的提升了MapReduce的效率。
使用MapReduce模型解決大數(shù)據(jù)并行計(jì)算的問題，帶來的最大優(yōu)勢(shì)，是它和Hadoop的同屬一家人。因?yàn)橥瑢儆贛apReduce并行編程模型，而不是MPI和OpenMP其它模型，因此，復(fù)雜的算法，只要能夠以Java算法表達(dá)，在Hadoop上運(yùn)行的，就能以Scala算法表達(dá)，在Spark上運(yùn)行，而速度有倍數(shù)的提升。相比之下，在MPI和Hadoop算法之間切換，難度就大多了。
（2）函數(shù)式編程
Spark由Scala寫就，而支持的語言亦是Scala。其原因之一就是Scala支持函數(shù)式編程。這一來造就了Spark的代碼簡(jiǎn)潔，二來使得基于Spark開發(fā)的程序，也特別的簡(jiǎn)潔。一次完整的MapReduce，Hadoop中需要?jiǎng)?chuàng)建一個(gè)Mapper類和Reduce類，而Spark只需要?jiǎng)?chuàng)建相應(yīng)的一個(gè)map函數(shù)和reduce函數(shù)即可，代碼量大大降低
（3）Mesos
Spark將分布式運(yùn)行的需要考慮的事情，都交給了Mesos，自己不Care，這也是它代碼能夠精簡(jiǎn)的原因之一。這也算是偷了一個(gè)大懶吧，呵呵
（4）HDFS和S3
Spark支持2種分布式存儲(chǔ)系統(tǒng)：HDFS和S3。應(yīng)該算是目前最主流的兩種了。對(duì)文件系統(tǒng)的讀取和寫入功能是Spark自己提供的，借助Mesos分布式實(shí)現(xiàn)。如果自己想做集群試驗(yàn)，又沒有HDFS環(huán)境，也沒有EC2環(huán)境的話，可以搞個(gè)NFS，確保所有MESOS的Slave都可以訪問，也可以模擬一下。
Spark的術(shù)語
（1）RDD（Resilient distributed datasets ）
彈性分布式數(shù)據(jù)集，Spark中最核心的模塊和類，也是設(shè)計(jì)精華所在。你將它理解為一個(gè)大的集合，將所有數(shù)據(jù)都加載到內(nèi)存中，方便進(jìn)行多次重用。第一，它是分布式的，可以分布在多臺(tái)機(jī)器上，進(jìn)行計(jì)算。第二，它是彈性的，在計(jì)算處理過程中，機(jī)器的內(nèi)存不夠時(shí)，它會(huì)和硬盤進(jìn)行數(shù)據(jù)交換，某種程度上會(huì)減低性能，但是可以確保計(jì)算得以繼續(xù)進(jìn)行。關(guān)于RDD的詳細(xì)闡述，后面會(huì)單獨(dú)再起一篇文章。
（2）Local模式和Mesos模式
Spark支持Local調(diào)用和Mesos集群兩種模式，在Spark上開發(fā)算法程序，可以在本地模式調(diào)試成功后，直接改用Mesos集群運(yùn)行，除了文件的保存位置需要考慮以外，算法理論上不需要做任何修改。
Spark的本地模式支持多線程，有一定的單機(jī)并發(fā)處理能力。但是不算很強(qiáng)勁。本地模式可以保存結(jié)果在本地或者分布式文件系統(tǒng)，而Mesos模式一定需要保存在分布式或者共享文件系統(tǒng)。
（3）Transformations和Actions
對(duì)于RDD，有兩種類型的動(dòng)作，一種是Transformation，一種是Action。它們本質(zhì)區(qū)別是：
- Transformation返回值還是一個(gè)RDD。它使用了鏈?zhǔn)秸{(diào)用的設(shè)計(jì)模式，對(duì)一個(gè)RDD進(jìn)行計(jì)算后，變換成另外一個(gè)RDD，然后這個(gè)RDD又可以進(jìn)行另外一次轉(zhuǎn)換。這個(gè)過程是分布式的
- Action返回值不是一個(gè)RDD。它要么是一個(gè)Scala的普通集合，要么是一個(gè)值，要么是空，最終或返回到Driver程序，或把RDD寫入到文件系統(tǒng)中
關(guān)于這兩個(gè)動(dòng)作，在Spark開發(fā)指南中會(huì)有就進(jìn)一步的詳細(xì)介紹，它們是基于Spark開發(fā)的核心。這里將Spark的官方ppt中的一張圖略作改造，闡明一下兩種動(dòng)作的區(qū)別。
Spark On Mesos
為了在Mesos框架上運(yùn)行，安裝Mesos的規(guī)范和設(shè)計(jì)，Spark實(shí)現(xiàn)兩個(gè)類，一個(gè)是SparkScheduler，在Spark中類名是MesosScheduler；一個(gè)是SparkExecutor，在Spark中類名是Executor。有了這兩個(gè)類，Spark就可以通過Mesos進(jìn)行分布式的計(jì)算。
Spark會(huì)將RDD和MapReduce函數(shù)，進(jìn)行一次轉(zhuǎn)換，變成標(biāo)準(zhǔn)的Job和一系列的Task。提交給SparkScheduler，SparkScheduler會(huì)把Task提交給Mesos Master，由Master分配給不同的Slave，最終由Slave中的Spark Executor，將分配到的Task一一執(zhí)行，并且返回，組成新的RDD，或者直接寫入到分布式文件系統(tǒng)。
基于Spark的項(xiàng)目
基于Spark的項(xiàng)目有2個(gè)，也是AMP實(shí)驗(yàn)室出品
第一個(gè)是Spark內(nèi)部的Bagel，Pregel on Spark，可以用Spark進(jìn)行圖計(jì)算，這是個(gè)非常有用的小項(xiàng)目。Bagel自帶了一個(gè)例子，實(shí)現(xiàn)了Google的PageRank算法，實(shí)驗(yàn)數(shù)據(jù)在http://download.freebase.com/wex/，感興趣的可以下載試試。
第二個(gè)是Shark，Hive On Spark，將Hive的語法遷移到Spark上，將SQL翻譯為Spark的mapreduce運(yùn)行，并且可以直接讀取hive的元數(shù)據(jù)庫(kù)和對(duì)應(yīng)數(shù)據(jù)。這個(gè)項(xiàng)目還在獲得了2012的SIGMOD大會(huì)上獲得最佳Demo獎(jiǎng)，目前已經(jīng)進(jìn)入Alpha版本，很快將會(huì)有正式版本發(fā)布，希望它能夠是一個(gè)完全兼容現(xiàn)有hive，速度快10x倍的牛B產(chǎn)品。
Spark隨談（二）—— 安裝攻略
本來安裝這件事情，不用單獨(dú)開一篇談的。但是Spark的安裝實(shí)在是一件點(diǎn)蛋疼的事情，這和Spark的語言和框架兩者有頗大的關(guān)系。
Spark是Scala語言寫的，所以要先安裝Java和Scala，而底層的調(diào)度框架是Mesos，Mesos是C++寫的，所以又對(duì)機(jī)器的glibc和gcc環(huán)境有一定的要求。裝好了Mesos和Spark，還要把2者銜接起來，版本要選擇正確，這幾個(gè)步驟，中間任何一步錯(cuò)了都Spark都不能以集群方式正常運(yùn)行，所以Spark的安裝，還是頗有點(diǎn)周折的，在這里把完整的過程記錄下來，包括mesos 0.9的安裝過程，希望后來者盡量不需要掉坑里。
本攻略基于版本是Spark 0.5 和Mesos 0.9，服務(wù)器是RedHat Enterprise 6.1 32位，其它服務(wù)器的命令可能會(huì)稍微有所區(qū)別
系列文章見： Spark隨談 http://www.linuxidc.com/Linux/2013-08/88592.htm
1、安裝Spark
1.1 安裝Java
推薦版本是JDK1.6.0 u18,具體下載安裝過程就不說了，最后一定要設(shè)定JAVA_HOME，這個(gè)是后面步驟，尤其是mesos安裝必須
export JAVA_HOME=/usr/java/jdk export PATH=$JAVA_HOME/bin:$PATH1.2 安裝Scala
wget http://www.scala-lang.org/downloads/distrib/files/scala-2.9.2.tgz tar xvf scala-2.9.2.tgz mkdir /usr/share/scala cp -r scala-2.9.2/* /usr/share/scala export SCALA_HOME=/usr/share/scala export PATH=$PATH:$SCALA_HOME/bin/1.3 安裝Spark
wget -O mesos-spark-v0.5.0-0.tar.gz https://github.com/mesos/spark/tarball/v0.5.0 tar -xzvf mesos-spark-v0.5.0-0.tar.gz mv mesos-spark-0472cf8 spark cd spark sbt/sbt compile至此，Spark的基本安裝已經(jīng)完畢，可以嘗試用本地模式運(yùn)行
./run spark.examples.SparkPi local看到正確的Pi結(jié)果，表示Spark安裝第一步完成，本地模式運(yùn)行OK
2、安裝Mesos
Mesos 0.9安裝，必須具備下列條件：
glibc 2.9（必須2.9以上）
gcc-c++ 4.1
python 2.6
python-devel
cppunit-devel
libtool
Redhat 6上述條件基本上已經(jīng)具備了，Redhat 5的話，glibc有可能低于2.5，必須升級(jí)，才能完成mesos的編譯安裝，否則就別折騰了，洗洗睡吧 ：）
wget http://people.apache.org/~benh/mesos-0.9.0-incubating-RC3/mesos-0.9.0-incubating.tar.gz tar zxvf mesos-0.9.0-incubating.tar.gz cd mesos-0.9.0 mkdir build cd build ../configure --with-python-headers=/usr/include/python2.6 --with-java-home=$JAVA_HOME --with-java-headers=$JAVA_HOME/include --with-webui --with-included-zookeeper --prefix=/usr/local/mesos make make install祈禱吧，一切順利的話，mesos就會(huì)被安裝到/usr/local/mesos下,最后關(guān)鍵一步，設(shè)置MESOS_HOME
export MESOS_HOME=/usr/local/mesos3、啟動(dòng)Mesos
手工模式啟動(dòng)：
3.1 啟動(dòng)Master
cd /usr/local/mesos
(sbin/mesos-master –log_dir=/usr/local/mesos/logs & ) &
出現(xiàn)下面的提示Master就成功
Starting Mesos master
Master started on ***:5050
Master ID: ***
Elected as master!
Loading webui script at ‘/usr/local/new_mesos/share/mesos/webui/master/webui.py’
Bottle server starting up (using WSGIRefServer())…
Listening on http://0.0.0.0:8080/
Use Ctrl-C to quit.
3.2 啟動(dòng)Slave
(sbin/mesos-slave -m 127.0.0.1:5050 –log_dir=/home/andy/mesos/logs –work_dir=/home/andy/mesos/works & ) &
使用–resources=”mem:20240;cpus:10″參數(shù)，可以根據(jù)具體的機(jī)器情況，指定分配的資源
Starting Mesos slave
Slave started on ***:42584
Slave resources: cpus=16; mem=23123
New master detected at master@***:5050
Registered with master; given slave ID ***
Loading webui script at ‘/usr/local/new_mesos/share/mesos/webui/slave/webui.py’
Bottle server starting up (using WSGIRefServer())…
Listening on http://0.0.0.0:8081/
Use Ctrl-C to quit.
4、啟動(dòng)Spark On Mesos
好了，終于來到最關(guān)鍵的一步了，在Mesos上運(yùn)行Spark，要把Spark和Mesos連接到一起了。Spark是披著Scala外衣的Java，Mesos是C++，他們的通道，不可避免的就是JNI
配置的關(guān)鍵是Spark的配置文件，Spark帶了樣例文件conf/spark-env.sh.template，并有詳細(xì)的解釋，根據(jù)我們之前的安裝路徑，參考該文件，配置如下：
#保持與系統(tǒng)的MESOS_HOME一致 export MESOS_HOME=/usr/local/mesos/#新版本的配置項(xiàng)，直接指定libmesso.so的位置，該so和spark目錄下的mesos-0.9.0.jar必須一致，是spark和mesos溝通的關(guān)鍵 export MESOS_NATIVE_LIBRARY=/usr/local/mesos/lib/libmesos.so#舊版本的配置項(xiàng)，其它的so，目前看來不需要了 export SPARK_LIBRARY_PATH=/usr/local/mesos/lib#自定義的程序jar包，可以放在該目錄下 export SPARK_CLASSPATH=...#保持與系統(tǒng)的SCALA_HOME一致 export SCALA_HOME=/usr/share/scala#必須小于或者等于Slave中的mem，Slave resources: cpus=16; mem=23123 #本地模式下，運(yùn)行大任務(wù)也需要修改該參數(shù)，默認(rèn)是512m，很小 export SPARK_MEM=10g好了，一切就緒之后，嘗試運(yùn)行下面的命令：
cd spark ./run spark.examples.SparkPi 127.0.0.1:5050 (注意，和以前的mesos版本不一樣，不需要打master@127.0.0.1:5050，否則mesos會(huì)報(bào)錯(cuò)的）如果你再次成功的看到Pi值，恭喜，Spark的安裝又成功了一步。
Spark隨談——開發(fā)指南（譯）
本文翻譯自官方博客，略有添加：https://github.com/mesos/spark/wiki/Spark-Programming-Guide，謝謝師允tx的校正。希望能夠給希望嘗試Spark的朋友，帶來一些幫助。目前的版本是0.5.0
系列文章見： Spark隨談 http://www.linuxidc.com/Linux/2013-08/88592.htm
Spark開發(fā)指南?
從高的層面來看，其實(shí)每一個(gè)Spark的應(yīng)用，都是一個(gè)Driver類，通過運(yùn)行用戶定義的main函數(shù)，在集群上執(zhí)行各種并發(fā)操作和計(jì)算
Spark提供的最主要的抽象，是一個(gè)彈性分布式數(shù)據(jù)集(RDD)，它是一種特殊集合，可以分布在集群的節(jié)點(diǎn)上，以函數(shù)式編程操作集合的方式，進(jìn)行各種各樣的并發(fā)操作。它可以由hdfs上的一個(gè)文件創(chuàng)建而來，或者是Driver程序中，從一個(gè)已經(jīng)存在的集合轉(zhuǎn)換而來。用戶可以將數(shù)據(jù)集緩存在內(nèi)存中，讓它被有效的重用，進(jìn)行并發(fā)操作。最后，分布式數(shù)據(jù)集可以自動(dòng)的從結(jié)點(diǎn)失敗中恢復(fù)，再次進(jìn)行計(jì)算。
Spark的第二個(gè)抽象，是并行計(jì)算中使用的共享變量。默認(rèn)來說，當(dāng)Spark并發(fā)運(yùn)行一個(gè)函數(shù)時(shí)，它是以多個(gè)的task，在不同的結(jié)點(diǎn)上運(yùn)行，它傳遞每一個(gè)變量的一個(gè)拷貝，到每一個(gè)獨(dú)立task使用到的函數(shù)中，因此這些變量并非共享的。然而有時(shí)候，我們需要在任務(wù)中能夠被共享的變量，或者在任務(wù)與驅(qū)動(dòng)程序之間共享。Spark支持兩種類型的共享變量：
廣播變量： 可以在內(nèi)存的所有結(jié)點(diǎn)中被訪問，用于緩存變量（只讀）
累加器： 只能用來做加法的變量，例如計(jì)數(shù)和求和
本指南通過一些樣例展示這些特征。讀者最好是熟悉Scala，尤其是閉包的語法。請(qǐng)留意，Spark可以通過Spark-Shell的解釋器進(jìn)行交互式運(yùn)行。你可能會(huì)需要它。
接入Spark
為了寫一個(gè)Spark的應(yīng)用，你需要將Spark和它的依賴，加入到CLASSPATH中。最簡(jiǎn)單的方法，就是運(yùn)行sbt/sbt assembly來編譯Spark和它的依賴，打到一個(gè)Jar里面core/target/scala_2.9.1/spark-core-assembly-0.0.0.jar，然后將它加入到你的CLASSPATH中。或者你可以選擇將spark發(fā)布到maven的本地緩存中，使用sbt/sbt publish。它將在組織org.spark-project下成為一個(gè)spark-core.
另外，你會(huì)需要導(dǎo)入一些Spark的類和隱式轉(zhuǎn)換， 將下面幾行加入到你程序的頂部
import spark.SparkContext
import SparkContext._
初始化Spark
寫Spark程序需要做的第一件事情，就是創(chuàng)建一個(gè)SparkContext對(duì)象，它將告訴Spark如何訪問一個(gè)集群。這個(gè)通常是通過下面的構(gòu)造器來實(shí)現(xiàn)的：
new SparkContext(master, jobName, [sparkHome], [jars])
Master參數(shù)是一個(gè)字符串，指定了連接的Mesos集群，或者用特殊的字符串“l(fā)ocal”來指明用local模式運(yùn)行。如下面的描述一般，JobName是你任務(wù)的名稱，當(dāng)在集群上運(yùn)行的時(shí)候，將會(huì)在Mesos的Web UI監(jiān)控界面顯示。后面的兩個(gè)參數(shù)，是用在將你的代碼，部署到mesos集群上運(yùn)行時(shí)使用的，后面會(huì)提到。
在Spark的解釋器中，一個(gè)特殊的SparkContext變量已經(jīng)為你創(chuàng)建，變量名字叫sc。創(chuàng)建你自己的SparkContext是不會(huì)生效的。你可以通過設(shè)置MASTER環(huán)境變量，來讓master連接到需要的上下文。
MASTER=local; ./spark-shell
Master的命名
Master的名字可以是以下3個(gè)格式中的一種
| Master Name | Meaning |
| local | 本地化運(yùn)行Spark，使用一個(gè)Worker線程（沒有并行） ? |
| local[K] | 本地化運(yùn)行Spark，使用K個(gè)Worker線程（根據(jù)機(jī)器的CPU核數(shù)設(shè)定） ? |
| HOST:PORT | 將Spark連接到指定的Mesos Master，在集群上運(yùn)行。Host參數(shù)是Mesos Master的Hostname， 端口是master配置的端口，默認(rèn)為5050. 注意：在早期的Mesos版本（spark的old-mesos分支），你必須使用master@HOST:PORT. |
集群部署
如果你想你的任務(wù)運(yùn)行在一個(gè)集群上，你需要指定2個(gè)可選參數(shù)：
- SparkHome：Spark在集群機(jī)器上的安裝路徑（必須全部一致）
- Jars：在本地機(jī)器上，包含了你任務(wù)的代碼和依賴的Jars文件列表。 Spark會(huì)把它們部署到所有的集群結(jié)點(diǎn)上。 你需要使用自己的編譯系統(tǒng)將你的作業(yè)，打包成一套jars文件。例如，如果你使用sbt，那么sbt-assembly插件是一個(gè)好方法，將你的代碼和依賴，變成一個(gè)單一的jar文件。
如果有一些類庫(kù)是公用的，需要在不同的作業(yè)間共享，你可能需要手工拷貝到mesos的結(jié)點(diǎn)上，在conf/spark-env中，通過設(shè)置SPARK_CLASSPATH環(huán)境變量指向它們。詳細(xì)信息可以參考配置
分布式數(shù)據(jù)集
Spark圍繞的核心概念，是彈性分布式數(shù)據(jù)集（RDD），一個(gè)有容錯(cuò)機(jī)制，可以被并行操作的集合。目前有兩種類型的RDD： 并行集合(Parrallelized Collections)，接收一個(gè)已經(jīng)存在的Scala集合，在它上面運(yùn)行各種并發(fā)計(jì)算； Hadoop數(shù)據(jù)集（Hadoop DataSets），在一個(gè)文件的每條記錄上，運(yùn)行各種函數(shù)。只要文件系統(tǒng)是Hdfs，或者h(yuǎn)adoop支持的任意存儲(chǔ)系統(tǒng)。這兩種RDD都可以通過相同的方式進(jìn)行操作。
并行集合
并行集合是通過調(diào)用SparkContext的parallelize方法，在一個(gè)已經(jīng)存在的Scala集合（只要是seq對(duì)象就可以）上創(chuàng)建而來。集合的對(duì)象將會(huì)被拷貝來創(chuàng)建一個(gè)分布式數(shù)據(jù)集，可以被并行操作。下面通過spark解釋器的例子，展示如何從一個(gè)數(shù)組創(chuàng)建一個(gè)并發(fā)集合
scala> val data = Array(1, 2, 3, 4, 5)
data: Array[Int] = Array(1, 2, 3, 4, 5)
scala> val distData = sc.parallelize(data)
distData: spark.RDD[Int] = spark.ParallelCollection@10d13e3e
一旦被創(chuàng)建，分布數(shù)據(jù)集（distData）可以被并行操作。例如，我們可以調(diào)用distData.reduce(_ +＿) 來將數(shù)組的元素相加。我們會(huì)在后續(xù)的分布式數(shù)據(jù)集做進(jìn)一步描述。
創(chuàng)建并行集合的一個(gè)重要參數(shù)，是slices的數(shù)目，它指定了將數(shù)據(jù)集切分為幾份。在集群模式中，Spark將會(huì)在一份slice上起一個(gè)Task。典型的，你可以在集群中的每個(gè)cpu上，起2-4個(gè)Slice （也就是每個(gè)cpu分配2-4個(gè)Task）。一般來說，Spark會(huì)嘗試根據(jù)集群的狀況，來自動(dòng)設(shè)定slices的數(shù)目。然而，你也可以手動(dòng)的設(shè)置它，通過parallelize方法的第二個(gè)參數(shù)（例如：sc.parallelize(data, 10)).
Hadoop數(shù)據(jù)集
Spark可以創(chuàng)建分布式數(shù)據(jù)集，從任何存儲(chǔ)在HDFS文件系統(tǒng)或者Hadoop支持的其它文件系統(tǒng)（包括本地文件，Amazon S3， Hypertable， HBase等等）上的文件。 Spark可以支持Text File, SequenceFiles 及其它任何Hadoop輸入格式
文本文件的RDDs可以通過SparkContext的textFile方法創(chuàng)建，該方法接受文件的URI地址（或者機(jī)器上的文件本地路徑，或者一個(gè)hdfs://, sdn://,kfs://,其它URI).這里是一個(gè)調(diào)用例子：
scala> val distFile = sc.textFile(“data.txt”)
distFile: spark.RDD[String] = spark.HadoopRDD@1d4cee08
一旦被創(chuàng)建，distFile可以進(jìn)行數(shù)據(jù)集操作。例如，我們可以使用如下的map和reduce操作將所有行數(shù)的長(zhǎng)度相加：
distFile.map(_.size).reduce(_ + _ )
方法也接受可選的第二參數(shù)，來控制文件的分片數(shù)目。默認(rèn)來說，Spark為每一塊文件創(chuàng)建一個(gè)分片（HDFS默認(rèn)的塊大小為64MB)，但是你可以通過傳入一個(gè)更大的值來指定更多的分片。注意，你不能指定一個(gè)比塊個(gè)數(shù)更少的片值（和hadoop中，Map數(shù)不能小于Block數(shù)一樣）
對(duì)于SequenceFiles，使用SparkContext的sequenceFile[K, V]方法，K和V是文件中的key和values類型。他們必須是Hadoop的Writable的子類，例如IntWritable和Text。另外，Spark允許你指定幾種原生的通用Writable類型，例如：sequencFile[Int, String]會(huì)自動(dòng)讀取IntWritable和Texts
最后，對(duì)于其他類型的Hadoop輸入格式，你可以使用SparkContext.hadoopRDD方法，它可以接收任意類型的JobConf和輸入格式類，鍵類型和值類型。按照對(duì)Hadoop作業(yè)一樣的方法，來設(shè)置輸入源就可以了。
分布式數(shù)據(jù)集操作
分布式數(shù)據(jù)集支持兩種操作：
轉(zhuǎn)換（transformation）:根據(jù)現(xiàn)有的數(shù)據(jù)集創(chuàng)建一個(gè)新的數(shù)據(jù)集
動(dòng)作（actions）:在數(shù)據(jù)集上運(yùn)行計(jì)算后，返回一個(gè)值給驅(qū)動(dòng)程序
例如，Map是一個(gè)轉(zhuǎn)換，將數(shù)據(jù)集的每一個(gè)元素，都經(jīng)過一個(gè)函數(shù)進(jìn)行計(jì)算后，返回一個(gè)新的分布式數(shù)據(jù)集作為結(jié)果。而另一方面，Reduce是一個(gè)動(dòng)作，將數(shù)據(jù)集的所有元素，用某個(gè)函數(shù)進(jìn)行聚合，然后將最終結(jié)果返回驅(qū)動(dòng)程序，而并行的reduceByKey還是返回一個(gè)分布式數(shù)據(jù)集
所有Spark中的轉(zhuǎn)換都是惰性的，也就是說，并不會(huì)馬上發(fā)生計(jì)算。相反的，它只是記住應(yīng)用到基礎(chǔ)數(shù)據(jù)集上的這些轉(zhuǎn)換（Transformation）。而這些轉(zhuǎn)換（Transformation），只會(huì)在有一個(gè)動(dòng)作（Action）發(fā)生，要求返回結(jié)果給驅(qū)動(dòng)應(yīng)用時(shí)，才真正進(jìn)行計(jì)算。這個(gè)設(shè)計(jì)讓Spark更加有效率的運(yùn)行。例如，我們可以實(shí)現(xiàn)，通過map創(chuàng)建一個(gè)數(shù)據(jù)集，然后再用reduce，而只返回reduce的結(jié)果給driver，而不是整個(gè)大的數(shù)據(jù)集。
spark提供的一個(gè)重要轉(zhuǎn)換操作是Caching。當(dāng)你cache一個(gè)分布式數(shù)據(jù)集時(shí)，每個(gè)節(jié)點(diǎn)會(huì)存儲(chǔ)該數(shù)據(jù)集的所有片，并在內(nèi)存中計(jì)算，并在其它操作中重用。這將會(huì)使得后續(xù)的計(jì)算更加的快速（通常是10倍），緩存是spark中一個(gè)構(gòu)造迭代算法的關(guān)鍵工具，也可以在解釋器中交互使用。
下面的表格列出目前支持的轉(zhuǎn)換和動(dòng)作：
轉(zhuǎn)換（Transformations）
| Transformation | Meaning |
| map(func) ? | 返回一個(gè)新的分布式數(shù)據(jù)集，由每個(gè)原元素經(jīng)過func函數(shù)轉(zhuǎn)換后組成 |
| filter(func) ? | 返回一個(gè)新的數(shù)據(jù)集，由經(jīng)過func函數(shù)后返回值為true的原元素組成 |
| flatMap(func) | 類似于map，但是每一個(gè)輸入元素，會(huì)被映射為0到多個(gè)輸出元素（因此，func函數(shù)的返回值是一個(gè)Seq，而不是單一元素） |
| sample(withReplacement, frac, seed) ? | 根據(jù)給定的隨機(jī)種子seed，隨機(jī)抽樣出數(shù)量為frac的數(shù)據(jù) |
| union(otherDataset) ? | 返回一個(gè)新的數(shù)據(jù)集，由原數(shù)據(jù)集和參數(shù)聯(lián)合而成 |
| groupByKey([numTasks]) ? | 在一個(gè)由（K,V）對(duì)組成的數(shù)據(jù)集上調(diào)用，返回一個(gè)（K，Seq[V])對(duì)的數(shù)據(jù)集。注意：默認(rèn)情況下，使用8個(gè)并行任務(wù)進(jìn)行分組，你可以傳入numTask可選參數(shù)，根據(jù)數(shù)據(jù)量設(shè)置不同數(shù)目的Task （groupByKey和filter結(jié)合，可以實(shí)現(xiàn)類似Hadoop中的Reduce功能） |
| reduceByKey(func, [numTasks]) | 在一個(gè)（K，V)對(duì)的數(shù)據(jù)集上使用，返回一個(gè)（K，V）對(duì)的數(shù)據(jù)集，key相同的值，都被使用指定的reduce函數(shù)聚合到一起。和groupbykey類似，任務(wù)的個(gè)數(shù)是可以通過第二個(gè)可選參數(shù)來配置的。 |
| join(otherDataset, [numTasks]) | 在類型為（K,V)和（K,W)類型的數(shù)據(jù)集上調(diào)用，返回一個(gè)（K,(V,W))對(duì)，每個(gè)key中的所有元素都在一起的數(shù)據(jù)集 |
| groupWith(otherDataset, [numTasks]) | 在類型為（K,V)和(K,W)類型的數(shù)據(jù)集上調(diào)用，返回一個(gè)數(shù)據(jù)集，組成元素為（K, Seq[V], Seq[W]) Tuples。這個(gè)操作在其它框架，稱為CoGroup |
| cartesian(otherDataset) | 笛卡爾積。但在數(shù)據(jù)集T和U上調(diào)用時(shí)，返回一個(gè)(T，U）對(duì)的數(shù)據(jù)集，所有元素交互進(jìn)行笛卡爾積。 |
| sortByKey([ascendingOrder]) | 在類型為( K, V )的數(shù)據(jù)集上調(diào)用，返回以K為鍵進(jìn)行排序的（K，V）對(duì)數(shù)據(jù)集。升序或者降序由boolean型的ascendingOrder參數(shù)決定 （類似于Hadoop的Map-Reduce中間階段的Sort，按Key進(jìn)行排序） |
Actions（動(dòng)作）
| Action | Meaning |
| reduce(func) | 通過函數(shù)func聚集數(shù)據(jù)集中的所有元素。Func函數(shù)接受2個(gè)參數(shù)，返回一個(gè)值。這個(gè)函數(shù)必須是關(guān)聯(lián)性的，確保可以被正確的并發(fā)執(zhí)行 |
| collect() | 在Driver的程序中，以數(shù)組的形式，返回?cái)?shù)據(jù)集的所有元素。這通常會(huì)在使用filter或者其它操作后，返回一個(gè)足夠小的數(shù)據(jù)子集再使用，直接將整個(gè)RDD集Collect返回，很可能會(huì)讓Driver程序OOM |
| count() | 返回?cái)?shù)據(jù)集的元素個(gè)數(shù) |
| take(n) | 返回一個(gè)數(shù)組，由數(shù)據(jù)集的前n個(gè)元素組成。注意，這個(gè)操作目前并非在多個(gè)節(jié)點(diǎn)上，并行執(zhí)行，而是Driver程序所在機(jī)器，單機(jī)計(jì)算所有的元素 (Gateway的內(nèi)存壓力會(huì)增大，需要謹(jǐn)慎使用） |
| first() | 返回?cái)?shù)據(jù)集的第一個(gè)元素（類似于take（1）） |
| saveAsTextFile(path) | 將數(shù)據(jù)集的元素，以textfile的形式，保存到本地文件系統(tǒng)，hdfs或者任何其它hadoop支持的文件系統(tǒng)。Spark將會(huì)調(diào)用每個(gè)元素的toString方法，并將它轉(zhuǎn)換為文件中的一行文本 |
| saveAsSequenceFile(path) | 將數(shù)據(jù)集的元素，以sequencefile的格式，保存到指定的目錄下，本地系統(tǒng)，hdfs或者任何其它hadoop支持的文件系統(tǒng)。RDD的元素必須由key-value對(duì)組成，并都實(shí)現(xiàn)了Hadoop的Writable接口，或隱式可以轉(zhuǎn)換為Writable（Spark包括了基本類型的轉(zhuǎn)換，例如Int，Double，String等等） |
| foreach(func) | 在數(shù)據(jù)集的每一個(gè)元素上，運(yùn)行函數(shù)func。這通常用于更新一個(gè)累加器變量，或者和外部存儲(chǔ)系統(tǒng)做交互 |
緩存
調(diào)用RDD的cache()方法，可以讓它在第一次計(jì)算后，將結(jié)果保持存儲(chǔ)在內(nèi)存。數(shù)據(jù)集的不同部分，將會(huì)被存儲(chǔ)在計(jì)算它的不同的集群節(jié)點(diǎn)上，讓后續(xù)的數(shù)據(jù)集使用更快。緩存是有容錯(cuò)功能的，如果任一分區(qū)的RDD數(shù)據(jù)丟失了，它會(huì)被使用原來創(chuàng)建它的轉(zhuǎn)換，再計(jì)算一次（不需要全部重新計(jì)算，只計(jì)算丟失的分區(qū)）
Shared Variables
共享變量
一般來說，當(dāng)一個(gè)函數(shù)被傳遞給Spark操作（例如map和reduce），通常是在集群結(jié)點(diǎn)上運(yùn)行，在函數(shù)中使用到的所有變量，都做分別拷貝，供函數(shù)操作，而不會(huì)互相影響。這些變量會(huì)被拷貝到每一臺(tái)機(jī)器，而在遠(yuǎn)程機(jī)器上，在對(duì)變量的所有更新，都不會(huì)被傳播回Driver程序。然而，Spark提供兩種有限的共享變量，供兩種公用的使用模式：���播變量和累加器
廣播變量
廣播變量允許程序員保留一個(gè)只讀的變量，緩存在每一臺(tái)機(jī)器上，而非每個(gè)任務(wù)保存一份拷貝。他們可以使用，例如，給每個(gè)結(jié)點(diǎn)一個(gè)大的輸入數(shù)據(jù)集，以一種高效的方式。Spark也會(huì)嘗試，使用一種高效的廣播算法，來減少溝通的損耗。
廣播變量是從變量V創(chuàng)建的，通過調(diào)用SparkContext.broadcast(v)方法。這個(gè)廣播變量是一個(gè)v的分裝器，它的只可以通過調(diào)用value方法獲得。如下的解釋器模塊展示了如何應(yīng)用：
scala> val broadcastVar = sc.broadcast(Array(1, 2, 3))
broadcastVar: spark.Broadcast[Array[Int]] = spark.Broadcast(b5c40191-a864-4c7d-b9bf-d87e1a4e787c)
scala> broadcastVar.value
res0: Array[Int] = Array(1, 2, 3)
在廣播變量被創(chuàng)建后，它能在集群運(yùn)行的任何函數(shù)上，被取代v值進(jìn)行調(diào)用，從而v值不需要被再次傳遞到這些結(jié)點(diǎn)上。另外，對(duì)象v不能在被廣播后修改，是只讀的，從而保證所有結(jié)點(diǎn)的變量，收到的都是一模一樣的。
累加器
累加器是只能通過組合操作“加”起來的變量，可以高效的被并行支持。他們可以用來實(shí)現(xiàn)計(jì)數(shù)器（如同MapReduce中）和求和。Spark原生就支持Int和Double類型的計(jì)數(shù)器，程序員可以添加新的類型。
一個(gè)計(jì)數(shù)器，可以通過調(diào)用SparkContext.accumulator(V)方法來創(chuàng)建。運(yùn)行在集群上的任務(wù)，可以使用+=來加值。然而，它們不能讀取計(jì)數(shù)器的值。當(dāng)Driver程序需要讀取值的時(shí)候，它可以使用.value方法。
如下的解釋器，展示了如何利用累加器，將一個(gè)數(shù)組里面的所有元素相加
scala> val accum = sc.accumulator(0)
accum: spark.Accumulator[Int] = 0
scala> sc.parallelize(Array(1, 2, 3, 4)).foreach(x => accum += x)
…
10/09/29 18:41:08 INFO SparkContext: Tasks finished in 0.317106 s
scala> accum.value
res2: Int = 10
更多資料
在Spark的網(wǎng)站上，你可以看到Spark樣例程序
另外，Spark包括了一些例子，在examples/src/main/scala上，有些既有Spark版本，又有本地非并行版本，允許你看到如果要讓程序以集群化的方式跑起來的話，需要做什么改變。你可以運(yùn)行它們，通過將類名傳遞給spark中的run腳本 — 例如./run spark.examples.SparkPi. 每一個(gè)樣例程序，都會(huì)打印使用幫助，當(dāng)運(yùn)行時(shí)沒任何參數(shù)時(shí)。
出自 淘寶 明風(fēng)
出處：http://www.linuxidc.com/Linux/2013-08/88592.htm
參考：http://spark.apache.org/
總結(jié)
- 上一篇： Hadoop Streaming
- 下一篇： A Scala Tutorial for