IBM專家親自解讀 Spark2.0 操作指南

---

Spark 背景介紹

1、什么是Spark

在Apache的網(wǎng)站上，有非常簡單的一句話，”Spark is a fast and general engine ”,就是Spark是一個(gè)統(tǒng)一的計(jì)算引擎，而且突出了fast。那么具體是做什么的呢？是做large-scale的processing，即大數(shù)據(jù)的處理。

“Spark is a fast and general engine for large-scale processing”這句話非常簡單，但是它突出了Spark的一些特點(diǎn):第一個(gè)特點(diǎn)就是spark是一個(gè)并行式的、內(nèi)存的計(jì)算密集型的計(jì)算引擎。

那么來說內(nèi)存的，因?yàn)镾park是基于Map Reduce的，但是它的空間數(shù)據(jù)不是存在于HDFS上面，而是存在于內(nèi)存中，所以他是一個(gè)內(nèi)存式的計(jì)算，這樣就導(dǎo)致Spark的計(jì)算速度非常得快，同時(shí)它可以部署在集群上，所以它可以分布到各個(gè)的節(jié)點(diǎn)上，并行式地計(jì)算；Spark上還有很多機(jī)器學(xué)習(xí)和數(shù)據(jù)挖掘的學(xué)習(xí)包，用戶可以利用學(xué)習(xí)包進(jìn)行數(shù)據(jù)的迭代式計(jì)算，所以它又是一個(gè)計(jì)算密集型的計(jì)算工具。

2、Spark的發(fā)展歷程

了解完什么是Spark之后，我們看一下Spark的發(fā)展歷程。

Spark 2009年作為研究項(xiàng)目創(chuàng)建，13年成為Apache的孵化項(xiàng)目，14年成為Apache的頂級項(xiàng)目，Spark2.0還沒有正式發(fā)布，目前只有比較draft的版本。

3、Spark2.0的最新特性

Spark2.0是剛出的，今天主要講解它的兩個(gè)部分，一個(gè)是它的new feature，就是它有哪些新的特性；另一部分是community，大家知道Spark是一個(gè)開源社區(qū)，社區(qū)對Spark的發(fā)展功不可沒。

在feature這一部分，可以看到，Spark2.0中有比較重要的兩個(gè)部分，其中一個(gè)就是Structured API。

Spark2.0統(tǒng)一了DataFrame和Dataset，并且引入了新的SparkSession。SparkSession提供了一個(gè)新的切入點(diǎn)，這個(gè)切入點(diǎn)統(tǒng)一了sql和sql context，對用戶來說是透明的，用戶不需要再去區(qū)分用什么context或者怎么創(chuàng)建，直接用SparkSession就可以了。還有一個(gè)是結(jié)構(gòu)化的流，streaming。在Spark2.0中，流和bash做了一個(gè)統(tǒng)一，這樣的話對用戶來說也是透明的，就不在區(qū)分什么是流處理，什么是批量處理的數(shù)據(jù)了。

后面幾個(gè)特性，比如MLlib，相信對data scientists非常有吸引力。MLlib可以將用戶訓(xùn)練過的模型存儲下來，等需要的時(shí)候再導(dǎo)入所需要的訓(xùn)練模型；從R上來說，原來SparkR上支持的只是單機(jī)單節(jié)點(diǎn)的，不支持分布式的計(jì)算，但是R的分布式的開發(fā)在Spark2.0中是非常有力的feature。此外，在Spark2.0中，SQL 2003的support可以讓Spark在對結(jié)構(gòu)化的數(shù)據(jù)進(jìn)行處理的時(shí)候，基本上支持了所有的SQL語句。

4、為什么使用Spark

在傳統(tǒng)方法中，MapReduce需要大量的磁盤I/O，從對比圖中可以看到，MapReduce會將大量的數(shù)據(jù)存在HDFS上，而Spark因?yàn)槭莾?nèi)存式的，就不需要大量的磁盤I/O，這一塊就會非常快。

