【智能大数据分析 | 实验一】MapReduce实验：单词计数

【作者主页】Francek Chen
【专栏介绍】$\lceil$智能大数据分析$\rfloor$ 智能大数据分析是指利用先进的技术和算法对大规模数据进行深入分析和挖掘，以提取有价值的信息和洞察。它结合了大数据技术、人工智能（AI）、机器学习（ML）和数据挖掘等多种方法，旨在通过自动化的方式分析复杂数据集，发现潜在的价值和关联性，实现数据的自动化处理和分析，从而支持决策和优化业务流程。与传统的人工分析相比，智能大数据分析具有自动化、深度挖掘、实时性和可视化等特点。智能大数据分析广泛应用于各个领域，包括金融服务、医疗健康、零售、市场营销等，帮助企业做出更为精准的决策，提升竞争力。
【GitCode】专栏资源保存在我的GitCode仓库：https://gitcode.com/Morse_Chen/Intelligent_bigdata_analysis。

---

一、实验目的

基于 MapReduce 思想，编写 WordCount 程序。

二、实验要求

1. 理解 MapReduce 编程思想；
2. 会编写 MapReduce 版本 WordCount；
3. 会执行该程序；
4. 自行分析执行过程。

三、实验原理

MapReduce 是一种计算模型，简单的说就是将大批量的工作（数据）分解（MAP）执行，然后再将结果合并成最终结果（REDUCE）。这样做的好处是可以在任务被分解后，可以通过大量机器进行并行计算，减少整个操作的时间。

适用范围：数据量大，但是数据种类小可以放入内存。

基本原理及要点：将数据交给不同的机器去处理，数据划分，结果归约。

理解 MapReduce 和 Yarn：在新版 Hadoop 中，Yarn 作为一个资源管理调度框架，是 Hadoop 下 MapReduce 程序运行的生存环境。其实 MapRuduce 除了可以运行 Yarn 框架下，也可以运行在诸如 Mesos，Corona 之类的调度框架上，使用不同的调度框架，需要针对 Hadoop 做不同的适配。

一个完成的 MapReduce 程序在 Yarn 中执行过程如下：

（1）ResourcManager JobClient 向 ResourcManager 提交一个 job。

（2）ResourcManager 向 Scheduler 请求一个供 MRAppMaster 运行的 container，然后启动它。

（3）MRAppMaster 启动起来后向 ResourcManager 注册。

（4）ResourcManagerJobClient 向 ResourcManager 获取到 MRAppMaster 相关的信息，然后直接与 MRAppMaster 进行通信。

（5）MRAppMaster 算 splits 并为所有的 map 构造资源请求。

（6）MRAppMaster 做一些必要的 MR OutputCommitter 的准备工作。

（7）MRAppMaster 向 RM(Scheduler) 发起资源请求，得到一组供 map/reduce task 运行的 container，然后与 NodeManager 一起对每一个 container 执行一些必要的任务，包括资源本地化等。

（8）MRAppMaster 监视运行着的 task 直到完成，当 task 失败时，申请新的 container 运行失败的 task。

（9）当每个 map/reduce task 完成后，MRAppMaster 运行 MR OutputCommitter 的 cleanup 代码，也就是进行一些收尾工作。

（10）当所有的 map/reduce 完成后，MRAppMaster 运行 OutputCommitter 的必要的 job commit 或者 abort APIs。

（11）MRAppMaster 退出。

（一）MapReduce 编程

编写在 Hadoop 中依赖 Yarn 框架执行的 MapReduce 程序，并不需要自己开发 MRAppMaster 和 YARNRunner，因为 Hadoop 已经默认提供通用的 YARNRunner 和 MRAppMaster 程序， 大部分情况下只需要编写相应的 Map 处理和 Reduce 处理过程的业务程序即可。

编写一个 MapReduce 程序并不复杂，关键点在于掌握分布式的编程思想和方法，主要将计算过程分为以下五个步骤：

（1）迭代。遍历输入数据，并将之解析成 key/value 对。

（2）将输入 key/value 对映射 (map) 成另外一些 key/value 对。

（3）依据 key 对中间数据进行分组 (grouping)。

