spark入門介紹(菜鳥必看)經驗技術分享

什麼是Spark

Apache Spark是一個圍繞速度、易用性和複雜分析建構的大資料處理架構。最初在2009年由加州大學伯克利分校的AMPLab開發，并于2010年成為Apache的開源項目之一。

與Hadoop和Storm等其他大資料和MapReduce技術相比，Spark有如下優勢。

分享之前我還是要推薦下我自己建立的大資料學習交流Qun710219868 進Qun聊邀請碼填寫 南風（必填）我就知道是你了

首先，Spark為我們提供了一個全面、統一的架構用于管理各種有着不同性質（文本資料、圖表資料等）的資料集和資料源（批量資料或實時的流資料）的大資料處理的需求。

Spark可以将Hadoop叢集中的應用在記憶體中的運作速度提升100倍，甚至能夠将應用在磁盤上的運作速度提升10倍。

Spark讓開發者可以快速的用Java、Scala或Python編寫程式。它本身自帶了一個超過80個高階操作符集合。而且還可以用它在shell中以互動式地查詢資料。

除了Map和Reduce操作之外，它還支援SQL查詢，流資料，機器學習和圖表資料處理。開發者可以在一個資料管道用例中單獨使用某一能力或者将這些能力結合在一起使用。

在這個Apache Spark文章系列的第一部分中，我們将了解到什麼是Spark，它與典型的MapReduce解決方案的比較以及它如何為大資料處理提供了一套完整的工具。

Hadoop和Spark

Hadoop這項大資料處理技術大概已有十年曆史，而且被看做是首選的大資料集合處理的解決方案。MapReduce是一路計算的優秀解決方案，不過對于需要多路計算和算法的用例來說，并非十分高效。資料處理流程中的每一步都需要一個Map階段和一個Reduce階段，而且如果要利用這一解決方案，需要将所有用例都轉換成MapReduce模式。

在下一步開始之前，上一步的作業輸出資料必須要存儲到分布式檔案系統中。是以，複制和磁盤存儲會導緻這種方式速度變慢。另外Hadoop解決方案中通常會包含難以安裝和管理的叢集。而且為了處理不同的大資料用例，還需要內建多種不同的工具（如用于機器學習的Mahout和流資料處理的Storm）。

如果想要完成比較複雜的工作，就必須将一系列的MapReduce作業串聯起來然後順序執行這些作業。每一個作業都是高時延的，而且隻有在前一個作業完成之後下一個作業才能開始啟動。

而Spark則允許程式開發者使用有向無環圖（DAG）開發複雜的多步資料管道。而且還支援跨有向無環圖的記憶體資料共享，以便不同的作業可以共同處理同一個資料。

Spark運作在現有的Hadoop分布式檔案系統基礎之上（HDFS）提供額外的增強功能。它支援将Spark應用部署到現存的Hadoop v1叢集（with SIMR – Spark-Inside-MapReduce）或Hadoop v2 YARN叢集甚至是Apache Mesos之中。

我們應該将Spark看作是Hadoop MapReduce的一個替代品而不是Hadoop的替代品。其意圖并非是替代Hadoop，而是為了提供一個管理不同的大資料用例和需求的全面且統一的解決方案。

Spark特性

Spark通過在資料處理過程中成本更低的洗牌（Shuffle）方式，将MapReduce提升到一個更高的層次。利用記憶體資料存儲和接近實時的處理能力，Spark比其他的大資料處理技術的性能要快很多倍。

Spark還支援大資料查詢的延遲計算，這可以幫助優化大資料處理流程中的處理步驟。Spark還提供進階的API以提升開發者的生産力，除此之外還為大資料解決方案提供一緻的體系架構模型。

Spark将中間結果儲存在記憶體中而不是将其寫入磁盤，當需要多次處理同一資料集時，這一點特别實用。Spark的設計初衷就是既可以在記憶體中又可以在磁盤上工作的執行引擎。當記憶體中的資料不适用時，Spark操作符就會執行外部操作。Spark可以用于處理大于叢集記憶體容量總和的資料集。

Spark會嘗試在記憶體中存儲盡可能多的資料然後将其寫入磁盤。它可以将某個資料集的一部分存入記憶體而剩餘部分存入磁盤。開發者需要根據資料和用例評估對記憶體的需求。Spark的性能優勢得益于這種記憶體中的資料存儲。

Spark的其他特性包括：

