Spark入門指南

   一、初識Spark和Hadoop

     Apache Spark 是一個新興的大數據處理通用引擎，提供了分布式的內存抽象。Spark 正如其名，最大的特點就是快（Lightning-fast），可比 Hadoop MapReduce 的處理速度快 100 倍。

     Hadoop實質上更多是一個分布式數據基礎設施: 它將巨大的數據集分派到一個由普通計算機組成的集群中的多個節點進行存儲，意味著你不需要購買和維護昂貴的服務器硬件。

同時，Hadoop還會索引和跟蹤這些數據，讓大數據處理和分析效率達到前所未有的高度。Spark則是一個專門用來對那些分布式存儲的大數據進行處理的工具，它并不會進行分布式數據的存儲。

Hadoop除了提供為大家所共識的HDFS分布式數據存儲功能之外，還提供了叫做MapReduce的數據處理功能。所以我們完全可以拋開Spark，使用Hadoop自身的MapReduce來完成數據的處理。

當然，Spark也不是非要依附在Hadoop身上才能生存。但它沒有提供文件管理系統，所以，它必須和其他的分布式文件系統進行集成才能運作。我們可以選擇Hadoop的HDFS,也可以選擇其他的基于云的數據系統平臺。但Spark默認來說還是被用在Hadoop上面的，畢竟大家都認為它們的結合是最好的。

二、安裝Spark

以下都是在Ubuntu 16.04 系統上的操作

1.安裝Java JDK并配置好環境變量（這部分就不詳細說了）

2.安裝Hadoop

    2.1 創建Hadoop用戶：

打開終端，輸入命令：

sudo useradd -m hadoop -s /bin/bash

添加hadoop用戶，并設置/bin/bash作為shell

    2.2 設置Hadoop用戶的登錄密碼：

sudo passwd hadoop

然后根據系統提示輸入兩次自己的密碼，再給hadoop用戶添加管理員權限：

sudo adduser hadoop sudo

    2.3 將當前的用戶切換到剛剛創建的hadoop用戶（屏幕右上角有個齒輪，點進去就看到）

    2.4 更新系統的apt。打開終端，輸入命令：

sudo apt-get update

    2.5 安裝ssh、配置ssh無密碼登錄

集群、單節點模式都需要遇到SSH登錄。Ubuntu默認安裝了SSH Client，但需要自己安裝SSH Server：

sudo apt-get install openssh-server

安裝后，直接登錄本機

ssh localhost

SSH首次登錄需要確認，根據提示輸入：yes，然后再按提示輸入剛剛設置的hadoop的密碼，就登錄了。

    2.6 下載Hadoop

下載地址：http://mirror.bit.edu.cn/apache/hadoop/common/

選擇“stable”文件夾，點擊下載“hadoop-2.x.y.tar.gz”文件。默認會下載到“下載”目錄中，

在該文件夾下打開終端，將該文件解壓到/usr/local文件中,執行命令：

sudo tar -zxf ~/hadoop-2.9.0.tar.gz
cd /usr/local/
sudo mv ./hadoop-2.9.0/ ./hadoop    #將文件名修改為hadoop
sudo chown -R hadoop ./hadoop    #修改文件權限

Hadoop的文件夾解壓之后就可以直接使用，檢查一下Hadoop是否可以正常使用，如果正常則顯示Hadoop的版本信息

cd /usr/local/hadoop
./bin/hadoop version

這里就初步完成了Hadoop的安裝，還有很多配置什么的用到的時候在寫，比如偽分布式系統配置。

3.安裝Spark

  3.1 下載Spark：http://spark.apache.org/downloads.html

第一項我選擇的版本是最新當前的最新版本：2.3.1，第二項選擇“Pre-build with user-provided Apache Hadoop”，然后點擊第三項后面的下載“spark-2.3.1-bin-without-hadoop-tgz”。

  3.2 解壓文件

這一步與Hadoop的解壓是一樣的，我們都把它解壓到/usr/local路徑下：

$ sudo tar -zxf ~/下載/spark-2.3.1-bin-without-hadoop.tgz -C /usr/local/
$ cd /usr/local
$ sudo mv ./spark-2.3.1-bin-without-hadoop/ ./spark
$ sudo chown -R hadoop:hadoop ./spark

   3.3 設置環境變量

執行如下命令拷貝一個配置文件：

$ cd /usr/local/spark
$ ./conf/spark-env.sh.template ./conf/spark-env.sh

然后編輯spark-env.sh：

$ vim ./conf/spark-env.sh