性能方面，在通用的任務(wù)上，Spark可以提高20-100倍的速度，因此Spark性能的第一點(diǎn)就是快；第二個(gè)就是比較高效，用過Scala開發(fā)程序的人應(yīng)該有感受，Spark語法的表達(dá)非常強(qiáng)大，原來可能用十行去描述一段匹配的代碼，Scala可能一行就可以做到，所以它效率非常地高，包括它也支持一些主流的編程的語言，java，Python，Scala，還有R等。

此外，Spark2.0可以利用已有的資產(chǎn)。大家知道hadoop的生態(tài)系統(tǒng)是非常有吸引力的，Spark可以很好地和hadoop的生態(tài)系統(tǒng)結(jié)合在一起。上面我們提到了社區(qū)的貢獻(xiàn)，社區(qū)的貢獻(xiàn)者不斷得對Spark進(jìn)行 improvement，使得Spark的發(fā)展越來越好，而且速度越來愈快。

以上這些特點(diǎn)導(dǎo)致了Spark現(xiàn)在越來越流行，更多的data scientists包括學(xué)者都愿意去使用Spark，Spark讓大數(shù)據(jù)的計(jì)算更簡單，更高效，更智能。

5、IBM對Spark的支持

IBM內(nèi)部對Spark也是越來越重視，主要支持力度體現(xiàn)在社區(qū)培育、產(chǎn)品化和Spark Core上。社區(qū)方面，big data university的在線課程內(nèi)容十分豐富，包括數(shù)據(jù)科學(xué)家、包括最基礎(chǔ)的語言的開發(fā)，包括Spark、Hadoop生態(tài)基礎(chǔ)的培訓(xùn)都很多，所以它培訓(xùn)了超過了一百萬的數(shù)據(jù)科學(xué)家，并且贊助了AMP Lab，AMP Lab就是Spark開源社區(qū)的開發(fā)者。

第二個(gè)就是對Spark Core的貢獻(xiàn)，因?yàn)樵贗BM內(nèi)部，已經(jīng)建立了Spark技術(shù)中心，超過了300名的工程師在進(jìn)行Spark Core的開發(fā)。并且IBM開源的機(jī)器學(xué)習(xí)庫，也成為了databricks的合作伙伴。

產(chǎn)品方面，在CDL就有一些Spark產(chǎn)品，集成到IBM本身的AOP環(huán)境里面，（注：AOP也是一個(gè)開源的軟件包），包括Big Insight里面都集成了Spark的元素，IBM總共投入了超過3500名的員工在Spark的相關(guān)工作上。

Spark 基礎(chǔ)

1、Spark核心組件

在Spark Build-in組件中，最基礎(chǔ)的就是Spark Core，它是所有應(yīng)用程序架構(gòu)的基礎(chǔ)。SparkSQL、Spark Streaming、MLLib、GraphX都是Spark Build-in組件提供的應(yīng)用組件的子架構(gòu)。

SparkSQL是對結(jié)構(gòu)化數(shù)據(jù)的處理，Spark Streaming是對實(shí)時(shí)流數(shù)據(jù)的處理 ，MLLib就是對機(jī)器學(xué)習(xí)庫的處理，GraphX是對并行圖計(jì)算的處理。

不管是哪一個(gè)應(yīng)用上的子架構(gòu)，它都是基于RDD上的應(yīng)用框架。實(shí)際上用戶可以基于RDD來開發(fā)出不同領(lǐng)域上的子框架，運(yùn)用Spark Build-in組件來執(zhí)行。

2、Spark應(yīng)用程序的架構(gòu)

在每一個(gè)Spark應(yīng)用程序中，只有一個(gè)Driver Program，和若干個(gè)Executor。大家可以看到右邊的Work Node，我們可以認(rèn)為Work Node就是一個(gè)物理機(jī)器，所有的應(yīng)用程序都是從Driver開始的，Driver Program會先初始化一個(gè)SparkContext，作為應(yīng)用程序的入口，每一個(gè)Spark應(yīng)用程序只有一個(gè)SparkContext。SparkContext作為入口，再去初始化一些作業(yè)調(diào)度和任務(wù)調(diào)度，通過Cluster Manager將任務(wù)分配到各個(gè)節(jié)點(diǎn)上，由Worker Node上面的執(zhí)行器來執(zhí)行任務(wù)。一個(gè)Spark應(yīng)用程序有多個(gè)Executor，一個(gè)Executor上又可以執(zhí)行多個(gè)task，這就是Spark并行計(jì)算的框架。