支援比Map和Reduce更多的函數。

優化任意操作算子圖（operator graphs）。

可以幫助優化整體資料處理流程的大資料查詢的延遲計算。

提供簡明、一緻的Scala，Java和Python API。

提供互動式Scala和Python Shell。目前暫不支援Java。

Spark是用Scala程式設計語言編寫而成，運作于Java虛拟機（JVM）環境之上。目前支援如下程式設計語言編寫Spark應用：

Scala

Java

Python

Clojure

R

Spark生态系統

除了Spark核心API之外，Spark生态系統中還包括其他附加庫，可以在大資料分析和機器學習領域提供更多的能力。

這些庫包括：

Spark Streaming:

Spark Streaming基于微批量方式的計算和處理，可以用于處理實時的流資料。它使用DStream，簡單來說就是一個彈性分布式資料集（RDD）系列，處理實時資料。

Spark SQL:

Spark SQL可以通過JDBC API将Spark資料集暴露出去，而且還可以用傳統的BI和可視化工具在Spark資料上執行類似SQL的查詢。使用者還可以用Spark SQL對不同格式的資料（如JSON，Parquet以及資料庫等）執行ETL，将其轉化，然後暴露給特定的查詢。

Spark MLlib:

MLlib是一個可擴充的Spark機器學習庫，由通用的學習算法和工具組成，包括二進制分類、線性回歸、聚類、協同過濾、梯度下降以及底層優化原語。

Spark GraphX:

GraphX是用于圖計算和并行圖計算的新的（alpha）Spark API。通過引入彈性分布式屬性圖（Resilient Distributed Property Graph），一種頂點和邊都帶有屬性的有向多重圖，擴充了Spark RDD。為了支援圖計算，GraphX暴露了一個基礎操作符集合（如subgraph，joinVertices和aggregateMessages）和一個經過優化的Pregel API變體。此外，GraphX還包括一個持續增長的用于簡化圖分析任務的圖算法和建構器集合。

除了這些庫以外，還有一些其他的庫，如BlinkDB和Tachyon。

BlinkDB是一個近似查詢引擎，用于在海量資料上執行互動式SQL查詢。BlinkDB可以通過犧牲資料精度來提升查詢響應時間。通過在資料樣本上執行查詢并展示包含有意義的錯誤線注解的結果，操作大資料集合。

Tachyon是一個以記憶體為中心的分布式檔案系統，能夠提供記憶體級别速度的跨叢集架構（如Spark和MapReduce）的可信檔案共享。它将工作集檔案緩存在記憶體中，進而避免到磁盤中加載需要經常讀取的資料集。通過這一機制，不同的作業/查詢和架構可以以記憶體級的速度通路緩存的檔案。

此外，還有一些用于與其他産品內建的擴充卡，如Cassandra（Spark Cassandra 連接配接器）和R（SparkR）。Cassandra Connector可用于通路存儲在Cassandra資料庫中的資料并在這些資料上執行資料分析。

下圖展示了在Spark生态系統中，這些不同的庫之間的互相關聯。

圖1. Spark架構中的庫

我們将在這一系列文章中逐漸探索這些Spark庫

Spark體系架構

Spark體系架構包括如下三個主要元件：

資料存儲

API

管理架構

接下來讓我們詳細了解一下這些元件。

資料存儲：

Spark用HDFS檔案系統存儲資料。它可用于存儲任何相容于Hadoop的資料源，包括HDFS，HBase，Cassandra等。

API：

利用API，應用開發者可以用标準的API接口建立基于Spark的應用。Spark提供Scala，Java和Python三種程式設計語言的API。

下面是三種語言Spark API的網站連結。

Scala API

資源管理：

Spark既可以部署在一個單獨的伺服器也可以部署在像Mesos或YARN這樣的分布式計算架構之上。

下圖2展示了Spark體系架構模型中的各個元件。

圖2 Spark體系架構

彈性分布式資料集

彈性分布式資料集（基于Matei的研究論文）或RDD是Spark架構中的核心概念。可以将RDD視作資料庫中的一張表。其中可以儲存任何類型的資料。Spark将資料存儲在不同分區上的RDD之中。

RDD可以幫助重新安排計算并優化資料處理過程。

此外，它還具有容錯性，因為RDD知道如何重新建立和重新計算資料集。

RDD是不可變的。你可以用變換（Transformation）修改RDD，但是這個變換所傳回的是一個全新的RDD，而原有的RDD仍然保持不變。