（4）以组为单位对数据进行归约 (reduce)。

（5）迭代。将最终产生的 key/value 对保存到输出文件中。

（二）Java API 解析

（1）InputFormat：用于描述输入数据的格式，常用的为 TextInputFormat 提供如下两个功能：

数据切分： 按照某个策略将输入数据切分成若干个 split，以便确定 Map Task 个数以及对应的 split。

为 Mapper 提供数据：给定某个 split，能将其解析成一个个 key/value 对。

（2）OutputFormat：用于描述输出数据的格式，它能够将用户提供的 key/value 对写入特定格式的文件中。

（3）Mapper/Reducer：Mapper/Reducer 中封装了应用程序的数据处理逻辑。

（4）Writable：Hadoop 自定义的序列化接口。实现该类的接口可以用作 MapReduce 过程中的 value 数据使用。

（5）WritableComparable：在 Writable 基础上继承了 Comparable 接口，实现该类的接口可以用作 MapReduce 过程中的 key 数据使用。（因为 key 包含了比较排序的操作）。

四、实验环境

虚拟机软件：VMware 16 Pro
Linux 操作系统版本：CentOS-7-64位
Java 版本：jdk1.7.0_79
Hadoop 版本：hadoop-2.7.1

JDK 和 Hadoop 的安装配置可参考：大数据存储技术（1）—— Hadoop简介及安装配置（以上版本 Java 和 Hadoop 的安装包可到文末链接下载）

五、实验步骤

本实验主要分为，确认前期准备，编写 MapReduce 程序，打包提交代码。查看运行结果这几个步骤，详细如下：

（一）启动 Hadoop

执行命令启动前面实验部署好的 Hadoop 系统。

start-all.sh

（二）验证 HDFS 上没有 wordcount 的文件夹

hadoop fs -ls /

此时 HDFS 上应该是没有 wordcount 文件夹。

（三）上传数据文件到 HDFS

先在 HDFS 上新建目录datasets，然后将数据文件上传到 HDFS。

hadoop fs -mkdir /datasets
hadoop fs -ls /
hadoop fs -put /usr/local/datasets/word /datasets

查看数据文件word的内容。

hadoop fs -cat /datasets/word

（四）编写 MapReduce 程序

首先，我们在 Eclipse 中新建一个 Map/Reduce Project。

接着，正确导入 Hadoop 的路径。

然后，开始编写程序。主要编写 Map 和 Reduce 类，其中 Map 过程需要继承org.apache.hadoop.mapreduce包中Mapper类，并重写其map方法；Reduce过程需要继承org.apache.hadoop.mapreduce包中Reduce类，并重写其reduce方法。

