黎文陽

（四川大學(xué)計(jì)算機(jī)學(xué)院，成都　610065）

大數(shù)據(jù)處理模型Apache Spark研究

黎文陽

（四川大學(xué)計(jì)算機(jī)學(xué)院，成都610065）

0　引言

MapReduce計(jì)算模型在大規(guī)模數(shù)據(jù)分析領(lǐng)域已取得很大成績(jī)，并被很多公司廣泛采用。這些系統(tǒng)都是基于非循環(huán)的數(shù)據(jù)流模型，有很好的容錯(cuò)性，同時(shí)為開發(fā)人員提供了高級(jí)接口以便于編寫并行程序。目前這些系統(tǒng)能很容易地訪問集群中的計(jì)算資源，但是不能充分地利用分布式內(nèi)存，導(dǎo)致了對(duì)那些重用中間結(jié)果的應(yīng)用不是很有效。這些應(yīng)用的特點(diǎn)是在多個(gè)并行操作之間重用數(shù)據(jù)，例如機(jī)器學(xué)習(xí)中的PageRank算法、K-means聚類算法、邏輯回歸算法等迭代式算法。交互式的數(shù)據(jù)挖掘算法中也經(jīng)常重用數(shù)據(jù)。Spark計(jì)算模型剛好解決了這些問題，并且能在Hadoop集群下部署，訪問HDFS文件系統(tǒng)。Spark將分布式內(nèi)存抽象成彈性分布式數(shù)據(jù)集（Resilient Distributed Datasets，RDD）［1］。RDD支持基于工作集的應(yīng)用，同時(shí)具有數(shù)據(jù)流模型的特點(diǎn)：自動(dòng)容錯(cuò)、位置感知調(diào)度和可伸縮性。RDD允許用戶在執(zhí)行多個(gè)查詢時(shí)顯式地將工作集緩存在內(nèi)存中，以便后續(xù)的查詢能夠重用，這極大地提升了查詢速度。

1　Spark簡(jiǎn)介

Spark是UC Berkeley AMPLab于2009年發(fā)起的，然后被Apache軟件基金會(huì)接管的類Hadoop MapReduce通用性并行計(jì)算框架，是當(dāng)前大數(shù)據(jù)領(lǐng)域最活躍的開源項(xiàng)目之一。Spark是基于MapReduce計(jì)算框架實(shí)現(xiàn)的分布式計(jì)算，擁有Hadoop MapReduce所具有的優(yōu)點(diǎn)；但不同于MapReduce的是中間輸出和結(jié)果可以保存在內(nèi)存中，從而不再需要讀寫HDFS，因此Spark更適用于數(shù)據(jù)挖掘與機(jī)器學(xué)習(xí)等需要迭代的算法。如圖1所示，邏輯回歸算法在Hadoop和Spark上的運(yùn)行時(shí)間對(duì)比圖，可以看出Spark的效率有很大的提升［3］。

Spark由Scala［4］語言實(shí)現(xiàn)的，Scala是一種基于JVM的函數(shù)式編程語言，提供了類似DryadLINQ［5］的編程接口。而且Spark還提供了一個(gè)修改的Scala語言解釋器，能方便地用于交互式編程，用戶可以定義變量、函數(shù)、類以及RDD。

Spark集成了豐富的編程工具，其中Spark SQL用于SQL語言和結(jié)構(gòu)化數(shù)據(jù)處理，Spark Streaming用于流處理，MLlib用于機(jī)器學(xué)習(xí)算法，GraphX用于圖處理。Spark不但能夠訪問多種數(shù)據(jù)源，例如 HDFS、Cassandra、HBase、Amazon S3，還提供了Scala、Java和Python三種語言的API接口，以便于編寫并行程序。而且Spark還能部署在已有的Hadoop系統(tǒng)上，由YARN進(jìn)行集群調(diào)度，極大地利用了Hadoop系統(tǒng)。

圖1　邏輯回歸算法在Hadoop和Spark上的運(yùn)行時(shí)間

1.1適用場(chǎng)景

Spark適用于那些在多個(gè)并行操作之間重用數(shù)據(jù)的應(yīng)用，而MapReduce在這方面效率并不高，因?yàn)镸apReduce和DAG引擎是基于非循環(huán)數(shù)據(jù)流的，即一個(gè)應(yīng)用被分成一些不同的作業(yè)（job），每個(gè)作業(yè)從磁盤中讀數(shù)據(jù)，然后再寫到磁盤［2］。Spark不太適合那些異步更新共享狀態(tài)的應(yīng)用，例如并行Web爬行器。Spark的適用場(chǎng)景有：

