Spark平臺下的RDD研究與應用

馬兆輝 趙睿哲 溫秀梅,2*

(1.河北建筑工程學院,河北 張家口 075000;2.張家口市大數據技術創新中心,河北 張家口 075000)

0 引 言

Spark基于RDD實現了一體化、多元化的大數據處理體系,強大的計算能力以及高度集成化的特點使得Spark在大數據計算領域具有得天獨厚的優勢.在Spark出現之前,Hadoop平臺下的MapReduce框架是最熱門的大數據計算框架,但是MapReduce框架仍暴露出很多缺點,其中最主要的是迭代計算的中間結果會不停寫入磁盤,造成了數據復制嚴重、磁盤開銷大等問題.同Spark框架相比,MapReduce框架表達能力有限,不得不借助第三方工具去完成更為復雜的任務.Spark框架是為了解決這些問題而設計的,Spark框架不僅擁有更豐富的函數,可以對更為復雜的海量數據進行快速操作,而且可以將中間結果存入內存,通過RDD之間存在的依賴關系形成DAG圖進行轉換操作,實現流水線進程,使用戶不必再擔心底層數據的特性,減少了磁盤的開銷,提升了運行速度,提高了容錯性,同時還開發出完整的Spark生態系統,減少了開發和維護成本,對大規模數據的處理更加方便快捷,其各個組件可以共同完成絕大部分的數據處理需求和場景.本文針對Spark中最為重要的核心組件之一RDD進行詳細介紹,并通過具體實驗進行說明.

1 相關知識

1.1 RDD

RDD(Resilient Distributed Dataset)是彈性分布式數據集,是一種抽象的分布式內存概念,是Spark平臺中運行計算的基本存儲單元[1].RDD具有強大的容錯功能,不僅可以并行處理元素,同時還是一個高度抽象的數據結構,包含多個分區.其創建方式主要包括兩大類:一類是來自共享文件系統、HDFS、HBase的外部文件系統,另一類則是通過任何數據源提供Hadoop Input Format.

RDD提供了豐富的操作來對集合中的元素進行操作.其支持兩種操作類型:Transformations和Actions.Transformations主要是從一個存在的RDD去產生一個新的RDD,而Actions的操作主要是在數據集上計算之后返回給Driver[2].

1)RDD的依賴

在對RDD進行轉換操作的過程中,每個操作都會在已有的RDD的基礎上產生新的RDD.由于RDD的惰性特性,新的RDD會依賴于原有的RDD,這樣RDD之間就會形成相應的依賴關系.

RDD的依賴關系分為兩大類,如圖1所示.

圖1 RDD的依賴關系

①窄依賴:是指每個父RDD的一個Partition最多被子RDD的一個Partition所使用,例如map、filter、union等都會產生窄依賴;

②寬依賴:是指一個父RDD的Partition會被多個子RDD的Partition所使用,例如groupByKey、reduceByKey、sortByKey等操作都會產生寬依賴[3].

2)RDD之間的轉換關系

Spark中最核心的部分就是RDD,RDD是一個不可變、粗粒度的數據集合.在RDD抽象數據模型中提供了豐富的轉換操作,然而所有的轉換操作都不會實際的執行,這正是由于其惰性特性,因此只會記錄操作的步驟,真正的計算發生在RDD的“動作”操作,如圖2所示.

圖2 RDD的轉換關系

3)RDD操作類型

RDD支持兩種操作:轉換(Transformation)和動作(Action,動作或行動).其中轉換操作用于對RDD的創建,是通過操作方法從已經存在的數據集中創建一個新的數據集,動作操作是數據執行部分,主要是指計算數據集里的數據后并將結果返回到Driver.

由于轉換操作都具有Lazy特性,即Spark不會立即進行實際的操作,只會記錄執行的流程,只有發出Action操作的時候才會真正執行[4].默認情況下,RDD的每個動作在執行的時候,都會將之前的數據重新計算一遍,為了保證計算的高效性以及計算結果的可重用性,在實際計算過程中,根據實際情況,在特定的計算環節上執行persist方法,將計算的中間結果持久化到內存或者磁盤上.如果進行了持久化操作,那么在進行Action操作的時候,就會從內存或者磁盤將已經計算好的數據取出直接用于后續計算,這樣節省了計算步驟和時間,同時也提高了整體的計算效率.

