日韩偷拍一区二区,国产香蕉久久精品综合网,亚洲激情五月婷婷,欧美日韩国产不卡

第 9 章 Spark技術(shù)

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

9.1 Spark框架

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

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

圖9-1 Spark 框架

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

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

Spark應(yīng)用程序在集群上以獨(dú)立進(jìn)程集合的形式運(yùn)行。如圖9-2所示，主程序(叫做Driver程序)中的SparkContext對(duì)象協(xié)調(diào)Spark應(yīng)用程序。SparkContext對(duì)象首先連接到多種集群管理器(如：YARN)，然后在集群節(jié)點(diǎn)上獲得Executor。SparkContext把應(yīng)用代碼發(fā)給Executor，Executor負(fù)責(zé)應(yīng)用程序的計(jì)算和數(shù)據(jù)存儲(chǔ)。

圖9-2 集群模式

每個(gè)應(yīng)用程序都擁有自己的Executor。Executor為應(yīng)用程序提供了一個(gè)隔離的運(yùn)行環(huán)境，以Task的形式執(zhí)行作業(yè)。對(duì)于Spark Shell來說，這個(gè)Driver就是與用戶交互的進(jìn)程。

9.1.1 安裝Spark

近期的Spark版本是1.6.1。它可以運(yùn)行在Windows或Linux機(jī)器上。運(yùn)行 Spark 需要 Java JDK 1.7，CentOS 6.x 系統(tǒng)默認(rèn)只安裝了 Java JRE，還需要安裝 Java JDK，并確保配置好 JAVA_HOME、PATH和CLASSPATH變量。此外，Spark 會(huì)用到 HDFS 與 YARN，因此讀者要先安裝好 Hadoop。我們可以從Spark官方網(wǎng)站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”版，可應(yīng)用到任意 Hadoop 版本。

l Pre-build for Hadoop 2.6 and later：基于 Hadoop 2.6 的預(yù)編譯版，需要與本機(jī)安裝的 Hadoop 版本對(duì)應(yīng)。可選的還有 Hadoop 2.4 and later、Hadoop 2.3、Hadoop 1.x，以及 CDH 4。

本書選擇的是 Pre-build with user-provided Hadoop，簡(jiǎn)單配置后可應(yīng)用到任意 Hadoop 版本。下載后，執(zhí)行如下命令進(jìn)行安裝：

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

安裝后，進(jìn)入conf目錄，以spark-env.sh.template文件為模塊創(chuàng)建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 就可以啟動(dòng)和運(yùn)行了。在 ./examples/src/main 目錄下有一些 Spark 的示例程序，有 Scala、Java、Python、R 等語言的版本。我們可以先運(yùn)行一個(gè)示例程序 SparkPi(即計(jì)算 π 的近似值)，執(zhí)行如下命令：

cd /usr/local/spark

./bin/run-example SparkPi

執(zhí)行時(shí)會(huì)輸出非常多的運(yùn)行信息，輸出結(jié)果不容易找到，可以通過 grep 命令進(jìn)行過濾(命令中的 2>&1 可以將所有的信息都輸出到 stdout 中)：

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

過濾后的運(yùn)行結(jié)果為 π 的 5 位小數(shù)近似值 。

9.2 Spark Shell

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

Spark Shell 提供了簡(jiǎn)單的方式來學(xué)習(xí) Spark API，也提供了交互的方式來分析數(shù)據(jù)。Spark Shell 支持 Scala 和 Python，本書選擇使用 Scala 來進(jìn)行介紹。Scala集成了面向?qū)ο蠛秃瘮?shù)語言的特性，并運(yùn)行于Java 虛擬機(jī)之上，兼容現(xiàn)有的 Java 程序。Scala 是 Spark 的主要編程語言，如果僅僅是寫 Spark 應(yīng)用，并非一定要用 Scala，用Java和Python都是可以的。使用 Scala 的優(yōu)勢(shì)是開發(fā)效率更高，代碼更精簡(jiǎn)，并且可以通過 Spark Shell 進(jìn)行交互式實(shí)時(shí)查詢，方便排查問題。執(zhí)行如下命令啟動(dòng) Spark Shell：

./bin/spark-shell

啟動(dòng)成功后會(huì)有“scala >”的命令提示符。這表明已經(jīng)成功啟動(dòng)了Spark Shell。在 Spark Shell 啟動(dòng)時(shí)，輸出日志的有這么幾條信息：

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

Spark context available as sc.

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

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

上述的sc是Spark創(chuàng)建的SparkContext，我們使用SparkContext對(duì)象加載本地文件README.md來創(chuàng)建RDD。輸出結(jié)果如下：

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

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

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

