1、在maven裡面新增引用，spark和hdfs的客戶端的。

groupId = org.apache.spark
artifactId = spark-core_2.9.3
version = 0.8.1-incubating 

groupId = org.apache.hadoop
artifactId = hadoop-client
version = <your-hdfs-version>

2、把assembly/target/spark-assembly_2.9.3-0.8.1-incubating.jar新增到classpath裡面，然後我們在程式裡面要新增以下引用。

import org.apache.spark.SparkContext
import org.apache.spark.SparkContext._

3、下面是官方的WorkCount的例子，可以參考一下。

/*** SimpleApp.scala ***/
import org.apache.spark.SparkContext
import org.apache.spark.SparkContext._

object SimpleApp {
  def main(args: Array[String]) {
    val logFile = "$YOUR_SPARK_HOME/README.md" // Should be some file on your system
    val sc = new SparkContext("local", "Simple App", "YOUR_SPARK_HOME", List("target/scala-2.9.3/simple-project_2.9.3-1.0.jar"))
    val logData = sc.textFile(logFile, 2).cache()
    val numAs = logData.filter(line => line.contains("a")).count()
    val numBs = logData.filter(line => line.contains("b")).count()
    println("Lines with a: %s, Lines with b: %s".format(numAs, numBs))
 

val sc = new SparkContext("local", "Simple App", "YOUR_SPARK_HOME", List("target/scala-2.9.3/simple-project_2.9.3-1.0.jar"))

SparkContext是SparkContext的上下文物件，是非常核心的一個類，它的例項化方法是new SparkContext(master, appName, [sparkHome], [jars])。
master：master的地址。
appName：應用的名稱。
sparkHome：spark的安裝地址。
jars：jar包的位置。

4、Spark總是圍繞這個一個概念來進行 resilient distributed dataset (RDD)，是可以並行操作的支援容錯的元素集合。目前支援兩種型別的RDDs，parallelized collections和Hadoop datasets。
（1）Parallelized collections是scala中存在的集合類，並且支援並行操作。
 

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

　正常情況之下，spark會自動設定並行任務所需要的cpu的分片，一般是每個cpu 2-4個分片，也可以自己手動設定，sc.parallelize(data, 10)。

 （2）Spark支援hadoop上的任何資料集，比如text files, SequenceFiles，還有其它的InputFormat。

   下面是text files的例子：

scala> val distFile = sc.textFile("data.txt")
distFile: spark.RDD[String] = spark.HadoopRDD@1d4cee08

   SequenceFiles則使用SparkContext’s sequenceFile[K, V] ，比如sequenceFile[Int, String]，Int對應的是IntWritable，String對應的是Text。

   別的資料格式使用SparkContext.hadoopRDD，之後再介紹，這個文件沒有介紹。

   正常情況之下，spark是一個block一個任務。

（3）RDDs只支援兩種操作: transformations,  從一個數據集轉換成另外一種; actions, 通過對一個數據集進行運算之後返回一個值。

     Spark當中所有的transformations都是延遲執行的，等到真正使用的時候才會進行運算。

     預設的，每一個經過transformed的RDD當有action作用於它的時候，它會重新計算一遍，除非我們進行persist (or cache) 操作。

     最後附錄一下RDD的API地址：http://spark.incubator.apache.org/docs/latest/api/core/index.html#org.apache.spark.rdd.RDD

　（4）RDD Persistence 

Spark最重要的一個功能就是可以把RDD持久化或者快取，當你進行一個持久化操作的時候，Spark會在所有節點的記憶體當中儲存這個RDD，第一個的時候計算，之後一直使用不需要再重新計算了。快取是實現迭代式演算法的關鍵。我們可以使用persist() or cache()方法來持久化一個RDD，它是容錯的，當這個RDD的任何分片丟失之後，它會在之前計算它的機器上重新計算。另外每一個RDD，有它自己的儲存Level，儲存在硬碟或者儲存在記憶體，但是序列化成Java物件（節省空間），或者在叢集間複製。要設定它，我們需要傳遞一個StorageLevel給persist()，cache()是預設的了是StorageLevel.MEMORY_ONLY (儲存為反序列化物件在記憶體當中)

　 當記憶體足夠的時候，我們可以使用MEMORY_ONLY；當記憶體不太好的時候，我們可以採用MEMORY_ONLY_SER，在記憶體中儲存為一個位元組陣列，速度還可以；當操作的資料集合足夠大的時候，我們就把中間結果寫到硬碟上；如果要支援容錯，就使用備份到2個節點上的方式。如果要自己定義一個的話，要使用StorageLevel的apply()方法。      

5、共享變數

   Spark提供了兩種限制的共享變數，Broadcast和Accumulators。

   （1）Broadcast允許程式設計師持有一個只讀的變數在各個節點之間，它一個常用的場景就是用它來儲存一個很大的輸入的資料集給每個節點使用，Spark會只用它獨有的廣播演算法來減少通訊損失。下面是例子：

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)

    （2）Accumulators是用來計數或者求總數的，使用SparkContext.accumulator(v)來給它一個初始化的值，然後用“+=”來進行操作，但是任務之間不能得到它的結果，只有驅動任務的程式可以得到它的結果。下面是例子：

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