hadoop-MR

1. MR执行原理

1. MAP阶段

在这里插入图片描述

2. Reducer

在这里插入图片描述

3.shuffle阶段

在这里插入图片描述

2.实操

1. 导入 maven 配置

 <!--  MR 操作  -->
<dependency>
    <groupId>org.apache.hadoop</groupId>
    <artifactId>hadoop-mapreduce-client-common</artifactId>
    <version>2.7.1</version>
</dependency>

<!-- https://mvnrepository.com/artifact/org.apache.hadoop/hadoop-mapreduce-client-core -->
<dependency>
    <groupId>org.apache.hadoop</groupId>
    <artifactId>hadoop-mapreduce-client-core</artifactId>
    <version>2.7.1</version>
</dependency>

2. 配置自定义的 Mapper

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;

/**
 *  第一个参数： 读取到的字节位置
 *  第二个参数:  读取到的内容的类型
 *  第三个参数： 输出的key的类型
 *  第四个参数： 输出的value的类型
 */
public class WorkMapper extends Mapper<LongWritable, Text, Text, IntWritable> {
    /**
     * 读取的内容
     *          hello a
     *          hello b
     *          hello c
     *          hello d
     * 异步方法，每次执行一行
     * @param key  开始的字符数 此处就是 0 8 16 24
     * @param value 读取到指定行的文本   如 hello a ...
     * @param context
     * @throws IOException
     * @throws InterruptedException
     */
    @Override
    protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
        String[] lineValues = value.toString().split(" ");
        for (String line : lineValues) {
            context.write(new Text(line), new IntWritable(1));
        }
    }
}

3. 配置Reducer

import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;
import java.io.IOException;

/**
 *  mapper >> reducer
 * 第一个参数  输入的第一个参数  
 * 第二个参数  输入的第二个参数
 * 第三个参数 输出的第一个参数
 * 第四个参数 输出的第二个参数
 */
public class WorkReduce extends Reducer<Text, IntWritable, Text, IntWritable> {
    @Override
    protected void reduce(Text key, Iterable<IntWritable> values, Context context) throws IOException, InterruptedException {
        Integer count = 0;
        for (IntWritable value : values) {
            count += value.get();
        }
        context.write(key, new IntWritable(count));
    }
}

4. 启动类

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;

public class WorkDriver {
    public static void main(String[] args) throws  Exception {
        // 设置操作Hadoop的用户,如果用户不是root，需要做此操作
        System.setProperty("HADOOP_USER_NAME","root");
        // windows 需要设置，并且要配置 环境变量
        System.setProperty("hadoop.home.dir", "E:\\dev\\hadoop");

        //1. 创建连接类
        Configuration configuration = new Configuration();
        configuration.set("fs.defaultFS","hdfs://192.168.80.111:9000");

        // 2. 创建查询 任务
        Job job = Job.getInstance(configuration);
        // 指定主执行类
        job.setJarByClass(WorkDriver.class);
        // 指定的 Mapper 类
        job.setMapperClass(WorkMapper.class);
        //指定 Mapper输出类型
        job.setMapOutputKeyClass(Text.class);
        job.setMapOutputValueClass(IntWritable.class);

        // 指定 自定义的 Reducer
        job.setReducerClass(WorkReduce.class);
        // 指定 Reducer 输出的类型
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(IntWritable.class);

        //输入文件的地址
        FileInputFormat.setInputPaths(job,new Path("/input/wordcount"));
        // 输出文件的地址  地址要不存在
        FileOutputFormat.setOutputPath(job,new Path("/output/out1"));
        boolean isSuccess = job.waitForCompletion(true);//执行
        System.out.println(isSuccess);
    }
}