l actions：在數(shù)據(jù)集上運(yùn)行計(jì)算后返回結(jié)果值。

l transformations：轉(zhuǎn)換。從現(xiàn)有RDD創(chuàng)建一個(gè)新的RDD。

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

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

輸出結(jié)果為：

res0: Long = 95

scala>textFile.first() // RDD 中的及時(shí)個(gè) item，對(duì)于文本文件，就是及時(shí)行內(nèi)容

輸出結(jié)果為：

res1: String = # Apache Spark

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

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

scala>linesWithSpark.count() // 統(tǒng)計(jì)行數(shù)

上面的linesWithSpark RDD有多個(gè)分區(qū)，每個(gè)分區(qū)上只有包含了Spark的若干文本行。輸出結(jié)果為：

res4: Long = 17

上述結(jié)果表明一共有17行內(nèi)容包含“Spark”，這與通過 Linux 命令 cat ./README.md | grep "Spark" -c 得到的結(jié)果一致，說明是正確的。action 和 transformation 可以用鏈?zhǔn)讲僮鞯姆绞浇Y(jié)合使用，使代碼更為簡(jiǎn)潔：

scala>textFile.filter(line => line.contains("Spark")).count() // 統(tǒng)計(jì)包含 Spark 的行數(shù)

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

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

輸出結(jié)果為：

res5: Int = 14

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

scala>import java.lang.Math

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

詞頻統(tǒng)計(jì)(WordCount)是Hadoop MapReduce的入門程序，Spark可以更容易地實(shí)現(xiàn)。首先結(jié)合flatMap、map和reduceKey來計(jì)算文件中每個(gè)單詞的詞頻：

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

輸出結(jié)果為(string，int)類型的鍵值對(duì)ShuffledRDD。這是因?yàn)閞educeByKey操作需要進(jìn)行Shuffle操作，返回的是一個(gè)Shuffle形式的ShuffleRDD：

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

然后使用collect聚合單詞計(jì)算結(jié)果：

scala>wordCounts.collect

輸出結(jié)果為：

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

Spark 支持將數(shù)據(jù)緩存在集群的內(nèi)存緩存中，當(dāng)數(shù)據(jù)需要反復(fù)訪問時(shí)這個(gè)特征非常有用。調(diào)用 cache()，就可以將數(shù)據(jù)集進(jìn)行緩存：

scala>textFilter.cache

9.3 Spark編程

無論Windows或Linux操作系統(tǒng)，都是基于Eclipse或Idea構(gòu)建開發(fā)環(huán)境，通過Java、Scala或Python語言進(jìn)行開發(fā)。根據(jù)開發(fā)語言的不同，我們需要預(yù)先準(zhǔn)備好JDK、Scala或Python環(huán)境，然后在Eclipse中下載安裝Scala或Python插件。

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

在終端中執(zhí)行如下命令，創(chuàng)建一個(gè)文件夾 sparkapp 作為應(yīng)用程序根目錄：

cd ~ # 進(jìn)入用戶主文件夾

mkdir ./sparkapp # 創(chuàng)建應(yīng)用程序根目錄

mkdir -p ./sparkapp/src/main/scala # 創(chuàng)建所需的文件夾結(jié)構(gòu)

9.3.1 編寫Spark API程序

在./sparkapp/src/main/scala下建立一個(gè)名為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 {

//使用關(guān)鍵字def聲明函數(shù)，必須為函數(shù)指定參數(shù)類型

def main(args: Array[String]) {

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

//創(chuàng)建SparkConf對(duì)象，該對(duì)象包含應(yīng)用程序的信息

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

//創(chuàng)建SparkContext對(duì)象，該對(duì)象可以訪問Spark集群

val sc = new SparkContext(conf

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

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

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

}

}

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

9.3.2 使用sbt編譯并打成jar包

該程序依賴 Spark API，因此我們需要通過sbt(或mvn)進(jìn)行編譯打包。我們以sbt為例，創(chuàng)建一個(gè)包含應(yīng)用程序代碼的jar包。在 ./sparkapp 中新建文件 simple.sbt(vim ./sparkapp/simple.sbt)，添加如下內(nèi)容，聲明該獨(dú)立應(yīng)用程序的信息以及與 Spark 的依賴關(guān)系：

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獲得。我們啟動(dòng)Spark Shell的過程中，當(dāng)輸出到 Spark 的符號(hào)圖形時(shí)，可以看到相關(guān)的版本信息。

Spark中沒有自帶sbt，需要手動(dòng)安裝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 中創(chuàng)建 sbt 腳本(vim ./sbt)，添加如下內(nèi)容：

#!/bin/bash

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

java $SBT_OPTS -jar `dirname