4)RDD的彈性特性

RDD之所以被稱為彈性數據集,其主要體現在以下幾個方面.

①自動將存儲在內存和磁盤中的數據切換.RDD是基于內存的,但是當內存“滿”的時候,會將一部分數據放到磁盤,前提是持久化級別設置成MEMORY_AND_DISK.

②基于Lineage的高效容錯.若計算步驟很多,如果其中某個環節出錯,可從指定位置恢復已經計算好的數據,有效避免了重新計算.當然可恢復的前提是在相應位置進行了計算結果數據的持久化.

③Task如果失敗會自動進行特定次數的重試.

④Satge如果失敗會自動進行特定次數的重試,而且只會計算失敗的分區.

⑤檢查點和持久化過程.在計算過程中,有的計算相對復雜,若計算鏈條相對較長或者其結果經常被訪問,可以將其結果進行緩存,以便后續直接訪問,以此來節省計算時間,提高整體運行速度.

⑥數據調度、DAG調度、Task調度和資源管理無關.Spark集群中任務調度和資源調度是分開的.

⑦數據分區的高度彈性.在計算過程中,當數據分區較小時會降低處理效率,為了提高處理效率,需要將小的分區合并成一個較大分區進行處理;而當數據分區較大時,由于內存大小限制,需要把分區劃分成較小的數據分區.可以根據不同的情況設置不同的分區數量和大小,提高或降低并行度[5].

5)RDD運行過程

(1)通過讀取集合或來自外部的數據源創建RDD對象;

(2)SparkContext通過RDD的相關操作構建一個DAG作為邏輯執行計劃;

(3)DAGScheduler根據ShuffleDenpendency將DAG劃分為多個階段,每個階段包含多個tasks,之后每個task會被TaskScheduler調度到不同節點的Executor上啟動執行.如圖3所示.

圖3 RDD在Spark中的運行過程

2 RDD的具體實現

使用本地系統創建RDD,在IDEA中實現電商用戶頁面單跳轉化率統計實驗和電商熱門品類中Top10活躍Session統計實驗的具體案例,各軟件具體版本如表1所示.

表1 軟件及對應版本表

2.1 創建RDD的方式

2.1.1 使用程序中的集合創建RDD

RDD的數據來源可以是程序中的集合,在Spark中可以通過parallelize和makeRDD將集合轉化成RDD,SparkContext中的parallelize方法可以指定分區個數.源碼如圖4所示.

圖4 使用程序中的集合創建RDD

2.1.2 使用本地文件系統創建RDD

RDD的數據來源也可以是本地的文件系統,這對于程序中需要進行相對較大的數據量測試是很有必要的.在Spark中可以通過textFile方法來讀取本地文件系統創建RDD.源碼如圖5所示.

圖5 使用本地文件創建RDD

2.1.3 使用HDFS創建RDD

HDFS可以作為RDD的數據來源,而且從HDFS上讀取數據來創建RDD的方式也是目前Spark生產系統中最常用的方式.源碼如圖6所示.

圖6 使用HDFS創建RDD

2.2 基于Spark的電商用戶頁面單跳轉化率統計實驗

頁面單跳轉化率是指一個用戶在一次電商購物過程中訪問的頁面路徑如首頁、產品列表頁、產品詳情頁、訂單頁面、支付頁面,首頁跳轉到產品列表頁叫一次單跳,訂單頁面跳到支付頁面也叫一次單跳.單跳轉化率就是統計頁面點擊的概率.根據頁面轉化率指標的大小,產品經理和運營總監可以分析網站的產品和頁面的表現,決定是否需要去優化網站的布局.電商網頁頁面路徑圖如圖7所示.

圖7 電商網頁頁面路徑圖

2.2.1 實驗分析

實驗數據采集自電商的用戶行為數據,共包括有180570條數據,其中每一條主要包含用戶的4種行為:搜索、點擊、下單和支付.實驗數據如圖8所示.

圖8 電商用戶行為數據