此外，Executor除了可以處理task，還可以將數(shù)據(jù)存在Cache或者HDFS上面。

3、Spark運(yùn)行模式

一般我們看到的是下圖中的前四種Spark運(yùn)行模式：Local、standalone、Yarn和Mesos。Cloud就是一種外部base的Spark的運(yùn)行環(huán)境。

Local就是指本地的模式，用戶可以在本地上執(zhí)行Spark程序，Local[N]即指的是使用多少個(gè)線程;Standalone是Spark自己自帶的一個(gè)運(yùn)行模式，需要用戶自己去部署spark到相關(guān)的節(jié)點(diǎn)上;Yarn和Mesos是做資源管理的，它也是Hadoop生態(tài)系統(tǒng)里面的，如果使用Yarn和Mesos，那么就是這兩者去做資源管理，Spark來做資源調(diào)度。

不管是那種運(yùn)行模式，它都還細(xì)分為兩種，一種是client模式：一種是cluster模式，那么怎么區(qū)分這兩種模式呢？可以用到架構(gòu)圖中的Driver Program。Driver Program如果在集群里面，那就是cluster模式；如果在集群外面，那就是client模式。

4、彈性分布式數(shù)據(jù)集RDD

RDD有幾個(gè)特點(diǎn)，一是它不可變，二是它被分區(qū)。我們在java或者C++里，所用的基本數(shù)據(jù)集、數(shù)組都可以被更改，但是RDD是不能被更改的，它只能產(chǎn)生新的RDD，也就是說Scala是一種函數(shù)式的編程語言。函數(shù)式的編程語言不主張就地更改現(xiàn)有的所有的數(shù)據(jù)，而是在已有的數(shù)據(jù)上產(chǎn)生一個(gè)新的數(shù)據(jù)，主要是做transform的工作，即映射的工作。

RDD不可更改，但可以分布到不同的Partition上，對用戶來說，就實(shí)現(xiàn)了以操作本地集合的方式來操作分布式數(shù)據(jù)集的抽象實(shí)現(xiàn)。RDD本身是一個(gè)抽象的概念，它不是真實(shí)存在的，那么它分配到各個(gè)節(jié)點(diǎn)上，對用戶來說是透明的，用戶只要按照自己操作本地?cái)?shù)據(jù)集的方法去操作RDD就可以了，不用管它是怎么分配到各個(gè)Partition上面的。

在操作上，RDD主要有兩種方式，一種是Transform，一種是Action。Transform的操作呢，就是將一個(gè)RDD轉(zhuǎn)換成一個(gè)新的RDD，但是它有個(gè)特點(diǎn)，就是延遲執(zhí)行；第二種操作就是Action，用戶要么寫數(shù)據(jù)，要么給應(yīng)用程序返回某些信息。當(dāng)你執(zhí)行Action的時(shí)候，Transform才會被觸發(fā)，這也就是延遲執(zhí)行的意思。

看一下右邊的代碼，這是一個(gè)Scala的代碼，在第一行，它去創(chuàng)建了一個(gè)Spark的Context，去讀一個(gè)文件。然后這個(gè)文件做了三個(gè)操作，第一個(gè)是map，第二個(gè)是filter，第三個(gè)是save，前面兩個(gè)動作就是一個(gè)Transform，map的意思就是映射，filter就是過濾，save就是寫。當(dāng)我”寫”的這個(gè)程度執(zhí)行到map和filter這一步時(shí)，它不會去執(zhí)行，等我的save動作開始的時(shí)候，它才會執(zhí)行去前面兩個(gè)。

