Spark集群实现统计文档单词频次实例

时间：2021-01-27 10:07:10　来源：达达文档网本文已影响人

郑海鹏

摘要：随着计算机数据分析、数据处理技术的不断发展和完善，大数据技术在社会中的实际应用场景越来越广泛，我们的生活正处在大数据时代。例如，众多电商平台利用大数据技术实现了电商用户的数据画像。依据客户的需求导向，对客户的商品需求进行精准的定位，进一步地满足客户的实际需求，增加电商平台的产品销售量及销售利润，知名搜索引擎百度依据客户的百度搜索日志，投其所好，推送和用户浏览日志密切相关的新闻，满足用户的网络需求。大数据正在逐步地影响和改变我们的生活，该文力图对大数据技术开发环境的配置做简要的阐述，并利用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.

【通联编辑：谢媛媛】

Spark集群实现统计文档单词频次实例

最新文章

热门文章