本文轉載自Spark快速入門指南 – Spark安裝與基礎使用

Apache Spark 是一個新興的大數據處理通用引擎，提供了分布式的內存抽象。Spark 正如其名，最大的特點就是快（Lightning-fast），可比 Hadoop MapReduce 的處理速度快 100 倍。此外，Spark 提供了簡單易用的 API，幾行代碼就能實現 WordCount。本教程主要參考官網快速入門教程，介紹了 Spark 的安裝，Spark shell 、RDD、Spark SQL、Spark Streaming 等的基本使用。

本教程的具體運行環境如下：

• CentOS 6.4
• Spark 1.6
• Hadoop 2.6.0
• Java JDK 1.7
• Scala 2.10.5

準備工作

運行 Spark 需要 Java JDK 1.7，CentOS 6.x 系統默認只安裝了 Java JRE，還需要安裝 Java JDK，并配置好 JAVA_HOME 變量。此外，Spark 會用到 HDFS 與 YARN，因此請先安裝 Hadoop，具體請瀏覽Hadoop安裝教程，在此就不再復述。

安裝 Spark

待 Hadoop 安裝好之后，我們再開始安裝 Spark。

官網下載地址：http://spark.apache.org/downloads.html

本教程選擇的是 Spark 1.6.0 版本，選擇 package type 為 “Pre-build with user-provided Hadoop [can use with most Hadoop distributions]”，再點擊給出的下載連接?http://www.apache.org/dyn/closer.lua/spark/spark-1.6.0/spark-1.6.0-bin-without-hadoop.tgz?就可以下載了，如下圖所示：

從官網下載 Spark

Package type

• Source code: Spark 源碼，需要編譯才能使用，另外 Scala 2.11 需要使用源碼編譯才可使用
• Pre-build with user-provided Hadoop: “Hadoop free” 版，可應用到任意 Hadoop 版本
• Pre-build for Hadoop 2.6 and later: 基于 Hadoop 2.6 的預先編譯版，需要與本機安裝的 Hadoop 版本對應。可選的還有 Hadoop 2.4 and later、Hadoop 2.3、Hadoop 1.x，以及 CDH 4。

為方便，本教程選擇的是 Pre-build with user-provided Hadoop，簡單配置后可應用到任意 Hadoop 版本。

下載后，執行如下命令進行安裝：

sudo tar -zxf ~/下載/spark-1.6.0-bin-without-hadoop.tgz -C /usr/local/

cd /usr/local

sudo mv ./spark-1.6.0-bin-without-hadoop/ ./spark

sudo chown -R hadoop:hadoop ./spark # 此處的 hadoop 為你的用戶名

Shell 命令

安裝后，需要在 ./conf/spark-env.sh 中修改 Spark 的 Classpath，執行如下命令拷貝一個配置文件：

cd /usr/local/spark

cp ./conf/spark-env.sh.template ./conf/spark-env.sh

Shell 命令

編輯 ./conf/spark-env.sh（vim ./conf/spark-env.sh） ，在最后面加上如下一行：

export SPARK_DIST_CLASSPATH=$(/usr/local/hadoop/bin/hadoop classpath)

保存后，Spark 就可以啟動、運行了。

運行 Spark 示例

注意，必須安裝 Hadoop 才能使用 Spark，但如果使用 Spark 過程中沒用到 HDFS，不啟動 Hadoop 也是可以的。此外，接下來教程中出現的命令、目錄，若無說明，則一般以 Spark 的安裝目錄（/usr/local/spark）為當前路徑，請注意區分。

在 ./examples/src/main 目錄下有一些 Spark 的示例程序，有 Scala、Java、Python、R 等語言的版本。我們可以先運行一個示例程序 SparkPi（即計算 π 的近似值），執行如下命令：

cd /usr/local/spark

./bin/run-example SparkPi

Shell 命令

執行時會輸出非常多的運行信息，輸出結果不容易找到，可以通過?grep?命令進行過濾（命令中的?2>&1?可以將所有的信息都輸出到 stdout 中，否則由于輸出日志的性質，還是會輸出到屏幕中）：

./bin/run-example SparkPi 2>&1 | grep "Pi is roughly"

Shell 命令

過濾后的運行結果如下圖所示，可以得到 π 的 5 位小數近似值 ：

從官網下載 Spark