●迭代式算法：許多機(jī)器學(xué)習(xí)算法都用一個(gè)函數(shù)對(duì)相同的數(shù)據(jù)進(jìn)行重復(fù)的計(jì)算，從而得到最優(yōu)解。MapReduce計(jì)算框架把每次迭代看成是一個(gè)MapReduce作業(yè) （job），而每個(gè)作業(yè)都要從磁盤重新加載數(shù)據(jù)，這就導(dǎo)致了效率不高，而Spark可以把中間數(shù)據(jù)緩存到內(nèi)存中加快計(jì)算效率。

●交互式數(shù)據(jù)分析：用戶經(jīng)常會(huì)用SQL對(duì)大數(shù)據(jù)集合做臨時(shí)查詢（Ad-Hoc Query）。Hive把每次查詢都當(dāng)作一個(gè)獨(dú)立的MapReduce作業(yè)，并且從磁盤加載數(shù)據(jù)，有很大的延遲，而Spark可以把數(shù)據(jù)加載到內(nèi)存中，然后重復(fù)的查詢。

●流應(yīng)用：即需要實(shí)時(shí)處理的應(yīng)用，這類應(yīng)用往往需要低延遲，高效率。

1.2組成部分

目前 Spark由四部分構(gòu)成：Spark SQL、MLlib、GraphX、Spark Streaming，如圖2所示。

圖2　Spark組成部分

（1）Spark SQL：是Spark處理SQL和結(jié)構(gòu)化數(shù)據(jù)工具，Spark引入了SchemaRDD的數(shù)據(jù)抽象，使其能以統(tǒng)一地、高效地訪問和查詢各種不同的數(shù)據(jù)源，例如Apache Hive表、parquet文件、JSON文件。Spark SQL兼容Apache Hive，能重用Hive的前端和元存儲(chǔ)。Spark SQL API能像查詢RDD一樣查詢結(jié)構(gòu)化的數(shù)據(jù)，并且Spark SQL還提供了JDBC/ODBC的服務(wù)端模式，以便建立JDBC/ODBC數(shù)據(jù)連接。

（2）MLlib（Machine Learning）：是Spark提供的機(jī)器學(xué)習(xí)庫(kù)，包含了常見的機(jī)器學(xué)習(xí)算法。Spark擅長(zhǎng)于迭代式計(jì)算，所以與MapReduce相比，MLlib中的算法效率更高，性能更好。常見的算法有：

①SVM、邏輯回歸（logistic regression）、線性回歸（linear regression）、樸素貝葉斯（naive Bayes）

②K均值算法（K-means）

③奇異值分解（singular value decomposition）

④特征提取與轉(zhuǎn)換（feature extraction and transformation）

（3）GraphX（Graph Processing）：是Spark處理圖（graph）的框架，利用Pregel API可以用RDD有效地轉(zhuǎn)換（transform）和連接（join）圖，實(shí)現(xiàn)圖算法。GraphX在速度上可與最快的專用圖處理系統(tǒng)相媲美。

（4）Spark Streaming：是Spark處理流應(yīng)用的庫(kù)。其結(jié)合了批處理查詢與交互式查詢，方便重用批處理的代碼和歷史數(shù)據(jù)?；驹硎荢park將流數(shù)據(jù)分成小的時(shí)間片斷（幾秒），以類似批處理的方式來處理這小部分?jǐn)?shù)據(jù)。Spark Streaming API能像編寫批處理作業(yè)一樣構(gòu)建可擴(kuò)展的流應(yīng)用。Spark Streaming也能訪問各種不同的數(shù)據(jù)源，例如能夠從HDFS、Flume、Kafka、Twitter和ZeroMQ中讀取數(shù)據(jù)。

1.3部署模式

Spark集群如圖3劃分，主要有驅(qū)動(dòng)程序、集群管理程序，以及各worker節(jié)點(diǎn)上的執(zhí)行程序［3］。

Spark應(yīng)用的主程序稱為驅(qū)動(dòng)程序 （driver program），其中包含SparkContext對(duì)象，用于連接集群管理程序（cluster manager）。集群管理程序用于分配集群中的資源，目前有三種：Spark獨(dú)自的集群管理程序、Mesos［6］和 YARN［7］。Spark應(yīng)用的運(yùn)行過程是，Spark-Context對(duì)象首先連接集群管理程序，然后Spark獲取集群中各個(gè)節(jié)點(diǎn)上的執(zhí)行程序（executor），執(zhí)行程序是用于計(jì)算和存儲(chǔ)數(shù)據(jù)的進(jìn)程，然后Spark把代碼發(fā)送到執(zhí)行程序，最后運(yùn)行執(zhí)行程序上的任務(wù)。所以Spark目前支持3種集群部署模式：Standalone模式、Apache Mesos模式、Hadoop YARN模式。

