Spark集群實現統計文檔單詞頻次實例

鄭海鵬

摘要：隨著計算機數據分析、數據處理技術的不斷發展和完善，大數據技術在社會中的實際應用場景越來越廣泛，我們的生活正處在大數據時代。例如，眾多電商平臺利用大數據技術實現了電商用戶的數據畫像。依據客戶的需求導向，對客戶的商品需求進行精準的定位，進一步地滿足客戶的實際需求，增加電商平臺的產品銷售量及銷售利潤，知名搜索引擎百度依據客戶的百度搜索日志，投其所好，推送和用戶瀏覽日志密切相關的新聞，滿足用戶的網絡需求。大數據正在逐步地影響和改變我們的生活，該文力圖對大數據技術開發環境的配置做簡要的闡述，并利用Spark（一款基于內存的計算框架，運行速度比MapReduce快100倍左右）集群實現對上傳至HDSF（分布式文件系統）中文檔內的單詞次數的統計，以此闡述大數據技術處理數據的運行流程。

關鍵詞：大數據;Spark集群;分布式文件系統;單詞次數統計

中圖分類號：T311? ? ? ? ?文獻標志碼：A

文章編號：1009-3044（2020）23-0033-02

1 背景

現代社會信息量增長迅猛，人們通過互聯網獲取的信息量非常之大，我們每天都會接收到來自網絡的大量信息，數據量極其龐大，人們早期使用的數據處理方式和方法，已經無法滿足海量數據的處理要求，大數據（BigData）技術應運而生[1]。不同以往的數據處理方法，大數據不僅可以處理以往的結構化數據，同時也可以處理半結構化及非結構化的數據，數據處理方式多樣化。大數據技術還可以讓多臺計算機并行地處理海量數據（集群），從而大大地縮短數據處理周期，提升數據處理的時效性。

2 Hadoop大數據技術概述

Hadoop是由Apache軟件基金會支持，采用Java語言開發。以Hadoop Distributed File System（簡稱HDFS：分布式文件系統）和Mapreduce（計算框架）為核心，以及一些支持Hadoop的相關子項目的通用工具組成的開源分布式數據處理系統。

2.1 HDFS分布式系統

HDFS分布式系統是Hadoop的存儲系統。該系統具有“一次寫入、多次讀取”的特點，即一個文件在一個時刻只能被一個調用者執行寫操作，但可以被多個調用者執行讀操作。HDFS以流式數據（stream）訪問模式來存儲超大文件，運行在多個硬件集群中。此外，該系統還具有高容錯性。系統中的數據分為元數據和節點數據分別存儲在Namenode和Datanode節點中，使得數據的存儲、讀寫更加高效。

2.2 MapReduce與Spark計算框架

2.2.1 MapReduce

MapReduce（以下簡稱MR）主要由Maper和Reducer兩個函數組成，Maper函數將數據處理成形式的數據并將處理后的數據傳給Reducer函數，Reducer函數主要用于合并。MR相比于Spark缺點較明顯，Maper函數處理數據過程中會產生多個中間結果，需要寫入磁盤，無法充分利用內存。

多個MR之間通過HDFS實現交換數據，任務調度和啟動開銷大; Map端和Reduce端均需要排序。

不適合迭代計算（如機器學習等），交互式處理（數據挖掘） 和流式處理（日志分析）[2]。

2.2.2 Spark

Spark是基于內存的計算框架。Spark計算框架提供Cache機制，支持反復迭代、多次數據共享，進入大大地縮減了數據讀取的IO開銷 ;使用多線程池模型來減少任務的啟動開銷;支持多種語言開發，如：Scala、Java、Python等;適合迭代計算、交互式及流式處理。

2.3 Yarn資源管理平臺

Yarn主要負責整個集群的資源調度，并負責管理集群所有任務的運行及任務資源的分配，主要由Resourcemanager（資源管理）、Nodemanager（節點管理）、ApplicationMaster（程序管理）等組成，在此僅做簡單介紹。

3 Spark集群的搭建

3.1 Spark集群軟件資源需求