Python 版本的 SparkPi 則需要通過 spark-submit 運行：

./bin/spark-submit examples/src/main/python/pi.py

Shell 命令

通過 Spark Shell 進行交互分析

Spark shell 提供了簡單的方式來學習 API，也提供了交互的方式來分析數據。Spark Shell 支持 Scala 和 Python，本教程選擇使用 Scala 來進行介紹。

Scala

Scala 是一門現代的多范式編程語言，志在以簡練、優雅及類型安全的方式來表達常用編程模式。它平滑地集成了面向對象和函數語言的特性。Scala 運行于 Java 平臺（JVM，Java 虛擬機），并兼容現有的 Java 程序。

Scala 是 Spark 的主要編程語言，如果僅僅是寫 Spark 應用，并非一定要用 Scala，用 Java、Python 都是可以的。使用 Scala 的優勢是開發效率更高，代碼更精簡，并且可以通過 Spark Shell 進行交互式實時查詢，方便排查問題。

執行如下命令啟動 Spark Shell：

./bin/spark-shell

Shell 命令

啟動成功后如圖所示，會有 “scala >” 的命令提示符。

成功啟動Spark Shell

基礎操作

Spark 的主要抽象是分布式的元素集合（distributed collection of items），稱為RDD（Resilient Distributed Dataset，彈性分布式數據集），它可被分發到集群各個節點上，進行并行操作。RDDs 可以通過 Hadoop InputFormats 創建（如 HDFS），或者從其他 RDDs 轉化而來。

我們從 ./README 文件新建一個 RDD，代碼如下（本文出現的 Spark 交互式命令代碼中，與位于同一行的注釋內容為該命令的說明，命令之后的注釋內容表示交互式輸出結果）：

val textFile = sc.textFile("file:///usr/local/spark/README.md")

// textFile: org.apache.spark.rdd.RDD[String] = MapPartitionsRDD[1] at textFile at <console>:27

scala

代碼中通過 “file://” 前綴指定讀取本地文件。Spark shell 默認是讀取 HDFS 中的文件，需要先上傳文件到 HDFS 中，否則會有“org.apache.hadoop.mapred.InvalidInputException: Input path does not exist: hdfs://localhost:9000/user/hadoop/README.md”的錯誤。

上述命令的輸出結果如下圖所示：

新建RDD

RDDs 支持兩種類型的操作

• actions: 在數據集上運行計算后返回值
• transformations: 轉換, 從現有數據集創建一個新的數據集

下面我們就來演示 count() 和 first() 操作：

textFile.count() // RDD 中的 item 數量，對于文本文件，就是總行數

// res0: Long = 95

?

textFile.first() // RDD 中的第一個 item，對于文本文件，就是第一行內容

// res1: String = # Apache Spark

scala

接著演示 transformation，通過 filter transformation 來返回一個新的 RDD，代碼如下：

val linesWithSpark = textFile.filter(line => line.contains("Spark")) // 篩選出包含 Spark 的行

?

linesWithSpark.count() // 統計行數

// res4: Long = 17

scala

可以看到一共有 17 行內容包含 Spark，這與通過 Linux 命令?cat ./README.md | grep "Spark" -c?得到的結果一致，說明是正確的。action 和 transformation 可以用鏈式操作的方式結合使用，使代碼更為簡潔：

textFile.filter(line => line.contains("Spark")).count() // 統計包含 Spark 的行數

// res4: Long = 17

scala

RDD的更多操作

RDD 的 actions 和 transformations 可用在更復雜的計算中，例如通過如下代碼可以找到包含單詞最多的那一行內容共有幾個單詞：

textFile.map(line => line.split(" ").size).reduce((a, b) => if (a > b) a else b)

// res5: Int = 14

scala

代碼首先將每一行內容 map 為一個整數，這將創建一個新的 RDD，并在這個 RDD 中執行 reduce 操作，找到最大的數。map()、reduce() 中的參數是 Scala 的函數字面量（function literals，也稱為閉包 closures），并且可以使用語言特征或 Scala/Java 的庫。例如，通過使用 Math.max() 函數（需要導入 Java 的 Math 庫），可以使上述代碼更容易理解：

import java.lang.Math

?

textFile.map(line => line.split(" ").size).reduce((a, b) => Math.max(a, b))

// res6: Int = 14

scala

