本文簡單介紹了Spark的使用方式。首先介紹Spark的交互界面的API使用，然后介紹如何使用Java、Scala以及Python編寫Spark應用。詳細的介紹請閱讀 Spark Programming Guide 。

在按照本文進行操作之前，請確保已安裝Spark。本文中的所有操作沒有使用HDFS，所以您可以安裝任何版本的Hadoop。

Spark交互式Shell的使用

基礎

Spark的交互式Shell提供了一個簡單的方式來學習Spark的API，同時也提供了強大的交互式數據處理能力。Spark Shell支持Scala和Python兩種語言。啟動支持Scala的Spark Shell方式為

./bin/spark-shell

Spark最重要的一個抽象概念是彈性分布式數據集(Resilient Distributed Dataset)簡稱RDD。RDDs可以通過Hadoop InputFormats(例如HDFS文件)創建，也可以由其它RDDs轉換而來。下面的例子是通過加載Spark目錄下的README.md文件生成 RDD的例子：

scala> val textFile = sc.textFile("README.md")
textFile: spark.RDD[String] = spark.MappedRDD@2ee9b6e3

RDDs有兩種操作：

• actions：返回計算值
• transformations：返回一個新RDDs的引用

actions示例如下：

scala> textFile.count() // Number of items in this RDD
res0: Long = 126

scala> textFile.first() // First item in this RDD
res1: String = # Apache Spark</pre> 
 
如下transformations示例，使用filter操作返回了一個新的RDD，該RDD為文件中數據項的子集，該子集符合過濾條件：

scala> val linesWithSpark = textFile.filter(line => line.contains("Spark"))
linesWithSpark: spark.RDD[String] = spark.FilteredRDD@7dd4af09
 
 
Spark也支持將actions和transformations一起使用：

scala> textFile.filter(line => line.contains("Spark")).count() // How many lines contain "Spark"?
res3: Long = 15
 
 
 更多RDD操作(More on RDD Operations) 

RDD的actions和transformations操作可以用于更加復雜的計算。下面是查找README.md文件中單詞數最多的行的單詞數目：
scala> textFile.map(line => line.split(" ").size).reduce((a, b) => if (a > b) a else b)
res4: Long = 15
 
 
上面代碼中，第一個map操作將一行文本按空格分隔，并計算單詞數目，將line映射為一個integer值，并創建了一個新的RDD保存這些 integer值。RDD調用reduce計算最大的單詞數。示例中map和reduce操作的參數是Scala的函數式編程風格，Spark支持 Scala、Java、Python的編程風格，并支持Scala/Java庫。例如，使用Scala中的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))
res5: Int = 15</pre> 
 
隨著Hadoop的流行，MapReduce變為一種常見的數據流模式。Spark可以輕松的實現MapReduce，使用Spark編寫MapReduce程序更加簡單：

scala> val wordCounts = textFile.flatMap(line => line.split(" ")).map(word => (word, 1)).reduceByKey((a, b) => a + b)
wordCounts: spark.RDD[(String, Int)] = spark.ShuffledAggregatedRDD@71f027b8
 
 
上面示例中，使用flatMap、map和reduceByKey操作來計算每個單詞在文件中出現的次數，并生成一個結構為的RDD。可以使用collect操作完成單詞統計結果的收集整合：

scala> wordCounts.collect()
res6: Array[(String, Int)] = Array((means,1), (under,2), (this,3), (Because,1), (Python,2), (agree,1), (cluster.,1), ...)
 
 
 緩存 

Spark支持將數據緩存到集群的分布式內存中。在數據會被重復訪問的情況下，將數據緩存到內存能減少數據訪問時間，從而提高運行效率。尤其是在數據分布在幾十或幾百個節點上時，效果更加明顯。下面為將數據linesWithSpark緩存到內存的示例：
scala> linesWithSpark.cache()
res7: spark.RDD[String] = spark.FilteredRDD@17e51082

