当前位置：首页 > news >正文

Hadoop之MapReduce

news 2026/8/3 9:27:53

一、MapReduce的概念

Hadoop的三大组件：HDFS、Yarn、MapReduce。
HDFS：解决的是分布式存储的问题。
MapReduce:解决的是计算问题。
Yarn:计算的时候，使用的资源如何协调（Windows操作系统）

2004年，谷歌发表了一篇名为《MapReduce》的论文，主要介绍了如何在分布式的存储系统上对数据进行高效率的计算。2005年，Nutch团队使用Java语言实现了这个技术，并命名为MapReduce。时至今日，MapReduce是Apache Hadoop的核心模块之一，是运行在HDFS上的分布式运算程序的编程框架，用于大规模数据集（大于1TB）的并行运算。其中的概念，"Map（映射）"和"Reduce（归约）"

mapReduce的优缺点：

优点：

1、易于编程
代码写起来有固定的格式，编写难度非常的小，号称是八股文【固定写法】。
2、良好的扩展性
代码的计算资源不够了，可以直接拓展几台即可解决
3、高容出错
如果负责计算的电脑挂掉了，可以将任务转移到其他电脑上，任务不会执行失败的。
4、非常适合大数据集的计算（PB级以上） 1P=1024T

缺点：

1、不适合做实时计算
mapreduce一个任务就要跑很长时间，不利于实时。不能做到秒级或者毫秒级的计算。
mapreduce 属于离线的技术。
2、不适合做流式计算
数据因为都是静态的，不是边产生数据，边计算。
固定计算：数据量是固定的，给了1T 就计算。
3、不适合做有向图（DAG）计算
多个应用程序之间有依赖关系，后一个程序需要依赖前面的程序的结果。这种场景就称之为有向图，mapreduce是不适合的。

二、MapReduce案例--WordCount

1、环境准备

安装hadoop之前要先安装jdk8环境，因为hadoop3.3.6依赖jdk1.8，并置%JAVA_HOME%

解压hadoop的安装包

配置环境变量

配置PATH

验证hadoop是否安装成功

最后一项：将这两个文件粘贴到下面的目录中

在将hadoop.dll 拷贝到 C:\windows\system32 这个文件夹下一份。

2、新建maven项目，并且导入包

引入依赖

<packaging>jar</packaging> <properties> <maven.compiler.source>8</maven.compiler.source> <maven.compiler.target>8</maven.compiler.target> <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding> </properties> <dependencies> <dependency> <groupId>org.apache.hadoop</groupId> <artifactId>hadoop-common</artifactId> <version>3.3.6</version> </dependency> <!-- https://mvnrepository.com/artifact/org.apache.hadoop/hadoop-client --> <dependency> <groupId>org.apache.hadoop</groupId> <artifactId>hadoop-client</artifactId> <version>3.3.6</version> </dependency> <!-- https://mvnrepository.com/artifact/org.apache.hadoop/hadoop-hdfs --> <dependency> <groupId>org.apache.hadoop</groupId> <artifactId>hadoop-hdfs</artifactId> <version>3.3.6</version> </dependency> </dependencies>

3、创建一些数据

在项目的根路径下，创建一个文件夹 data，创建数据的来源文件input文件夹，在input文件夹下面，新建file，a.txt, b.txt, c.txt

a.txt hello bigdata hello 1999 hello beijing hello world hello hello java good b.txt hello gaoxinqu hello bingbing hello chenchen hello ACMilan hello china c.txt hello hadoop hello java hello storm hello spark hello redis hello zookeeper hello hive hello hbase hello flume

4、编写代码

1）编写Map代码

package com.bigdata; import org.apache.hadoop.io.IntWritable; import org.apache.hadoop.io.LongWritable; import org.apache.hadoop.io.Text; import org.apache.hadoop.mapreduce.Mapper; import java.io.IOException; /** * * Mapper中的四个泛型跟什么照应： * 1、LongWritable 行偏移量，一般都是LongWritable ，这一行的数据是从第几个字符开始计算的，因为数据量很多这个值也会很大，所以使用Long * 2、Text 指的是这一行数据 * 3、Text Map任务输出的Key值的类型 单词 * 4、IntWritable Map任务输出的Key值的类型 1 * */ public class WordCountMapper extends Mapper<LongWritable, Text, Text, IntWritable> { /** * * @param key 指的是行偏移量 * @param value 指的是 这一行数据 : hello bigdata hello 1999 hello beijing hello * @param context * @throws IOException * @throws InterruptedException */ @Override protected void map(LongWritable key, Text value, Mapper<LongWritable, Text, Text, IntWritable>.Context context) throws IOException, InterruptedException { String line = value.toString(); // [hello,bigdata,hello,1999,hello,beijing,hello] String[] arr = line.split("\\s+"); // hello-> 1,bigdata->1,hello->1,1999->1,hello->1,beijing->1,hello->1 for (String word: arr) { context.write(new Text(word),new IntWritable(1)); } } }

2）编写Reduce代码

package com.bigdata; import org.apache.hadoop.io.IntWritable; import org.apache.hadoop.io.Text; import org.apache.hadoop.mapreduce.Reducer; import java.io.IOException; /** * reduce 是用来合并的 * reduce四个泛型： * 前两个，跟map的输出类型一样 * 后面两个泛型：reduce端的输出类型 * hello 5 * world 2 * ... */ public class WordCountReducer extends Reducer<Text, IntWritable,Text, IntWritable> { // reduce 这个方法，有多少个key值，就会调用多少次 @Override protected void reduce(Text key, Iterable<IntWritable> values, Context context) throws IOException, InterruptedException { // reduce 拿到的数据是什么样的呢 hello [1,1,1,1,1] world [1,1] int count = 0; // 第一种写法 for (IntWritable num : values) { int i = num.get(); count = count + i; } // hello 5 context.write(key,new IntWritable(count)); } }

3）编写测试代码

package com.bigdata; 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.lib.input.FileInputFormat; import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat; import java.io.IOException; public class WordCountDriver { public static void main(String[] args) throws IOException, InterruptedException, ClassNotFoundException { Configuration configuration = new Configuration(); // 使用本地的文件系统，而不是hdfs configuration.set("fs.defaultFS","file:///"); // 使用本地的资源（CPU,内存等）, 也可以使用yarn平台跑任务 configuration.set("mapreduce.framework.name","local"); Job job = Job.getInstance(configuration, "wordCount单词统计"); // 指定 map job.setMapperClass(WordCountMapper.class); // hello 1 job.setMapOutputKeyClass(Text.class); job.setMapOutputValueClass(IntWritable.class); // 设置reduceTask的数量 // reduce的数量决定了reduceTask的任务数量，每一个任务，结束后都会产生一个文件 part-r-xxxxx // 结论：reduceTask的数量可以和分区数量不一致，但是没有意义，一般两者保持一致。 job.setNumReduceTasks(1); // 指定 reduce job.setReducerClass(WordCountReducer.class); // hello 5 job.setOutputKeyClass(Text.class); job.setOutputValueClass(IntWritable.class); // 此处也可以使用绝对路径 FileInputFormat.setInputPaths(job,"D:\\project\\MapReduceDemo\\data\\input"); FileOutputFormat.setOutputPath(job,new Path("D:\\project\\MapReduceDemo\\data\\output")); boolean result = job.waitForCompletion(true); // 返回结果如果为true表示任务成功了，正常退出，否则非正常退出 System.exit(result?0:-1); } }

最后输出的结果如下：