圖3　Spark集群組成部分

（1）Standalone模式：即獨(dú)立模式，使用Spark自帶的集群管理程序，好處是不需要額外的軟件就能運(yùn)行，配置簡(jiǎn)單。

（2）Apache Mesos模式：需要安裝Mesos資源管理程序，Spark的集群管理就交給Mesos處理了。使用Mesos的好處是，可以在Spark與其他框架之間或多個(gè)Spark實(shí)例之間動(dòng)態(tài)地劃分。

（3）Hadoop YARN模式：利用已有的Hadoop集群，讓Spark在Hadoop集群中運(yùn)行，訪問HDFS文件系統(tǒng)，使用YARN資源調(diào)度程序。Spark on YARN是Spark 0.6.0版本加入的，該模式需要額外的配置參數(shù)。

Spark既可以在單機(jī)上運(yùn)行，也可以在Amazon EC2上運(yùn)行。Spark提供了相應(yīng)的配置文件、啟動(dòng)腳本、結(jié)束腳本用于配置、啟動(dòng)、結(jié)束Spark集群中的master 和slave。也提供了相應(yīng)的Web監(jiān)控系統(tǒng)與日志系統(tǒng)，方便地監(jiān)控與調(diào)試程序。而且還可以配置Zookeeper以保證高可用性。

2　Spark編程模型

Spark最主要的抽象就是彈性分布式數(shù)據(jù)集（Resilient Distributed Datasets，RDD）以及對(duì)RDD的并行操作（例如map、filter、groupBy、join）。而且，Spark還支持兩種受限的共享變量 （shared variables）：廣播變量（broadcast variables）和累加變量（accumulators）。

2.1RDD

RDD是只讀的對(duì)象集合，RDD分區(qū)分布在集群的節(jié)點(diǎn)中。如果某個(gè)節(jié)點(diǎn)失效，或者某部分?jǐn)?shù)據(jù)丟失，RDD都能重新構(gòu)建。Spark將創(chuàng)建RDD的一系列轉(zhuǎn)換記錄下來，以便恢復(fù)丟失的分區(qū)，這稱為血系（lineage）。每次對(duì)RDD數(shù)據(jù)集的操作之后的結(jié)果，都可以緩存到內(nèi)存中，下一個(gè)操作可以直接從內(nèi)存中輸入，省去了MapReduce大量的磁盤操作。RDD只支持粗粒度轉(zhuǎn)換，即在大量記錄上執(zhí)行的單個(gè)操作。雖然只支持粗粒度轉(zhuǎn)換限制了編程模型，但RDD仍然可以很好地適用于很多應(yīng)用，特別是支持?jǐn)?shù)據(jù)并行的批量分析應(yīng)用，包括數(shù)據(jù)挖掘、機(jī)器學(xué)習(xí)、圖算法等，因?yàn)檫@些程序通常都會(huì)在很多記錄上執(zhí)行相同的操作。

使用RDD的好處有：

●RDD只能從持久存儲(chǔ)或通過轉(zhuǎn)換（transformation）操作產(chǎn)生，相比于分布式共享內(nèi)存（DSM）可以更高效地實(shí)現(xiàn)容錯(cuò)，對(duì)于丟失部分?jǐn)?shù)據(jù)分區(qū)只需根據(jù)它的血系就可重新計(jì)算出來，而不需要做特定的檢查點(diǎn)（checkpoint）。

●RDD的不變性，可以實(shí)現(xiàn)類MapReduce的預(yù)測(cè)式執(zhí)行。

●RDD的數(shù)據(jù)分區(qū)特性，可以通過數(shù)據(jù)的本地性來提高性能，這與MapReduce是一樣的。

●RDD是可序列化的，當(dāng)內(nèi)存不足時(shí)可自動(dòng)改為磁盤存儲(chǔ)，把RDD存儲(chǔ)于磁盤上，此時(shí)性能會(huì)有大的下降但不會(huì)差于現(xiàn)有的MapReduce。

在Spark中，RDD是一個(gè)Scala對(duì)象，對(duì)RDD的并行操作即是調(diào)用對(duì)象上的方法。有四種方法創(chuàng)建一個(gè)RDD：

（1）通過一個(gè)文件系統(tǒng)中的文件創(chuàng)建，例如常見的HDFS文件。