5、Spark程序的執(zhí)行

了解了RDD和Spark運(yùn)行原理之后，我們來從整體看一下Spark程序是怎么執(zhí)行的。

還是之前的三行代碼，前兩步是Transform，最后一步是Action。那么這一系列RDD就做一系列的Transform，從第一步開始轉(zhuǎn)；DAG就是一個(gè)調(diào)度器，Spark context會初始化一個(gè)任務(wù)調(diào)度器，任務(wù)調(diào)度器就會將RDD的一系列轉(zhuǎn)換切分成不同的階段，由任務(wù)調(diào)度器將不同的階段上分成不同的task set，通過Cluster Manager去調(diào)度這些task，把這些task set分布到不同的Executor上去執(zhí)行。

6、Spark DataFrame

很多人會問，已經(jīng)有RDD，為什么還要有DataFrame？DataFrame API是2015年發(fā)布的，Spark1.3之后就有，它是以命名列的方式去組織分布式的數(shù)據(jù)集。

Spark上面原來主要是為了big data，大數(shù)據(jù)平臺，它很多都是非結(jié)構(gòu)化數(shù)據(jù)。非結(jié)構(gòu)化數(shù)據(jù)需要用戶自己去組織映射，而DataFrame就提供了一些現(xiàn)成的，用戶可以通過操作關(guān)系表去操作大數(shù)據(jù)平臺上的數(shù)據(jù)。這樣很多的data scientists就可以使用原來使關(guān)系數(shù)據(jù)庫的只是和方式去操作大數(shù)據(jù)平臺上的數(shù)據(jù)。

DataFrame支持的數(shù)據(jù)源也很多，比如說JSON、Hive、JDBC等。

DataFrame還有存在的另外一個(gè)理由：我們可以分析上表，藍(lán)色部分代表著RDD去操縱不同語言的同樣數(shù)量集時(shí)的性能。可以看到，RDD在Python上的性能比較差，Scala的性能比較好一些。但是從綠色的部分來看，用DataFrame來編寫程序的時(shí)候，他們的性能是一樣的，也就是說RDD在操作不同的語言時(shí)，性能表現(xiàn)不一樣，但是用DataFrame去操作時(shí)，性能表現(xiàn)是一樣的，并且性能總體要高于RDD。

下面是DataFrame的一個(gè)簡單示例。

右邊同樣是用Scala寫的一段代碼，這就是sqlContext，因?yàn)樗С諮SON文件，直接點(diǎn)JsonFile，讀進(jìn)來這個(gè)json文件。下面直接對這個(gè)DataFrame

df.groupBy(“ages”).count().show()，show出來的方式就是一個(gè)表的方式。這個(gè)操作就很簡單，用戶不用再做map操作，如果是用RDD操作的話，用戶需要自己對數(shù)列里的每一塊數(shù)據(jù)作處理。

7、Spark編程語言

在編程語言上，Spark目前支持的有以下四種：

8、Spark使用方式

使用上，如果本地有Spark集群，就有兩種操作方式：一種是用Spark-shell，即交互式命令行；交互式的命令操作很簡單，就和java一樣，一行一行敲進(jìn)去，它會交互式地告訴你，一行一行包括的是什么；這個(gè)地方也可以把一段代碼復(fù)制過去，邊運(yùn)行邊調(diào)試。一般來講，交互式命令用Local模式就可以了。

第二種是直接用Spark-submit，一般在開發(fā)工程項(xiàng)目時(shí)使用較多；Spark-submit有幾個(gè)必要的參數(shù)，一個(gè)是master，就是運(yùn)行模式必須有；還有幾個(gè)參數(shù)也必須有，比如class，java包的位置等。此外可以根據(jù)Spark-submit后面的help命令，來查看submit有多少參數(shù)，每個(gè)參數(shù)是什么意思。

此外可以通過Web-based NoteBook來使用Spark，在IBM的workbench上提供了Jupyter和Zepplin兩種NoteBook的方式。

詳情請咨詢！

客服熱線：023-66090381