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