（2）通過并行化Scala集合，即把一個(gè)集合切分成很多片，然后發(fā)送到各種節(jié)點(diǎn)。

（3）通過對(duì)已有的RDD執(zhí)行轉(zhuǎn)換操作，可以得到一個(gè)新的RDD。例如通過flatMap可以把類型1的RDD轉(zhuǎn)換成類型2的RDD。

（4）通過把RDD持久化。RDD默認(rèn)是惰性的，即只有當(dāng)RDD在執(zhí)行并行操作時(shí)，RDD才被物化，執(zhí)行完后即被釋放。用戶可以通過顯式的cache或save操作使RDD持久化。

2.2并行操作

作用在RDD上的并行操作有兩種：轉(zhuǎn)換（transformation）和動(dòng)作（action），轉(zhuǎn)換返回一個(gè)新的RDD，動(dòng)作返回一個(gè)值或把RDD寫到文件系統(tǒng)中。轉(zhuǎn)換是惰性的，即從一個(gè)RDD轉(zhuǎn)換成另一個(gè)RDD不是馬上執(zhí)行的，Spark只是記錄這樣的操作，并不執(zhí)行，等到有動(dòng)作操作時(shí)才會(huì)啟動(dòng)計(jì)算過程。常見的轉(zhuǎn)換（transformation）如表1所示，常見的動(dòng)作（action）如表2所示。

表1　常見的轉(zhuǎn)換（transformation）操作

注：有些操作只對(duì)key有效，例如join、groupByKey, reduceByKey。除了這些操作以外，用戶還可以請(qǐng)求將RDD緩存起來。而且，用戶還可以通過Partitioner類獲取RDD的分區(qū)順序，然后將另一個(gè)RDD按照同樣的方式分區(qū)。

2.3共享變量

通常情況下，Spark中的map、filter、reduce等函數(shù)的參數(shù)是一個(gè)函數(shù)（閉包），運(yùn)行時(shí)這些函數(shù)參數(shù)被復(fù)制到各個(gè)worker節(jié)點(diǎn)上，互不干擾。Spark還提供了共享變量用于其他用途，常見的有兩種：

（1）廣播變量（broadcast variables）：對(duì)于大量的只讀數(shù)據(jù)，當(dāng)有多個(gè)并行操作時(shí)，最好只復(fù)制一次而不是每執(zhí)行一次函數(shù)就復(fù)制一次到各個(gè)worker節(jié)點(diǎn)。廣播變量就是用于這種情況，它只是包裝了一下原有的數(shù)據(jù)，然后只復(fù)制一次到各worker節(jié)點(diǎn)。

（2）累加變量（accumulators）：累加變量只能用于關(guān)聯(lián)操作，并且只有驅(qū)動(dòng)程序才能讀取。只要某個(gè)類型有“add”操作和“0”值都可以是累加變量。累加變量經(jīng)常用于實(shí)現(xiàn)MapReduce的計(jì)數(shù)器，而且由于是只加性的，所以很容易實(shí)現(xiàn)容錯(cuò)性。

表2　常見的動(dòng)作（action）操作

3　編程示例

Spark提供了 Scala、Java和 Python三種語言的API。而且Spark程序既可以通過交互式Spark shell運(yùn)行 （Scala或Python語言），又能以獨(dú)立的程序運(yùn)行（Scala、Java或Python語言）。下面這些編程示例使用Scala語言，在Spark shell下執(zhí)行［8］。Scala是一種基于JVM的函數(shù)式編程語言。

Spark的首要抽象便是彈性分布式數(shù)據(jù)集RDD，所以編程的主要任務(wù)就是編寫驅(qū)動(dòng)程序，創(chuàng)建RDD，然后對(duì)RDD執(zhí)行并行操作。RDD既能從外部文件中創(chuàng)建（例如HDFS文件），又能從對(duì)其他RDD執(zhí)行轉(zhuǎn)換操作（transformation）得到。對(duì)RDD有兩種操作：一種是轉(zhuǎn)換操作，產(chǎn)生新的RDD；另一種是動(dòng)作操作（action），開始一個(gè)作業(yè)并返回值。首先配置好Spark運(yùn)行環(huán)境，然后啟動(dòng)Spark集群，在此不再細(xì)述。

3.1文本搜索

本示例搜索某個(gè)HDFS日志文件的ERROR信息：

Spark還能顯示的緩存RDD，只需執(zhí)行cache操作：

3.2單詞統(tǒng)計(jì)

