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

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

Spark交互式Shell的使用

基礎

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

./bin/spark-shell

Spark最重要的一個抽象概念是彈性分布式數(shù)據(jù)集(Resilient Distributed Dataset)簡稱RDD。RDDs可以通過Hadoop InputFormats(例如HDFS文件)創(chuàng)建，也可以由其它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

如下transformations示例，使用filter操作返回了一個新的RDD，該RDD為文件中數(shù)據(jù)項的子集，該子集符合過濾條件：

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文件中單詞數(shù)最多的行的單詞數(shù)目：

scala> textFile.map(line => line.split(" ").size).reduce((a, b) => if (a > b) a else b)
res4: Long = 15

上面代碼中，***個map操作將一行文本按空格分隔，并計算單詞數(shù)目，將line映射為一個integer值，并創(chuàng)建了一個新的RDD保存這些 integer值。RDD調用reduce計算***的單詞數(shù)。示例中map和reduce操作的參數(shù)是Scala的函數(shù)式編程風格，Spark支持 Scala、Java、Python的編程風格，并支持Scala/Java庫。例如，使用Scala中的Math.max()函數(shù)讓程序變得更加簡潔易讀：

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

隨著Hadoop的流行，MapReduce變?yōu)橐环N常見的數(shù)據(jù)流模式。Spark可以輕松的實現(xiàn)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操作來計算每個單詞在文件中出現(xiàn)的次數(shù)，并生成一個結構為的RDD?？梢允褂胏ollect操作完成單詞統(tǒng)計結果的收集整合：

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

緩存

Spark支持將數(shù)據(jù)緩存到集群的分布式內存中。在數(shù)據(jù)會被重復訪問的情況下，將數(shù)據(jù)緩存到內存能減少數(shù)據(jù)訪問時間，從而提高運行效率。尤其是在數(shù)據(jù)分布在幾十或幾百個節(jié)點上時，效果更加明顯。下面為將數(shù)據(jù)linesWithSpark緩存到內存的示例：

scala> linesWithSpark.cache()
res7: spark.RDD[String] = spark.FilteredRDD@17e51082

scala> linesWithSpark.count()
res8: Long = 19

scala> linesWithSpark.count()
res9: Long = 19

獨立應用

假設我們想使用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))
  }
}

上面程序分別統(tǒng)計了README中包含字符‘a’以及‘b’的行數(shù)。與前面Spark shell例子不同的是，我們需要初始化SparkContext。

我們通過SparkContext創(chuàng)建了一個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"

為了讓sbt正確執(zhí)行，我們需要對SimpleApp.scala和simple.sbt根據(jù)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

該示例的代碼邏輯同上一段Scala示例代碼。與Scala示例類似，首先初始化了SparkContext，通過SparkContext創(chuàng)建了JavaSparkContext對象。并創(chuàng)建了RDDs以及執(zhí)行transformations操作。***，通過繼承了spark.api.java.function.Function的類將函數(shù)傳給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

現(xiàn)在，就可以使用Maven打包應用，以及使用命令./bin/spark-submit.執(zhí)行該應用程序。示例如下：

# 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

轉載請注明出處：http://www.cnblogs.com/BYRans/