數據中每一行表示用戶的一個行為數據,采用“_”分割字段,如果搜索關鍵字是null即表示無效搜索;如果品類id和產品id為-1即表示為無效點擊;用戶可同時下單多個產品,即品類id和產品id可為多個,多個數據之間采用“,”進行分割,若不是下單行為即用null表示;用戶也可同時支付多個產品,與下單行為類似,若不是支付行為即用null表示.

實驗過程首先對數據進行處理,將每一行的數據分割開,讀取到規定的頁面后,通過reduceByKey轉換算子和countByKey行動算子統計出來每個頁面的訪問次數和每個用戶的頁面單跳跳轉路徑并按時間升序排序,然后過濾出單跳跳轉目標相同的路徑并統計次數,最后計算單跳轉化率.實驗過程如圖9所示.

圖9 實驗流程圖

2.2.2 實驗結果

實驗數據中共有49個頁面,頁面id由1到49,本文實驗結果選取id從1到20的頁面轉化率進行展示.實驗結果如圖10所示,每一行數據由單跳跳轉路徑和單跳轉化率構成.

圖10 頁面單跳轉化率實驗結果圖

2.3 基于Spark的電商熱門品類中Top10活躍Session統計實驗

品類是指產品的分類,部分電商品類可分為多級,此實驗中品類為一級,實驗按照每個品類的點擊、下單、支付的數量來統計出各個品類的數量,并選出數量排名前10的品類作為熱門品類.對于排名前10的品類,分別獲取每個品類點擊次數排名前10的SessionId,這個功能可以看到,某個用戶群體最感興趣的品類以及各個品類中最典型的用戶的Session的行為.

2.3.1 實驗分析

實驗數據仍使用電商的用戶行為數據.

實驗過程首先分別統計每個品類點擊的次數、下單的次數和支付的次數,通過遍歷全部日志數據,根據品類id和操作類型分別累加各個品類的數量,遍歷完成之后就得到了每個品類id和操作類型的數量,按照點擊下單支付的順序來排序,得到Top10熱門品類.過程如圖11所示.

圖11 Top10熱門品類實驗過程圖

過濾出熱門品類Top10的日志,將熱門品類Top10的數據類型轉換為RDD[(categoryId,sessionId),1],并統計數量,將數據類型轉換為RDD[(categoryId,sessionId),count],統計出每個品類中Session的數量,接下來對每個品類中的Session數量進行排序,并取出前10.實驗過程如圖12所示.

圖12 Top10活躍Session實驗流程圖

2.3.2 實驗結果

Top10熱門品類實驗結果圖如圖13所示,結果圖中10個結果為Top10熱門品類,每一行由品類id、點擊量、下單量、支付量組成包裝類,再按照點擊量的順序降序來排序.

圖13 Top10熱門品類實驗結果圖

Top10活躍Session實驗結果圖如圖14所示,每一行數據由Top10熱門品類的品類ID和一個List組成,List中包含10個SessionInfo包裝類,包裝類中包含點擊數最高的SessionId和點擊數,并根據SessionInfo中的點擊數降序排序.

圖14 Top10活躍Session實驗結果圖

3 結 語

Spark不僅含有Hadoop平臺下MapReduce框架所具備的優點,也很好地解決了MapReduce中存在的一些問題,同時在功能上為更好的適應現代大數據環境做了延伸和擴展,使其在操作更簡潔方便的前提下執行速度提高了近百倍.其中,RDD是作為Spark技術中數據操作的基本單位.本文主要論述了RDD的屬性、RDD之間的依賴、常見的轉換關系、操作類型、彈性特性和運行原理,創建RDD的幾種常見的方式以及電商用戶頁面單跳轉化率統計實驗和電商熱門品類中Top10活躍Session統計實驗實現.Spark不僅可以支持Scala、Java、Python等多種語言編程,還支持DataFrame、DataSet等多種數據類型,而且提供了一個完整而強大的生態系統,其中有SQL查詢、流式計算、機器學習和圖計算組件,這些組件可以應用在一個程序中,完成更加復雜的需求.在現代大數據環境下,為了更好解決實際生活中的復雜問題,充分理解并掌握RDD的運行可以讓Spark的運行節省大量的數據處理時間,從而有效地優化數據處理過程,提高整個過程的效率.