Hadoop MapReduce 是常見的數據流模式，在 Spark 中同樣可以實現（下面這個例子也就是 WordCount）：

val wordCounts = textFile.flatMap(line => line.split(" ")).map(word => (word, 1)).reduceByKey((a, b) => a + b) // 實現單詞統計

// wordCounts: org.apache.spark.rdd.RDD[(String, Int)] = ShuffledRDD[4] at reduceByKey at <console>:29

?

wordCounts.collect() // 輸出單詞統計結果

// res7: Array[(String, Int)] = Array((package,1), (For,2), (Programs,1), (processing.,1), (Because,1), (The,1)...)

scala

緩存

Spark 支持在集群范圍內將數據集緩存至每一個節點的內存中，可避免數據傳輸，當數據需要重復訪問時這個特征非常有用，例如查詢體積小的“熱”數據集，或是運行如 PageRank 的迭代算法。調用 cache()，就可以將數據集進行緩存：

linesWithSpark.cache()

scala

Spark SQL 和 DataFrames

Spark SQL 是 Spark 內嵌的模塊，用于結構化數據。在 Spark 程序中可以使用 SQL 查詢語句或?DataFrame API。DataFrames 和 SQL 提供了通用的方式來連接多種數據源，支持 Hive、Avro、Parquet、ORC、JSON、和 JDBC，并且可以在多種數據源之間執行 join 操作。

下面仍在 Spark shell 中演示一下 Spark SQL 的基本操作，該部分內容主要參考了?Spark SQL、DataFrames 和 Datasets 指南。

Spark SQL 的功能是通過 SQLContext 類來使用的，而創建 SQLContext 是通過 SparkContext 創建的。在 Spark shell 啟動時，輸出日志的最后有這么幾條信息

16/01/16 13:25:41 INFO repl.SparkILoop: Created spark context.. Spark context available as sc. 16/01/16 13:25:41 INFO repl.SparkILoop: Created sql context.. SQL context available as sqlContext.

這些信息表明 SparkContent 和 SQLContext 都已經初始化好了，可通過對應的 sc、sqlContext 變量直接進行訪問。

使用 SQLContext 可以從現有的 RDD 或數據源創建 DataFrames。作為示例，我們通過 Spark 提供的 JSON 格式的數據源文件 ./examples/src/main/resources/people.json 來進行演示，該數據源內容如下：

{"name":"Michael"}

{"name":"Andy", "age":30}

{"name":"Justin", "age":19}

json

執行如下命令導入數據源，并輸出內容：

val df = sqlContext.read.json("file:///usr/local/spark/examples/src/main/resources/people.json")

// df: org.apache.spark.sql.DataFrame = [age: bigint, name: string]

?

df.show() // 輸出數據源內容

// +----+-------+

// | age| name|

// +----+-------+

// |null|Michael|

// | 30| Andy|

// | 19| Justin|

// +----+-------+

scala

接著，我們來演示 DataFrames 處理結構化數據的一些基本操作：

df.select("name").show() // 只顯示 "name" 列

// +-------+

// | name|

// +-------+

// |Michael|

// | Andy|

// | Justin|

// +-------+

?

df.select(df("name"), df("age") + 1).show() // 將 "age" 加 1

// +-------+---------+

// | name|(age + 1)|

// +-------+---------+

// |Michael| null|

// | Andy| 31|

// | Justin| 20|

// +-------+---------+

?

df.filter(df("age") > 21).show() # 條件語句

// +---+----+

// |age|name|

// +---+----+

// | 30|Andy|

// +---+----+

?

df.groupBy("age").count().show() // groupBy 操作

// +----+-----+

// | age|count|

// +----+-----+

// |null| 1|

// | 19| 1|

// | 30| 1|

// +----+-----+

scala

當然，我們也可以使用 SQL 語句來進行操作：

df.registerTempTable("people") // 將 DataFrame 注冊為臨時表 people

val result = sqlContext.sql("SELECT name, age FROM people WHERE age >= 13 AND age <= 19") // 執行 SQL 查詢

result.show() // 輸出結果

// +------+---+

// | name|age|

// +------+---+

// |Justin| 19|

// +------+---+

scala

更多的功能可以查看完整的?DataFrames API?，此外 DataFrames 也包含了豐富的?DataFrames Function?可用于字符串處理、日期計算、數學計算等。

