Hadoop--MapRecude入门

Hadoop--MapRecude入门 什么是MR

MR是一个分布式运算程序的编程框架，核心功能是将用户编写的业务逻辑和自带默认组件整合为一个分布式运算程序，并发的运行在集群中。

MR的优点

1. 易于编程 简单的实现一些接口，就可以完成一个分布式运算程序

2. 良好的扩展性 资源不足时可以增加集群中的机器来增强运算能力

3. 高容错性 当集群中的一台机器挂掉时，MR可以把上面的任务转移到其他机器上继续执行

4. 适合PB/TB级的海量数据处理

MR的缺点

1. 不擅长实时的计算 毫秒/秒级的运算
2. 不擅长流式计算 动态的任务，来一条处理一条
3. 不擅长有向无环图计算 对于上一个任务的结果作为下一个任务的输入类型的计算任务。

MR核心思想

1. 往往需要分为两个阶段

2. 第一个阶段，Map Task并发，分开运行互不干涉

3. 第二个阶段，Reduce Task并发，分开运行互不相干，输入依赖上一个阶段的输出。

4. MR只能包含一个Map和一个Reduce；当需要处理的任务过于复杂时，需要多个MR程序串行运行。

MR进程

一个完整的MR程序在分布式运行时有三类实例进程：

1. MrAppMaster 负责整个程序的过程调度及状态调度
2. MapTask 负责Map阶段的整个数据处理流程
3. ReduceTask：负责Reduce阶段的整个数据处理流程

MR编程规范 Mapper

1. 用户自定义的Mapper要继承已有的父类
2. Mapper的输入数据以kv对形式
3. Mapper中的业务逻辑必须写在map()方法中
4. Mapper的输出形式为kv对
5. map方法（MapTask进程）对每一个kv仅调用一次

Reducer

1. 用户自定义的Reducer要继承已有的父类
2. Reducer的输入类型要与Mapper的输出类型一致
3. Reducer的业务逻辑写在reduce()方法中
4. reduce方法（ReduceTask进程）对每一组相同k的kv组调用一次（在mapper和reducer的中间过程会将mapper的输出进行一定的处理）

Driver

​ 可以看成Yarn的客户端，用于提交封装了MR程序相关运行参数的job对象到Yarn集群

WordCount案例 明确需求

文件中的单词形式

a a

c c c

b

输出形式

a 2

b 1

c 3

编写过程/步骤

在Mapper中数据以kv对输入,类型为,k是该行在文件中偏移量，v是该行的数据；先将Text类型转为String，再进行数据处理，输出kv是,k是文本数据，v是1，即这个单词出现一次。

在Reducer中数据以Mapper中输出kv作为输入，对kv组的v遍历累加，得到k的总个数，输出.

Driver即main主类，获取配置参数，初始化一个Job对象，给Job对象设置设置Mapper输出、最终输出、文件输入输出路径等信息。提交Job给Yarn。

WordCountMapper类

public class WordCountMapper extends Mapper
{
    Text k;
    IntWritable v;
    
    public WordCountMapper() {
        this.k = new Text();
        this.v = new IntWritable(1);
    }
    
    protected void map(LongWritable key, Text value, Mapper.Context context) throws IOException, InterruptedException {
        String line = value.toString();
        String[] split;
        String[] words = split = line.split(" ");
        for (String word : split) {
            this.k.set(word);
            context.write((Object)this.k, (Object)this.v);
        }
    }
}

WordCountReducer类

public class WordCountReducer extends Reducer
{
    private int sum;
    IntWritable v;
    
    public WordCountReducer() {
        this.v = new IntWritable();
    }
    
    protected void reduce(Text key, Iterable values,Reducer.Context context) throws IOException, InterruptedException {
        this.sum = 0;
        for (IntWritable value : values) {
            this.sum += value.get();
        }
        this.v.set(this.sum);
        context.write((Object)key, (Object)this.v);
    }
}

WordCountDriver类

public class WordCountDriver
{
    public static void main(String[] args) throws IOException, InterruptedException, ClassNotFoundException {
        Configuration conf = new Configuration();
        Job job = Job.getInstance(conf);
        job.setJarByClass((Class)WordCountDriver.class);//设置主类jar包
        job.setMapperClass((Class)WordCountMapper.class);//设置Mapper
        job.setReducerClass((Class)WordCountReducer.class);//设置Reducer
        job.setMapOutputKeyClass((Class)Text.class);//设置map方法的输出k类型
        job.setMapOutputValueClass((Class)IntWritable.class);//设置map方法的输出v类型
        job.setOutputKeyClass((Class)Text.class);//设置最终k的输出类型
        job.setOutputValueClass((Class)IntWritable.class);//设置最终v的输出类型
        FileInputFormat.setInputPaths(job,  new Path(args[0]));//设置输入数据路径
        FileOutputFormat.setOutputPath(job, new Path(args[1]));//设置输出数据的路径
        //job.setInputFormatClass((Class)CombineTextInputFormat.class);//设置InputFormat为CombineTextInputFormat，默认为TextInputFormat
        //CombineTextInputFormat.setMaxInputSplitSize(job, 4194304L);//一个切片最大值4mb
        boolean b = job.waitForCompletion(true);
        System.exit(b ? 0 : 1);
    }
}

注意：输入输出路径在window本地时可以直接输入地址，但要移植到Hadoop集群中，必须更改为args[0]……