日本免费精品视频,男人的天堂在线免费视频,成人久久久精品乱码一区二区三区,高清成人爽a毛片免费网站

第 9 章 Spark技術

Apache Spark 是一個新興的大數據處理通用引擎，提供了分布式的內存抽象。Spark較大的特點就是快(Lightning-Fast)，可比 Hadoop MapReduce 的處理速度快 100 倍。此外，Spark 提供了簡單易用的 API，幾行代碼就能實現 WordCount。本章介紹Spark 的框架，Spark Shell 、RDD、Spark SQL、Spark Streaming 等的基本使用。

9.1 Spark框架

Spark作為新一代大數據快速處理平臺，集成了大數據相關的各種能力。Hadoop的中間數據需要存儲在硬盤上，這產生了較高的延遲。而Spark基于內存計算，解決了這個延遲的速度問題。Spark本身可以直接讀寫Hadoop上任何格式數據，這使得批處理更加快速。

圖9-1是以Spark為核心的大數據處理框架。最底層為大數據存儲系統，如：HDFS、HBase等。在存儲系統上面是Spark集群模式(也可以認為是資源管理層)，這包括Spark自帶的獨立部署模式、YARN和Mesos集群資源管理模式，也可以是Amazon EC2。Spark內核之上是為應用提供各類服務的組件。Spark內核API支持Java、Python、Scala等編程語言。Spark Streaming提供高性、高吞吐量的實時流式處理服務，能夠滿足實時系統要求；MLib提供機器學習服務，Spark SQL提供了性能比Hive快了很多倍的SQL查詢服務，GraphX提供圖計算服務。

圖9-1 Spark 框架

從上圖看出，Spark有效集成了Hadoop組件，可以基于Hadoop YARN作為資源管理框架，并從HDFS和HBase數據源上讀取數據。YARN是Spark目前主要使用的資源管理器。Hadoop能做的，Spark基本都能做，而且做的比Hadoop好。Spark依然是Hadoop生態圈的一員，它替換的主要是MR的計算模型而已。資源調度依賴于YARN，存儲則依賴于HDFS。

Spark的大數據處理平臺是建立在統一抽象的RDD之上。RDD是彈性分布式數據集(Resilient Distributed Dataset)的英文簡稱，它是一種特殊數據集合，支持多種來源，有容錯機制，可以被緩存，支持并行操作。Spark的一切都是基于RDD的。RDD就是Spark輸入的數據。

Spark應用程序在集群上以獨立進程集合的形式運行。如圖9-2所示，主程序(叫做Driver程序)中的SparkContext對象協調Spark應用程序。SparkContext對象首先連接到多種集群管理器(如：YARN)，然后在集群節點上獲得Executor。SparkContext把應用代碼發給Executor，Executor負責應用程序的計算和數據存儲。

圖9-2 集群模式

每個應用程序都擁有自己的Executor。Executor為應用程序提供了一個隔離的運行環境，以Task的形式執行作業。對于Spark Shell來說，這個Driver就是與用戶交互的進程。

9.1.1 安裝Spark

近期的Spark版本是1.6.1。它可以運行在Windows或Linux機器上。運行 Spark 需要 Java JDK 1.7，CentOS 6.x 系統默認只安裝了 Java JRE，還需要安裝 Java JDK，并確保配置好 JAVA_HOME、PATH和CLASSPATH變量。此外，Spark 會用到 HDFS 與 YARN，因此讀者要先安裝好 Hadoop。我們可以從Spark官方網站spark.apache.org/downloads.html上下載Spark，如圖9-3所示。

圖9-3 下載安裝包

有幾種Package type，分別為：

l Source code：Spark 源碼，需要編譯才能使用。

l Pre-build with user-provided Hadoop：“Hadoop free”版，可應用到任意 Hadoop 版本。

l 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.1-bin-without-hadoop.tgz -C /usr/local/

cd /usr/local

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

sudo chown -R hadoop:hadoop ./spark

9.1.2 配置Spark

安裝后，進入conf目錄，以spark-env.sh.template文件為模塊創建spark-env.sh文件，然后修改其配置信息，命令如下：