RDD支援兩種類型的操作：

變換（Transformation）

行動（Action）

變換：變換的傳回值是一個新的RDD集合，而不是單個值。調用一個變換方法，不會有任何求值計算，它隻擷取一個RDD作為參數，然後傳回一個新的RDD。

變換函數包括：map，filter，flatMap，groupByKey，reduceByKey，aggregateByKey，pipe和coalesce。

行動：行動操作計算并傳回一個新的值。當在一個RDD對象上調用行動函數時，會在這一時刻計算全部的資料處理查詢并傳回結果值。

行動操作包括：reduce，collect，count，first，take，countByKey以及foreach。

如何安裝Spark

安裝和使用Spark有幾種不同方式。你可以在自己的電腦上将Spark作為一個獨立的架構安裝或者從諸如Cloudera，HortonWorks或MapR之類的供應商處擷取一個Spark虛拟機鏡像直接使用。或者你也可以使用在雲端環境（如Databricks Cloud）安裝并配置好的Spark。

在本文中，我們将把Spark作為一個獨立的架構安裝并在本地啟動它。最近Spark剛剛釋出了1.2.0版本。我們将用這一版本完成示例應用的代碼展示。

如何運作Spark

當你在本地機器安裝了Spark或使用了基于雲端的Spark後，有幾種不同的方式可以連接配接到Spark引擎。

下表展示了不同的Spark運作模式所需的Master URL參數。

如何與Spark互動

Spark啟動并運作後，可以用Spark shell連接配接到Spark引擎進行互動式資料分析。Spark shell支援Scala和Python兩種語言。Java不支援互動式的Shell，是以這一功能暫未在Java語言中實作。

可以用spark-shell.cmd和pyspark.cmd指令分别運作Scala版本和Python版本的Spark Shell。

Spark網頁控制台

不論Spark運作在哪一種模式下，都可以通過通路Spark網頁控制台檢視Spark的作業結果和其他的統計資料，控制台的URL位址如下：

http://localhost:4040

Spark控制台如下圖3所示，包括Stages，Storage，Environment和Executors四個标簽頁

圖3. Spark網頁控制台

共享變量

Spark提供兩種類型的共享變量可以提升叢集環境中的Spark程式運作效率。分别是廣播變量和累加器。

廣播變量：廣播變量可以在每台機器上緩存隻讀變量而不需要為各個任務發送該變量的拷貝。他們可以讓大的輸入資料集的叢集拷貝中的節點更加高效。

下面的代碼片段展示了如何使用廣播變量。

//

// Broadcast Variables

val broadcastVar = sc.broadcast(Array(1, 2, 3))

broadcastVar.value

累加器：隻有在使用相關操作時才會添加累加器，是以它可以很好地支援并行。累加器可用于實作計數（就像在MapReduce中那樣）或求和。可以用add方法将運作在叢集上的任務添加到一個累加器變量中。不過這些任務無法讀取變量的值。隻有驅動程式才能夠讀取累加器的值。

下面的代碼片段展示了如何使用累加器共享變量：

// Accumulators

val accum = sc.accumulator(0, "My Accumulator")

sc.parallelize(Array(1, 2, 3, 4)).foreach(x => accum += x)

accum.value

Spark應用示例

本篇文章中所涉及的示例應用是一個簡單的字數統計應用。這與學習用Hadoop進行大資料處理時的示例應用相同。我們将在一個文本檔案上執行一些資料分析查詢。本示例中的文本檔案和資料集都很小，不過無須修改任何代碼，示例中所用到的Spark查詢同樣可以用到大容量資料集之上。

為了讓讨論盡量簡單，我們将使用Spark Scala Shell。

首先讓我們看一下如何在你自己的電腦上安裝Spark。

前提條件：

為了讓Spark能夠在本機正常工作，你需要安裝Java開發工具包（JDK）。這将包含在下面的第一步中。

同樣還需要在電腦上安裝Spark軟體。下面的第二步将介紹如何完成這項工作。

注：下面這些指令都是以Windows環境為例。如果你使用不同的作業系統環境，需要相應的修改系統變量和目錄路徑已比對你的環境。

I. 安裝JDK

1）從Oracle網站上下載下傳JDK。推薦使用JDK 1.7版本。