最后，完整代码如下。

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.TextInputFormat;
import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;
import org.apache.hadoop.mapreduce.lib.partition.HashPartitioner;import java.io.IOException;
import java.util.StringTokenizer;public class WordCount {public static class TokenizerMapper extends Mapper<Object, Text, Text, IntWritable> {private final static IntWritable one = new IntWritable(1);private Text word = new Text();//map方法，划分一行文本，读一个单词写出一个<单词,1>public void map(Object key, Text value, Context context)throws IOException, InterruptedException {StringTokenizer itr = new StringTokenizer(value.toString());while (itr.hasMoreTokens()) {word.set(itr.nextToken());context.write(word, one);//写出<单词,1>}}}//定义reduce类，对相同的单词，把它们中的VList值全部相加public static class IntSumReducer extends Reducer<Text, IntWritable, Text, IntWritable> {private IntWritable result = new IntWritable();public void reduce(Text key, Iterable<IntWritable> values,Context context)throws IOException, InterruptedException {int sum = 0;for (IntWritable val : values) {sum += val.get();//相当于<Hello,1><Hello,1>，将两个1相加}result.set(sum);context.write(key, result);//写出这个单词，和这个单词出现次数<单词，单词出现次数>}}public static void main(String[] args) throws Exception {//主方法，函数入口Configuration conf = new Configuration();           //实例化配置文件类Job job = new Job(conf, "WordCount");             //实例化Job类job.setInputFormatClass(TextInputFormat.class);     //指定使用默认输入格式类TextInputFormat.setInputPaths(job, args[0]);      //设置待处理文件的位置job.setJarByClass(WordCount.class);               //设置主类名job.setMapperClass(TokenizerMapper.class);        //指定使用上述自定义Map类job.setCombinerClass(IntSumReducer.class);        //指定开启Combiner函数job.setMapOutputKeyClass(Text.class);            //指定Map类输出的，K类型job.setMapOutputValueClass(IntWritable.class);     //指定Map类输出的，V类型job.setPartitionerClass(HashPartitioner.class);       //指定使用默认的HashPartitioner类job.setReducerClass(IntSumReducer.class);         //指定使用上述自定义Reduce类job.setNumReduceTasks(Integer.parseInt(args[2]));  //指定Reduce个数job.setOutputKeyClass(Text.class);                //指定Reduce类输出的,K类型job.setOutputValueClass(Text.class);               //指定Reduce类输出的,V类型job.setOutputFormatClass(TextOutputFormat.class);  //指定使用默认输出格式类TextOutputFormat.setOutputPath(job, new Path(args[1]));    //设置输出结果文件位置System.exit(job.waitForCompletion(true) ? 0 : 1);    //提交任务并监控任务状态}
}

注意：这里的 jre 要设置成 1.7 版本的，不然后面运行程序会导致版本不兼容的相关错误。

（五）使用 Eclipse 开发工具将该代码打包

假定打包后的文件名为hdpAction.jar，并使用 WinSCP 上传至 Linux 系统。

可使用如下命令向 YARN 集群提交本应用。

yarn jar hdpAction.jar /datasets/word /wordcount 1

其中yarn为命令，jar为命令参数，后面紧跟打包后的代码地址，/datasets/word为输入文件在 HDFS 中的位置，/wordcount为输出文件在 HDFS 中的位置，1通常指的是要使用的reduce任务的数量。

六、实验结果

1. 程序运行成功控制台上的显示内容

2. 在HDFS上查看结果

hadoop fs -ls /wordcount
hadoop fs -cat /wordcount/part-r-00000

七、实验心得

  在进行 MapReduce 单词计数实验的过程中，我深刻体会到分布式计算的强大能力和处理大规模数据的复杂性。通过这个实验，我不仅掌握了 MapReduce 的基本流程，还加深了对其背后原理的理解。

  实验的第一步是 Mapper 阶段。在这一阶段，我的任务是将输入的文本数据进行拆分，提取出每一个单词，并为其分配一个计数值。这个过程虽然看似简单，但实际上却涉及到数据的清洗和格式化。在实际操作中，我意识到单词的分隔符、大小写的处理以及标点符号的剔除都对最终结果产生重大影响。因此，我在编写 Mapper 代码时，特别注重了这些细节，确保能准确无误地提取出每一个单词。接下来是 Reducer 阶段，主要任务是对来自多个 Mapper 的输出进行汇总。通过对单词及其计数进行合并，我能够得到每个单词在整个数据集中的总出现次数。这一过程让我体会到聚合操作的重要性，以及如何高效地处理和合并数据。特别是在面对大规模数据时，优化排序和分组算法显得尤为关键。为了提高效率，我还学习了如何利用内存中的数据结构，以便更快地进行查找和更新。

  通过这个实验，我不仅掌握了 MapReduce 的基本语法和使用方法，更重要的是，我理解了其背后的设计思想和应用场景。在大数据时代，MapReduce 作为一种有效的分布式计算模型，能够帮助我们处理海量数据，为数据分析、机器学习等领域提供强有力的支持。最后，这个实验让我认识到，编写高效的 MapReduce 程序需要对数据结构和算法有深刻的理解，尤其是在性能优化方面。未来，我希望能在此基础上深入研究更高级的分布式计算框架，如 Apache Spark 等，进一步提升自己的技术水平。

  总的来说，这次 MapReduce 单词计数实验不仅提升了我的编程能力，还让我对大数据处理有了更深入的认识，增强了我在未来相关领域继续深入研究的动力。我期待将所学应用于实际项目中，探索更多可能性。

附：以上文中的数据文件及相关资源下载地址：
链接：https://pan.quark.cn/s/a40065ab5906
提取码：4r8J