cd /usr/local/spark

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

編輯 ./conf/spark-env.sh(vim ./conf/spark-env.sh)，在文件的加上如下一行：

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

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

cd /usr/local/spark

./bin/run-example SparkPi

執行時會輸出非常多的運行信息，輸出結果不容易找到，可以通過 grep 命令進行過濾(命令中的 2>&1 可以將所有的信息都輸出到 stdout 中)：

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

過濾后的運行結果為 π 的 5 位小數近似值 。

9.2 Spark Shell

以前的統計和機器學習依賴于數據抽樣。從統計的角度來看，抽樣如果足夠隨機，其實可以很精準地反應全集的結果，但事實上往往很難做到隨機，所以通常做出來也會不準。現在大數據解決了這個問題，它不是通過優化抽樣的隨機來解決，而是通過全量數據來解決。要解決全量的數據就需要有強大的處理能力，Spark首先具備強大的處理能力，其次Spark Shell帶來了即席查詢。做算法的工程師，以前經常是在小數據集上跑個單機，然后看效果不錯，一到全量上，就可能和單機效果很不一樣。有了Spark后就不一樣了，尤其是有了Spark Shell。可以邊寫代碼，邊運行，邊看結果。Spark提供了很多的算法，最常用的是貝葉斯、word2vec、線性回歸等。作為算法工程師，或者大數據分析師，一定要學會用Spark Shell。

Spark Shell 提供了簡單的方式來學習 Spark API，也提供了交互的方式來分析數據。Spark Shell 支持 Scala 和 Python，本書選擇使用 Scala 來進行介紹。Scala集成了面向對象和函數語言的特性，并運行于Java 虛擬機之上，兼容現有的 Java 程序。Scala 是 Spark 的主要編程語言，如果僅僅是寫 Spark 應用，并非一定要用 Scala，用Java和Python都是可以的。使用 Scala 的優勢是開發效率更高，代碼更精簡，并且可以通過 Spark Shell 進行交互式實時查詢，方便排查問題。執行如下命令啟動 Spark Shell：

./bin/spark-shell

啟動成功后會有“scala >”的命令提示符。這表明已經成功啟動了Spark Shell。在 Spark Shell 啟動時，輸出日志的有這么幾條信息：

16/04/16 17:25:47 INFO repl.SparkILoop: Created spark context..

Spark context available as sc.

這些信息表明 SparkContext已經初始化好了，可通過對應的sc變量直接進行訪問。Spark 的主要抽象是分布式的數據集合RDD，它可被分發到集群各個節點上，進行并行操作。一個RDD可以通過 Hadoop InputFormats 創建(如 HDFS)，或者從其他 RDDs轉化而來。下面我們從 ./README 文件新建一個 RDD，代碼如下：

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

上述的sc是Spark創建的SparkContext，我們使用SparkContext對象加載本地文件README.md來創建RDD。輸出結果如下：

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

上述返回結果為一個MapPartitionsRDD文件。需要說明的是，加載HDFS文件和本地文件都是使用textFile ，區別在于前綴“hdfs://”為HDFS文件，而“file:// ”為本地文件。上述代碼中通過“file://”前綴指定讀取本地文件，直接返回MapPartitionsRDD。Spark Shell默認方式是讀取HDFS中的文件。從HDFS讀取的文件先轉換為HadoopRDD，然后隱式轉換成MapPartitionsRDD。

上面的例子使用Spark中的文本文件README.md創建一個RDD textFile，文件中包含了若干文本行。將該文本文件讀入RDD textFile時，其中的文本行將被分區，以便能夠分發到集群中并行化操作。我們可以想象，RDD有多個分區，每個分區上有多行的文本內容。RDDs 支持兩種類型的操作：

l actions：在數據集上運行計算后返回結果值。

l transformations：轉換。從現有RDD創建一個新的RDD。

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

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

輸出結果為：

res0: Long = 95

scala>textFile.first() // RDD 中的及時個 item，對于文本文件，就是及時行內容

輸出結果為：

res1: String = # Apache Spark