将JDK安裝到一個沒有空格的目錄下。對于Windows使用者，需要将JDK安裝到像c:\dev這樣的檔案夾下，而不能安裝到“c:\Program Files”檔案夾下。“c:\Program Files”檔案夾的名字中包含空格，如果軟體安裝到這個檔案夾下會導緻一些問題。

注：不要在“c:\Program Files”檔案夾中安裝JDK或（第二步中所描述的）Spark軟體。

2）完成JDK安裝後，切換至JDK 1.7目錄下的”bin“檔案夾，然後鍵入如下指令，驗證JDK是否正确安裝：

java -version

如果JDK安裝正确，上述指令将顯示Java版本。

II. 安裝Spark軟體：

從Spark網站上下載下傳最新版本的Spark。在本文發表時，最新的Spark版本是1.2。你可以根據Hadoop的版本選擇一個特定的Spark版本安裝。我下載下傳了與Hadoop 2.4或更高版本比對的Spark，檔案名是spark-1.2.0-bin-hadoop2.4.tgz。

将安裝檔案解壓到本地檔案夾中（如：c:\dev）。

為了驗證Spark安裝的正确性，切換至Spark檔案夾然後用如下指令啟動Spark Shell。這是Windows環境下的指令。如果使用Linux或Mac OS，請相應地編輯指令以便能夠在相應的平台上正确運作。

c:

cd c:\dev\spark-1.2.0-bin-hadoop2.4

bin\spark-shell

如果Spark安裝正确，就能夠在控制台的輸出中看到如下資訊。

….

15/01/17 23:17:46 INFO HttpServer: Starting HTTP Server

15/01/17 23:17:46 INFO Utils: Successfully started service 'HTTP class server' on port 58132.

Welcome to

/ __/__  ___ _____/ /__
_\ \/ _ \/ _ `/ __/  '_/           

// ./_,// //_\ version 1.2.0

/_/

Using Scala version 2.10.4 (Java HotSpot(TM) 64-Bit Server VM, Java 1.7.0_71)

Type in expressions to have them evaluated.

Type :help for more information.

15/01/17 23:17:53 INFO BlockManagerMaster: Registered BlockManager

15/01/17 23:17:53 INFO SparkILoop: Created spark context..

Spark context available as sc.

可以鍵入如下指令檢查Spark Shell是否工作正常。

sc.version

（或）

sc.appName

完成上述步驟之後，可以鍵入如下指令退出Spark Shell視窗：

:quit

如果想啟動Spark Python Shell，需要先在電腦上安裝Python。你可以下載下傳并安裝Anaconda，這是一個免費的Python發行版本，其中包括了一些比較流行的科學、數學、工程和資料分析方面的Python包。

然後可以運作如下指令啟動Spark Python Shell：

bin\pyspark

Spark示例應用

完成Spark安裝并啟動後，就可以用Spark API執行資料分析查詢了。

這些從文本檔案中讀取并處理資料的指令都很簡單。我們将在這一系列文章的後續文章中向大家介紹更進階的Spark架構使用的用例。

首先讓我們用Spark API運作流行的Word Count示例。如果還沒有運作Spark Scala Shell，首先打開一個Scala Shell視窗。這個示例的相關指令如下所示：

import org.apache.spark.SparkContext

import org.apache.spark.SparkContext._

val txtFile = "README.md"

val txtData = sc.textFile(txtFile)

txtData.cache()

我們可以調用cache函數将上一步生成的RDD對象儲存到緩存中，在此之後Spark就不需要在每次資料查詢時都重新計算。需要注意的是，cache()是一個延遲操作。在我們調用cache時，Spark并不會馬上将資料存儲到記憶體中。隻有當在某個RDD上調用一個行動時，才會真正執行這個操作。

現在，我們可以調用count函數，看一下在文本檔案中有多少行資料。

txtData.count()

然後，我們可以執行如下指令進行字數統計。在文本檔案中統計資料會顯示在每個單詞的後面。

val wcData = txtData.flatMap(l => l.split(" ")).map(word => (word, 1)).reduceByKey( + )

wcData.collect().foreach(println)

如果想檢視更多關于如何使用Spark核心API的代碼示例，請參考網站上的Spark文檔。

後續計劃

在後續的系列文章中，我們将從Spark SQL開始，學習更多關于Spark生态系統的其他部分。之後，我們将繼續了解Spark Streaming，Spark MLlib和Spark GraphX。我們也會有機會學習像Tachyon和BlinkDB等架構。