Spark集群由一臺主機 （Mater） 和 3 臺從機（ Node） 構成，Master用于管理Spark集群Namenode節點的元數據，從機Node（分別為node1、node2、node3）用于管理Datanode數據節點。具體硬件配置環境為 CPU： I8處理器; 內存： 8G; 固態硬盤256G+1 TB。在Spark集群搭建過程中，需要安裝的軟件包括CentOS（linux操作系統）、JDK、Hadoop、Spark、Scala、IDEA等。軟件版本及相關說明如表1所示。

3.2 Spark集群搭建

Spark集群主要由一個Namenode節點和三個Datanode節點組成，集群搭建順序如下：

1） 安裝 CentOS 操作系統;

2） 安裝Java 運行環境JDK;

3） ssh 免密登錄;

4） 安裝 Hadoop;

5） 安裝Scala ;

6） 安裝Spark搭建完全分布式環境。

3.3 Spark集群啟動

在已配置好的集群中選擇Master主機，在主機界面上右擊選擇->Open in terminal 打開終端，依次輸入以下命令：

1） 輸入 start-all.sh，啟動Hadoop集群中的HDFS和Yarn服務;

2） 輸入cd /usr/local/spark/sbin/ 進入Spark的sbin目錄;

3） 輸入 ./start-all.sh命令，啟動Spark集群的worker節點;

4）輸入cd /usr/local/spark/bin/ 進入Spark的bin目錄;

5） 輸入 ./spark-shell --master spark：//master：7077 --executor-memory 512m --totala-executor-cores 2 啟動Spark集群（如圖1所示），至此Spark集群搭建完成。

4 程序設計及實現

4.1 程序的設計思路

單詞統計過程中，主要以兩個單詞間含有空格分隔符作為區分依據，來區分前后的兩個單詞。當所有單詞被逐一區分開來后，則需要對所有的單個單詞（key）進行統一映射，生成鍵值對形式的數據格式（value值設定為1），然后依據Key值進行排序（升序或降序），這樣所有Key相等的鍵值對會排列在一起，再利用Reduce函數實現鍵值對的合并，從而使得Key值相同的數據最終被合并為一個鍵值對，在將之輸出用于實現單詞出現頻次的統計。

4.2 程序的代碼實現

首先啟動Spark集群，然后啟動IDEA軟件編寫程序（基于Scala語言）用于實現文本文件英文單詞次數統計。具體代碼如下（輸出結果如圖2所示）。

//Spark實現單詞次數統計

import org.apache.spark.rdd.RDD //加載內存數據集

import org.apache.spark.{SparkConf， SparkContext} //加載SparkConf及SparkContext類

//創建類：Spark_WorldCount用于實現單詞計數統計

object Spark_WorldCount {

def main（args： Array[String]）： Unit = {

val conf： SparkConf = newSparkConf（）.setAppName（"Spark_WorldCount"）.setMaster（"local[3]"）

val context： SparkContext = new SparkContext（conf）

val lines = context.textFile（args（0）） //讀取文件內容（args（0）為傳遞參數）

val words： RDD[String] =lines.flatMap（_.split（""））//以空格切分單詞

val maps： RDD[（String，Int）]=words.map（（_，1）） //形成鍵對值，Value=1

val reduced：RDD[（String，Int）]=maps.reduceByKey（_+_）//將相同Key值的Value進行累加

val res：RDD[（String，Int）]=reduced.sortBy（_._2，false）//按Value值降序排列

println（res.collect（）.toBuffer） //以可變數組的形式輸出排好序后的單詞鍵對

context.stop（）? //清空緩存

}}

參考文獻：

[1] 張圣杰. 分布式大數據管理系統的設計與實現研究[J]. 信息通信， 2019， 32（2）： 217-218.

[2] 張琴. 大數據處理統一引擎Apache Spark研究[J]. 現代制造技術與裝備， 2017（8）： 184-185， 187.

[3] 龔永罡， 田潤琳， 廉小親， 等. 基于MapReduce的三元N-gram算法的并行化研究[J]. 電子技術應用， 2019， 45（5）： 70-73， 77.

【通聯編輯：謝媛媛】