一、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);
    }
}

最后输出的结果如下：

4）遇到的错误

1、输出路径已经存在

删掉输出的目录，再次运行即可

2、双击运行 winutils.exe后报错

为了验证当前环境是否适配，我们有时候会先运行一下winutils.exe看看是否会报错

下载并安装微软官方的「VC++ 2015-2022 运行库 (x64)」，务必选择 64 位版本，我的百度网盘已经下好