本示例統(tǒng)計(jì)某個(gè)HDFS文件的各個(gè)單詞出現(xiàn)的次數(shù)，并將結(jié)果保存到HDFS文件：

3.3估算PI值

Spark也用于計(jì)算密集型任務(wù)，本示例使用“扔飛鏢法”估算PI值，在（0,0）-（1,1）的正方形中，隨機(jī)生成坐標(biāo)（x,y），統(tǒng)計(jì)落在圓內(nèi)的點(diǎn)數(shù)，那么落在圓內(nèi)的點(diǎn)數(shù)/總點(diǎn)數(shù)等于PI/4：

4　結(jié)語

本文大致介紹了Spark系統(tǒng)的基本概念與核心思想，并給出了編程示例。Spark最重要的抽象就是RDD，一種有效的、通用的分布式內(nèi)存抽象，它解決了集群環(huán)境下并行處理大數(shù)據(jù)的效率問題，比Hadoop MapReduce的效率高，特別適用于機(jī)器學(xué)習(xí)中的迭代式算法和交互式數(shù)據(jù)分析等特殊的應(yīng)用場(chǎng)景。目前，Spark是非常流行的內(nèi)存計(jì)算框架，一直在發(fā)布新版本，還處于比較活躍的開發(fā)階段。當(dāng)前遇到的技術(shù)挑戰(zhàn)有：①資源調(diào)度程序如何為Spark作業(yè)確定合適的資源需求；②Spark如何更好地與YARN集群管理程序配合，使系統(tǒng)最優(yōu)；③提供更多的編程API供用戶使用。

［1］Zaharia M,Chowdhury M,Das T,et al.Resilient Distributed Datasets:A Fault-Tolerant Abstraction for in-Memory Cluster Computing ［C］.Proceedings of the 9th USENIX Conference on Networked Systems Design and Implementation.USENIX Association,2012:2-2

［2］Zaharia M,Chowdhury M,Franklin M J,et al.Spark:Cluster Computing with Working Sets［C］.Proceedings of the 2nd USENIX Conference on Hot Topics in Cloud Computing，2010:10～10

［3］Spark［EB/OL］.http://spark.apache.org

［4］Scala［EB/OL］.https://www.scala-lang.org

［5］Yu Y,Isard M,Fetterly D,et al.DryadLINQ:A System for General-Purpose Distributed Data-Parallel Computing Using a High-Level Language［C］.OSDI.2008,8:1·14

［6］Hadoop MapReduce Tutorial［EB/OL］.http://hadoop.apache.org/docs/r1.2.1/mapred_tutorial.html

［7］Apache Mesos.http://mesos.apache.org

［8］Spark Programming Guides［EB/OL］.http://spark.apache.org/docs/1.1.0/quick-start.html

Spark；Hadoop；MapReduce；Big Data；Data Analysis

Research on Apache Spark for Big Data Processing

LI Wen-yang
（College of Computer Science,Sichuan University,Chengdu 610065）

1007-1423（2015）08-0055-06

10.3969/j.issn.1007-1423.2015.08.013

黎文陽（1990-），男，河南信陽人，碩士研究生，研究方向?yàn)榉植际脚c數(shù)據(jù)庫(kù)

2015-02-10

2015-02-28

Apache Spark是當(dāng)前流行的大數(shù)據(jù)處理模型，具有快速、通用、簡(jiǎn)單等特點(diǎn)。Spark是針對(duì)MapReduce在迭代式機(jī)器學(xué)習(xí)算法和交互式數(shù)據(jù)挖掘等應(yīng)用方面的低效率，而提出的新的內(nèi)存計(jì)算框架，既保留了MapReduce的可擴(kuò)展性、容錯(cuò)性、兼容性，又彌補(bǔ)了MapReduce在這些應(yīng)用上的不足。由于采用基于內(nèi)存的集群計(jì)算，所以Spark在這些應(yīng)用上比MapReduce快100倍。介紹Spark的基本概念、組成部分、部署模式，分析Spark的核心內(nèi)容與編程模型，給出相關(guān)的編程示例。

Spark；Hadoop；MapReduce；大數(shù)據(jù)；數(shù)據(jù)分析

Apache Spark is a popular model for large scale data processing at present,which is fast,general and easy.Compared with the MapReduce computing framework,Spark is efficient in iterative machine learning algorithms and interactive data mining applications while retaining the compatibility,scalability and fault-tolerance of MapReduce.With its in-memory computing,Spark is up to 100x faster than Hadoop MapReduce in memory.Presents the basic conception,component and the deploying mode of Spark,introduces the internal abstraction and the programming model,gives the programming examples.