一、案例准备

1、首先在本地创建两个文件，即文件A和文件B

touch A B

2、在文件A和文件B中分别添加以下内容
A:

China is my motherland
I love China

B:

I am from china

3、启动hadoop集群并在HDFS中创建input文件夹

//进入hadoop安装目录
cd /usr/local/hadoop//此处为我的hadoop安装目录
//启动hadoop集群
start-all.sh
//在HDFS中创建input文件夹
./bin/hadoop dfs -mkdir /input

4、将文件A和文件B上传至HDFS的input目录下

//我的文件A和文件B在/usr/local/my_data目录下，因为我的/usr/local/my_data只有文件A和文件B，所以我用*表示这个目录下的所有文件
./bin/hadoop dfs -put /usr/local/my_data/* /input 

二、案例解析

1、创建maven工程并导入hadoop相关jar包依赖

点击此链接可查看maven配置

hadoop依赖坐标

<dependencies>
        <dependency>
            <groupId>org.apache.hadoop</groupId>
            <artifactId>hadoop-common</artifactId>
            <version>2.7.4</version>
        </dependency>

        <dependency>
            <groupId>org.apache.hadoop</groupId>
            <artifactId>hadoop-hdfs</artifactId>
            <version>2.7.4</version>
        </dependency>

        <dependency>
            <groupId>org.apache.hadoop</groupId>
            <artifactId>hadoop-client</artifactId>
            <version>2.7.4</version>
        </dependency>

        <dependency>
            <groupId>junit</groupId>
            <artifactId>junit</artifactId>
            <version>4.11</version>
        </dependency>
    </dependencies>

2、编写Mapper处理逻辑

package com.itheima.mapreduce;

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;

/**
 * @author 超级无敌炫酷马少强
 * @create 2020-05-31
 */
/*
 * @Author 马少
 * @Description
 * KEYIN:默认情况下，是MapReduce所读到的一行文本的其实偏移量：Long，
 * 但是在hadoop中有自己的更精简的序列化接口，所以不直接用Long，而用LongWritable
 * VALUEIN:默认情况下，是MapReduce框架所读到的一行文本的内容：String，同上，用Text
 *
 * KEYOUT:是用户自定义逻辑处理完成之后输出数据的key，在此处是单词：String
 * VALUEOUT:是用户自定义逻辑处理完成之后输出数据的value：在此处是单词次数：Integer
 * @Date 10:58 2020/5/31
 * @Param
 * @return
 **/
public class WordCountMapper extends Mapper<LongWritable, Text, Text, IntWritable> {

        /*
         * @Author 马少
         * @Description
         * map阶段的业务逻辑就写在自定义的map()方法中
         * maptask会对每一个输入数据调用一次我们自定义的map()方法
         * @Date 11:13 2020/5/31
         * @Param [key, value, context]
         * @return void
         **/
        @Override
        public void map(LongWritable key, Text value, Context context
        ) throws IOException, InterruptedException {

            //将maptask传给我们的文本内容先转换成String
            String line = value.toString();
            //根据空格将这一行切分成单词
            String[] words = line.split(" ");

            //将单词输出以<单词，1>的格式输出
            for (String word : words) {
                //将单词作为key，将次数1作为value，以便于后续的数据分发，可以根据单词分发，
                // 以便于相同的单词会到相同的reduce task
                context.write(new Text(word), new IntWritable(1));
            }
        }
    }

3、编写reduce处理逻辑

public class WordCountReduce  extends Reducer<Text, IntWritable, Text, IntWritable>{

        /*
         * @Author 马少
         * @Description
         * <angelaba,1><angelaba,1><angelaba,1><angelaba,1>
         * <banana,1><banana,1><banana,1><banana,1>
         * <hello,1><hello,1><hello,1><hello,1><hello,1>
         * 入参key，是一组相同单词kv对的key
         *
         *
         * @Date 11:37 2020/5/31
         * @Param
         * @return
         **/

        @Override
        public void reduce(Text key, Iterable<IntWritable> values,
                           Context context
        ) throws IOException, InterruptedException {
            int count = 0;
//            迭代器写法
//            Iterator<IntWritable> iterator = values.iterator();
//            while (iterator.hasNext())
//            {
//                count += iterator.next().get();
//            }
            for (IntWritable value : values) {
                count += value.get();
            }
            context.write(key, new IntWritable(count));
        }

    }

3、编写main方法

package com.itheima.mapreduce;


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.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

import java.io.IOException;

/**
 * @author 超级无敌炫酷马少强
 * @create 2020-05-30
 */
public class WordCount {
    /*
    * @Author 马少
    * @Description
    * 相当于yarn集群的客户端
    * 需要在此封装我们的mr程序的相关运行参数，指定jar包最后提交给yarn
    * @Date 20:57 2020/5/31
    * @Param [args]
    * @return void
    **/
    public static void main(String[] args) throws Exception {

        Configuration conf = new Configuration();
        Job job = Job.getInstance(conf, "word count");
        //设置环境参数
        job.setJarByClass(WordCount.class);
        //指定本业务job要使用的mapper/reduce业务类
        job.setMapperClass(WordCountMapper.class);
        job.setReducerClass(WordCountReduce.class);
        //指定mapper输出数据的key和values类型
        job.setMapOutputKeyClass(Text.class);
        job.setMapOutputValueClass(IntWritable.class);
        //指定最终输出数据的key和values类型
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(IntWritable.class);
        //指定job输出的原始文件的所在目录
        //设置输入文件
        FileInputFormat.setInputPaths(job, new Path(args[0]));
        //设置输出文件
        FileOutputFormat.setOutputPath(job, new Path(args[1]));

        System.exit(job.waitForCompletion(true) ? 0 : 1);
    }

}

三、打包上传

1、将项目打成jar包

2、利用Xshll或Xftp上传jar包到虚拟机中

四、运行案例

1、运行案例

//运行命令
./bin/hadoop jar my_hadoop-1.0-SNAPSHOT.jar com/itheima/mapreduce/WordCount /input /output

命令解析
./bin/hadoop jar表示运行jar包（固定写法）
my_hadoop-1.0-SNAPSHOT.jar表示的是jar包的名称
com/itheima/mapreduce/WordCount表示的是WordCount类的全限定类名
/input表示在HDFS中我们需要进行统计的文件目录
/output表示在HDFS中我们存放输出结果的目录（注意：输出目录不能提前存在，应有程序自动生成）

2、查看运行结果

./bin/hadoop fs -ls /output

出现以下结果表示运行成功

3、查看结果内容

./bin/hadoop fs -cat /output/part-r-00000

• 以上我们就完成了MapReduce程序实现WordCount的案例啦！！！