Spark Streaming

流計算除了使用?Storm?框架，使用 Spark Streaming 也是一個很好的選擇。基于 Spark Streaming，可以方便地構建可拓展、高容錯的流計算應用程序。Spark Streaming 使用 Spark API 進行流計算，這意味著在 Spark 上進行流處理與批處理的方式一樣。因此，你可以復用批處理的代碼，使用 Spark Streaming 構建強大的交互式應用程序，而不僅僅是用于分析數據。

下面以一個簡單的 Spark Streaming 示例（基于流的單詞統計）來演示一下 Spark Streaming：本地服務器通過 TCP 接收文本數據，實時輸出單詞統計結果。該部分內容主要參考了?Spark Streaming 編程指南。

運行該示例需要 Netcat（在網絡上通過 TCP 或 UDP 讀寫數據），CentOS 6.x 系統中默認沒有安裝，經過測試，如果通過 yum 直接安裝，運行時會有 “nc: Protocol not available” 的錯誤，需要下載較低版本的 nc?才能正常使用。我們選擇 Netcat 0.6.1 版本，在終端中運行如下命令進行安裝：

wget http://downloads.sourceforge.net/project/netcat/netcat/0.6.1/netcat-0.6.1-1.i386.rpm -O ~/netcat-0.6.1-1.i386.rpm # 下載

sudo rpm -iUv ~/netcat-0.6.1-1.i386.rpm # 安裝

Shell 命令

安裝好 NetCat 之后，使用如下命令建立本地數據服務，監聽 TCP 端口 9999：

# 記為終端 1

nc -l -p 9999

Shell 命令

啟動后，該端口就被占用了，需要開啟另一個終端運行示例程序，執行如下命令：

# 需要另外開啟一個終端，記為終端 2，然后運行如下命令

/usr/local/spark/bin/run-example streaming.NetworkWordCount localhost 9999

Shell 命令

接著在終端 1 中輸入文本，在終端 2 中就可以實時看到單詞統計結果了。

Spark Streaming 的內容較多，本教程就簡單介紹到這，更詳細的內容可查看官網教程。最后需要關掉終端 2，并按 ctrl+c 退出 終端 1 的Netcat。

獨立應用程序（Self-Contained Applications）

接著我們通過一個簡單的應用程序 SimpleApp 來演示如何通過 Spark API 編寫一個獨立應用程序。使用 Scala 編寫的程序需要使用 sbt 進行編譯打包，相應的，Java 程序使用 Maven 編譯打包，而 Python 程序通過 spark-submit 直接提交。

應用程序代碼

在終端中執行如下命令創建一個文件夾 sparkapp 作為應用程序根目錄：

cd ~ # 進入用戶主文件夾

mkdir ./sparkapp # 創建應用程序根目錄

mkdir -p ./sparkapp/src/main/scala # 創建所需的文件夾結構

Shell 命令

在 ./sparkapp/src/main/scala 下建立一個名為 SimpleApp.scala 的文件（vim ./sparkapp/src/main/scala/SimpleApp.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 = "file:///usr/local/spark/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))

}

}

scala

該程序計算 /usr/local/spark/README 文件中包含 “a” 的行數 和包含 “b” 的行數。代碼第8行的 /usr/local/spark 為 Spark 的安裝目錄，如果不是該目錄請自行修改。不同于 Spark shell，獨立應用程序需要通過?val sc = new SparkContext(conf)?初始化 SparkContext，SparkContext 的參數 SparkConf 包含了應用程序的信息。

該程序依賴 Spark API，因此我們需要通過 sbt 進行編譯打包。在 ./sparkapp 中新建文件 simple.sbt（vim ./sparkapp/simple.sbt），添加內容如下，聲明該獨立應用程序的信息以及與 Spark 的依賴關系：

name := "Simple Project"version := "1.0"scalaVersion := "2.10.5"libraryDependencies += "org.apache.spark" %% "spark-core" % "1.6.0"

文件 simple.sbt 需要指明 Spark 和 Scala 的版本。啟動 Spark shell 的過程中，當輸出到 Spark 的符號圖形時，可以看到相關的版本信息。

查看 Spark 和 Scala 的版本信息

安裝 sbt

Spark 中沒有自帶 sbt，需要手動安裝 sbt，我們選擇安裝在 /usr/local/sbt 中：

