Spark集群實(shí)現(xiàn)統(tǒng)計(jì)文檔單詞頻次實(shí)例

鄭海鵬

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

關(guān)鍵詞：大數(shù)據(jù);Spark集群;分布式文件系統(tǒng);單詞次數(shù)統(tǒng)計(jì)

中圖分類號(hào)：T311? ? ? ? ?文獻(xiàn)標(biāo)志碼：A

文章編號(hào)：1009-3044（2020）23-0033-02

1 背景

現(xiàn)代社會(huì)信息量增長迅猛，人們通過互聯(lián)網(wǎng)獲取的信息量非常之大，我們每天都會(huì)接收到來自網(wǎng)絡(luò)的大量信息，數(shù)據(jù)量極其龐大，人們?cè)缙谑褂玫臄?shù)據(jù)處理方式和方法，已經(jīng)無法滿足海量數(shù)據(jù)的處理要求，大數(shù)據(jù)（BigData）技術(shù)應(yīng)運(yùn)而生[1]。不同以往的數(shù)據(jù)處理方法，大數(shù)據(jù)不僅可以處理以往的結(jié)構(gòu)化數(shù)據(jù)，同時(shí)也可以處理半結(jié)構(gòu)化及非結(jié)構(gòu)化的數(shù)據(jù)，數(shù)據(jù)處理方式多樣化。大數(shù)據(jù)技術(shù)還可以讓多臺(tái)計(jì)算機(jī)并行地處理海量數(shù)據(jù)（集群），從而大大地縮短數(shù)據(jù)處理周期，提升數(shù)據(jù)處理的時(shí)效性。

2 Hadoop大數(shù)據(jù)技術(shù)概述

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

2.1 HDFS分布式系統(tǒng)

HDFS分布式系統(tǒng)是Hadoop的存儲(chǔ)系統(tǒng)。該系統(tǒng)具有“一次寫入、多次讀取”的特點(diǎn)，即一個(gè)文件在一個(gè)時(shí)刻只能被一個(gè)調(diào)用者執(zhí)行寫操作，但可以被多個(gè)調(diào)用者執(zhí)行讀操作。HDFS以流式數(shù)據(jù)（stream）訪問模式來存儲(chǔ)超大文件，運(yùn)行在多個(gè)硬件集群中。此外，該系統(tǒng)還具有高容錯(cuò)性。系統(tǒng)中的數(shù)據(jù)分為元數(shù)據(jù)和節(jié)點(diǎn)數(shù)據(jù)分別存儲(chǔ)在Namenode和Datanode節(jié)點(diǎn)中，使得數(shù)據(jù)的存儲(chǔ)、讀寫更加高效。

2.2 MapReduce與Spark計(jì)算框架

2.2.1 MapReduce

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

多個(gè)MR之間通過HDFS實(shí)現(xiàn)交換數(shù)據(jù)，任務(wù)調(diào)度和啟動(dòng)開銷大; Map端和Reduce端均需要排序。

不適合迭代計(jì)算（如機(jī)器學(xué)習(xí)等），交互式處理（數(shù)據(jù)挖掘） 和流式處理（日志分析）[2]。

2.2.2 Spark

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

2.3 Yarn資源管理平臺(tái)

Yarn主要負(fù)責(zé)整個(gè)集群的資源調(diào)度，并負(fù)責(zé)管理集群所有任務(wù)的運(yùn)行及任務(wù)資源的分配，主要由Resourcemanager（資源管理）、Nodemanager（節(jié)點(diǎn)管理）、ApplicationMaster（程序管理）等組成，在此僅做簡單介紹。

3 Spark集群的搭建

3.1 Spark集群軟件資源需求

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

3.2 Spark集群搭建

Spark集群主要由一個(gè)Namenode節(jié)點(diǎn)和三個(gè)Datanode節(jié)點(diǎn)組成，集群搭建順序如下：

1） 安裝 CentOS 操作系統(tǒng);

2） 安裝Java 運(yùn)行環(huán)境JDK;

3） ssh 免密登錄;

4） 安裝 Hadoop;

5） 安裝Scala ;

6） 安裝Spark搭建完全分布式環(huán)境。

3.3 Spark集群啟動(dòng)

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

1） 輸入 start-all.sh，啟動(dòng)Hadoop集群中的HDFS和Yarn服務(wù);

2） 輸入cd /usr/local/spark/sbin/ 進(jìn)入Spark的sbin目錄;

3） 輸入 ./start-all.sh命令，啟動(dòng)Spark集群的worker節(jié)點(diǎn);

4）輸入cd /usr/local/spark/bin/ 進(jìn)入Spark的bin目錄;

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

4 程序設(shè)計(jì)及實(shí)現(xiàn)

4.1 程序的設(shè)計(jì)思路

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

4.2 程序的代碼實(shí)現(xiàn)

首先啟動(dòng)Spark集群，然后啟動(dòng)IDEA軟件編寫程序（基于Scala語言）用于實(shí)現(xiàn)文本文件英文單詞次數(shù)統(tǒng)計(jì)。具體代碼如下（輸出結(jié)果如圖2所示）。

//Spark實(shí)現(xiàn)單詞次數(shù)統(tǒng)計(jì)

import org.apache.spark.rdd.RDD //加載內(nèi)存數(shù)據(jù)集

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

//創(chuàng)建類：Spark_WorldCount用于實(shí)現(xiàn)單詞計(jì)數(shù)統(tǒng)計(jì)

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）） //讀取文件內(nèi)容（args（0）為傳遞參數(shù)）

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

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

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

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

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

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

}}

參考文獻(xiàn)：

[1] 張圣杰. 分布式大數(shù)據(jù)管理系統(tǒng)的設(shè)計(jì)與實(shí)現(xiàn)研究[J]. 信息通信， 2019， 32（2）： 217-218.

[2] 張琴. 大數(shù)據(jù)處理統(tǒng)一引擎Apache Spark研究[J]. 現(xiàn)代制造技術(shù)與裝備， 2017（8）： 184-185， 187.

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

【通聯(lián)編輯：謝媛媛】