鄭海鵬

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

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

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

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

1 背景

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

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

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

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

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

2.2 MapReduce與Spark計算框架

2.2.1 MapReduce

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

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

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

2.2.2 Spark

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

2.3 Yarn資源管理平臺

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

3 Spark集群的搭建

3.1 Spark集群軟件資源需求

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

3.2 Spark集群搭建

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

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

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

3） ssh 免密登錄;

4） 安裝 Hadoop;

5） 安裝Scala ;

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

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節(jié)點;

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 程序設計及實現(xiàn)

4.1 程序的設計思路

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

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

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

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

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

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

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

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）） //形成鍵對值，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） //以可變數(shù)組的形式輸出排好序后的單詞鍵對

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

}}

參考文獻：

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

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

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

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