上面這兩個例子都是action的例子。接著演示 transformation，通過 filter transformation來篩選出包含 Spark 的行，返回一個新的RDD，代碼如下：

scala>val linesWithSpark = textFile.filter(line => line.contains("Spark"

scala>linesWithSpark.count() // 統計行數

上面的linesWithSpark RDD有多個分區，每個分區上只有包含了Spark的若干文本行。輸出結果為：

res4: Long = 17

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

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

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

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

輸出結果為：

res5: Int = 14

上述代碼將每一行文本內容使用split進行分詞，并統計分詞后的單詞數。將每一行內容map為一個整數，這將創建一個新的 RDD，并在這個 RDD 中執行reduce操作，找到較大的數。map()、reduce()中的參數是Scala的函數字面量(function literals)，并且可以使用Scala/Java的庫。例如，通過使用 Math.max() 函數(需要導入Java的Math庫)，可以使上述代碼更容易理解：

scala>import java.lang.Math

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

詞頻統計(WordCount)是Hadoop MapReduce的入門程序，Spark可以更容易地實現。首先結合flatMap、map和reduceKey來計算文件中每個單詞的詞頻：

scala>val wordCounts = textFile.flatMap(line => line.split(" ")).map(word => (word, 1)).reduceByKey((a, b) => a b)

輸出結果為(string，int)類型的鍵值對ShuffledRDD。這是因為reduceByKey操作需要進行Shuffle操作，返回的是一個Shuffle形式的ShuffleRDD：

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

然后使用collect聚合單詞計算結果：

scala>wordCounts.collect

輸出結果為：

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

Spark 支持將數據緩存在集群的內存緩存中，當數據需要反復訪問時這個特征非常有用。調用 cache()，就可以將數據集進行緩存：

scala>textFilter.cache

9.3 Spark編程

無論Windows或Linux操作系統，都是基于Eclipse或Idea構建開發環境，通過Java、Scala或Python語言進行開發。根據開發語言的不同，我們需要預先準備好JDK、Scala或Python環境，然后在Eclipse中下載安裝Scala或Python插件。

下面我們通過一個簡單的應用程序 SimpleApp 來演示如何通過 Spark API 編寫一個獨立應用程序。不同于使用Spark Shell自動初始化的SparkContext，獨立應用程序需要自己初始化一個SparkContext，將一個包含應用程序信息的SparkConf對象傳遞給SparkContext構造函數。對于獨立應用程序，使用 Scala 編寫的程序需要使用 sbt 進行編譯打包，相應地，Java 程序使用 Maven 編譯打包，而 Python 程序通過 spark-submit 直接提交。

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

cd ~ # 進入用戶主文件夾

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

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

9.3.1 編寫Spark API程序

在./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聲明函數，必須為函數指定參數類型

def main(args: Array[String]) {

val logFile = "file:///usr/local/spark/README.md" // 一個本地文件

//創建SparkConf對象，該對象包含應用程序的信息

val conf = new SparkConf().setAppName("Simple Application"

//創建SparkContext對象，該對象可以訪問Spark集群

val sc = new SparkContext(conf

val logData = sc.textFile(logFile, 2).cache

//line=>line.contains(..)是匿名函數的定義，line是參數

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

}

}

上述程序計算 /usr/local/spark/README 文件中包含 “a” 的行數和包含 “b” 的行數。不同于 Spark Shell，獨立應用程序需要通過“val sc = new SparkContext(conf)”初始化 SparkContext，SparkContext 的參數 SparkConf 包含了應用程序的信息。

9.3.2 使用sbt編譯并打成jar包

該程序依賴 Spark API，因此我們需要通過sbt(或mvn)進行編譯打包。我們以sbt為例，創建一個包含應用程序代碼的jar包。在 ./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.1"

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

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

sudo mkdir /usr/local/sbt

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

cd /usr/local/sbt

下載sbt后，拷貝至 /usr/local/sbt 中。接著在 /usr/local/sbt 中創建 sbt 腳本(vim ./sbt)，添加如下內容：

#!/bin/bash

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

java $SBT_OPTS -jar `dirname