打開之后在這個文件的最后一行加上下面的內容：

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

     然后保存退出Vim，就可以使用Spark了。

   三、Spark入門示例

    在文件路徑“/usr/local/spark/examples/src/main”中，我們可以找到spark自帶的一些實例，如下圖可以看到Spark支持Scala、Python、Java、R語言等。

    1.Spark最簡單的方式就是使用交互式命令行提示符。打開PySpark終端，在命令行中打出pyspark：   

   ~$ pyspark

    2.PySpark將會自動使用本地Spark配置創建一個SparkContext。我們可以通過sc變量來訪問它，來創建第一個RDD：

>>>text=sc.textFile(“file\\\usr\local\spark\exp\test1.txt")
>>>print text

    3.轉換一下這個RDD，來進行分布式計算的“hello world”：“字數統計”

    首先導入了add操作符，它是個命名函數，可以作為加法的閉包來使用。我們稍后再使用這個函數。首先我們要做的是把文本拆分為單詞。我們創建了一個tokenize函數，參數是文本片段，返回根據空格拆分的單詞列表。然后我們通過給flatMap操作符傳遞tokenize閉包對textRDD進行變換創建了一個wordsRDD。你會發現，words是個PythonRDD，但是執行本應該立即進行。顯然，我們還沒有把整個數據集拆分為單詞列表。

    4.將每個單詞映射到一個鍵值對，其中鍵是單詞，值是1，然后使用reducer計算每個鍵的1總數

>>> wc = words.map(lambda x: (x,1))
>>> print wc.toDebugString()

    我使用了一個匿名函數（用了Python中的lambda關鍵字）而不是命名函數。這行代碼將會把lambda映射到每個單詞。因此，每個x都是一個單詞，每個單詞都會被匿名閉包轉換為元組(word, 1)。為了查看轉換關系，我們使用toDebugString方法來查看PipelinedRDD是怎么被轉換的?？梢允褂胷educeByKey動作進行字數統計，然后把統計結果寫到磁盤。

    5.使用reduceByKey動作進行字數統計，然后把統計結果寫到磁盤

>>> counts = wc.reduceByKey(add)
>>> counts.saveAsTextFile("wc")

   一旦我們最終調用了saveAsTextFile動作，這個分布式作業就開始執行了，在作業“跨集群地”（或者你本機的很多進程）運行時，你應該可以看到很多INFO語句。如果退出解釋器，你可以看到當前工作目錄下有個“wc”目錄。每個part文件都代表你本機上的進程計算得到的被保持到磁盤上的最終RDD。

    四、Spark數據形式

4.1 彈性分布式數據集(RDD)

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

獲得RDD的三種方式：

Parallelize：將一個存在的集合，變成一個RDD，這種方式試用于學習spark和做一些spark的測試

 >>>sc.parallelize(['cat','apple','bat’])

MakeRDD：只有scala版本才有此函數，用法與parallelize類似

textFile：從外部存儲中讀取數據來創建 RDD

>>>sc.textFile(“file\\\usr\local\spark\README.md”)

    RDD的兩個特性：不可變；分布式。

    RDD支持兩種操作；Transformation（轉化操作：返回值還是RDD）如map(),filter()等。這種操作是lazy(惰性)的，即從一個RDD轉換生成另一個RDD的操作不是馬上執行，只是記錄下來，只有等到有Action操作是才會真正啟動計算，將生成的新RDD寫到內存或hdfs里，不會對原有的RDD的值進行改變；Action（行動操作：返回值不是RDD）會實際觸發Spark計算，對RDD計算出一個結果，并把結果返回到內存或hdfs中，如count(),first()等。

4.2 RDD的緩存策略

   Spark最為強大的功能之一便是能夠把數據緩存在集群的內存里。這通過調用RDD的cache函數來實現：rddFromTextFile.cache，

   調用一個RDD的cache函數將會告訴Spark將這個RDD緩存在內存中。在RDD首次調用一個執行操作時，這個操作對應的計算會立即執行，數據會從數據源里讀出并保存到內存。因此，首次調用cache函數所需要的時間會部分取決于Spark從輸入源讀取數據所需要的時間。但是，當下一次訪問該數據集的時候，數據可以直接從內存中讀出從而減少低效的I/O操作，加快計算。多數情況下，這會取得數倍的速度提升。

Spark的另一個核心功能是能創建兩種特殊類型的變量：廣播變量和累加器。廣播變量（broadcast variable）為只讀變量，它由運行SparkContext的驅動程序創建后發送給會參與計算的節點。對那些需要讓各工作節點高效地訪問相同數據的應用場景，比如機器學習，這非常有用。Spark下創建廣播變量只需在SparkContext上調用一個方法即可：

>>> broadcastAList = sc.broadcast(list(["a", "b", "c", "d", "e"]))