scala> linesWithSpark.count()
res8: Long = 19
scala> linesWithSpark.count()
res9: Long = 19</pre> 
 
 獨立應用 

假設我們想使用Spark API編寫獨立應用程序。我們可以使用Scala、Java和Python輕松的編寫Spark應用。下面示例為一個簡單的應用示例:
 
  
Scala

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

object SimpleApp {
  def main(args: Array[String]) {
    val logFile = "YOUR_SPARK_HOME/README.md" // Should be some file on your system
    val conf = new SparkConf().setAppName("Simple Application")
    val sc = new SparkContext(conf)
    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))
  }
}</pre> 
 
上面程序分別統計了README中包含字符‘a’以及‘b’的行數。與前面Spark shell例子不同的是，我們需要初始化SparkContext。

我們通過SparkContext創建了一個SparkConf對象，SparkConf對象包含應用的基本信息。
我們基于Spark API編寫應用，所以我們需要編寫一個名為“simple.sbt”的sbt配置文件，用于指明Spark為該應用的一個依賴。下面的sbt配置文件示例中，還增加了Spark的一個依賴庫“spark-core”:
name := "Simple Project"

version := "1.0"
scalaVersion := "2.10.5"
libraryDependencies += "org.apache.spark" %% "spark-core" % "1.6.0"</pre> 
 
為了讓sbt正確執行，我們需要對SimpleApp.scala和simple.sbt根據sbt要求的目錄結構布局。如果布局正確，就可以生成該應用的JAR包，使用spark-submit命令即可運行該程序。

 
  
Javaga

 
/ SimpleApp.java /
import org.apache.spark.api.java.*;
import org.apache.spark.SparkConf;
import org.apache.spark.api.java.function.Function;

public class SimpleApp {
  public static void main(String[] args) {
    String logFile = "YOUR_SPARK_HOME/README.md"; // Should be some file on your system
    SparkConf conf = new SparkConf().setAppName("Simple Application");
    JavaSparkContext sc = new JavaSparkContext(conf);
    JavaRDD


  logData = sc.textFile(logFile).cache();      long numAs = logData.filter(new Function
</string></pre> 
 
該示例的代碼邏輯同上一段Scala示例代碼。與Scala示例類似，首先初始化了SparkContext，通過SparkContext創建了JavaSparkContext對象。并創建了RDDs以及執行transformations操作。最后，通過繼承了spark.api.java.function.Function的類將函數傳給Spark。

在這里，使用Maven進行編譯，Maven的pom.xml如下：
<project> 
  <groupId>edu.berkeley</groupId> 
  <artifactId>simple-project</artifactId> 
  <modelVersion>4.0.0</modelVersion> 
  <name>Simple Project</name> 
  <packaging>jar</packaging> 
  <version>1.0</version> 
  <dependencies> 
    <dependency> <!-- Spark dependency --> 
      <groupId>org.apache.spark</groupId> 
      <artifactId>spark-core_2.10</artifactId> 
      <version>1.6.0</version> 
    </dependency> 
  </dependencies> 
</project> 
 
  
   
  
  
   
    
     
    
   
  
 
按照Maven的要求架構配置文件位置：

$ find .
./pom.xml
./src
./src/main
./src/main/java
./src/main/java/SimpleApp.java
 
 
現在，就可以使用Maven打包應用，以及使用命令./bin/spark-submit.執行該應用程序。示例如下：

# Package a JAR containing your application
$ mvn package
...
[INFO] Building jar: {..}/{..}/target/simple-project-1.0.jar

Use spark-submit to run your application
$ YOUR_SPARK_HOME/bin/spark-submit \
  --class "SimpleApp" \
  --master local[4] \
  target/simple-project-1.0.jar
...
Lines with a: 46, Lines with b: 23</pre> 
 
轉載請注明出處：http://www.cnblogs.com/BYRans/
 </div>