sudo mkdir /usr/local/sbt

sudo chown -R hadoop /usr/local/sbt # 此處的 hadoop 為你的用戶名

cd /usr/local/sbt

Shell 命令

經筆者測試，按官網教程安裝 sbt 0.13.9 后，使用時可能存在網絡問題，無法下載依賴包，導致 sbt 無法正常使用，需要進行一定的修改。為方便，請使用筆者修改后的版本，下載地址：http://pan.baidu.com/s/1eRyFddw。

下載后，執行如下命令拷貝至 /usr/local/sbt 中：

cp ~/下載/sbt-launch.jar .

Shell 命令

接著在 /usr/local/sbt 中創建 sbt 腳本（vim ./sbt），添加如下內容：

#!/bin/bash

SBT_OPTS="-Xms512M -Xmx1536M -Xss1M -XX:+CMSClassUnloadingEnabled -XX:MaxPermSize=256M"

java $SBT_OPTS -jar `dirname $0`/sbt-launch.jar "$@"

Shell 命令

保存后，為 ./sbt 腳本增加可執行權限：

chmod u+x ./sbt

Shell 命令

最后檢驗 sbt 是否可用（首次運行會處于 “Getting org.scala-sbt sbt 0.13.9 …” 的下載狀態，請耐心等待。筆者等待了 7 分鐘才出現第一條下載提示）：

./sbt sbt-version

Shell 命令

下載過程中可能會類似 “Server access Error: java.security.ProviderException: java.security.KeyException url=https://jcenter.bintray.com/org/scala-sbt/precompiled-2_9_3/0.13.9/precompiled-2_9_3-0.13.9.jar” 的錯誤，可以忽略?？稍賵绦幸淮?./sbt sbt-version，只要能得到如下圖的版本信息就沒問題：

驗證 sbt 是否可用

如果由于網絡問題無法下載依賴，導致 sbt 無法正確運行的話，可以下載筆者提供的離線依賴包 sbt-0.13.9-repo.tar.gz 到本地中（依賴包的本地位置為 ~/.sbt 和 ~/.ivy2，檢查依賴關系時，首先檢查本地，本地未找到，再從網絡中下載），下載地址：http://pan.baidu.com/s/1sjTQ8yD。下載后，執行如下命令解壓依賴包：

tar -zxf ~/下載/sbt-0.13.9-local-repo.tar.gz ~

Shell 命令

通過這個方式，一般可以解決依賴包缺失的問題（讀者提供的依賴包僅適合于 Spark 1.6 版本，不同版本依賴關系不一樣）。

如果對 sbt 存在的網絡問題以及如何解決感興趣，請點擊下方查看。

點擊查看：解決 sbt 無法下載依賴包的問題

使用 sbt 打包 Scala 程序

為保證 sbt 能正常運行，先執行如下命令檢查整個應用程序的文件結構：

cd ~/sparkapp

find .

Shell 命令

文件結構應如下圖所示：

SimpleApp的文件結構

接著，我們就可以通過如下代碼將整個應用程序打包成 JAR（首次運行同樣需要下載依賴包，如果這邊遇到網絡問題無法成功，也請下載上述安裝 sbt 提到的離線依賴包 sbt-0.13.9-repo.tar.gz ）：

/usr/local/sbt/sbt package

Shell 命令

打包成功的話，會輸出如下圖內容：

SimpleApp的文件結構

生成的 jar 包的位置為 ~/sparkapp/target/scala-2.10/simple-project_2.10-1.0.jar。

通過 spark-submit 運行程序

最后，我們就可以將生成的 jar 包通過 spark-submit 提交到 Spark 中運行了，命令如下：

/usr/local/spark/bin/spark-submit --class "SimpleApp" ~/sparkapp/target/scala-2.10/simple-project_2.10-1.0.jar

# 輸出信息太多，可以通過如下命令過濾直接查看結果

/usr/local/spark/bin/spark-submit --class "SimpleApp" ~/sparkapp/target/scala-2.10/simple-project_2.10-1.0.jar 2>&1 | grep "Lines with a:"

Shell 命令

最終得到的結果如下：

Lines with a: 58, Lines with b: 26

自此，你就完成了你的第一個 Spark 應用程序了。

總結

以上是生活随笔為你收集整理的Spark快速入门指南 – Spark安装与基础使用的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。