mapreduce是一个运算框架，让多台机器进行并行进行运算，

他把所有的计算都分为两个阶段，一个是map阶段，一个是reduce阶段

 

map阶段：读取hdfs中的文件，分给多个机器上的maptask，分文件的时候是按照文件的大小分的

比如每个maptask都会处理128M的文件大小，然后有个500M的文件，就会启动ceil（500/128）个maptask

每读取文件的一行的处理，需要自己去写，注意每个maptask的处理逻辑都是一样的

处理出来的结果一定是一对key和value。

maptask里面的方法叫map(long k, string v, context);   k是文件的起始偏移量，v是内容，

context是要把产生的key，value对放入的容器。

 

 

 

reduce阶段：每个机器上有reducetask，其作用是对maptask产生的key和value进行聚合

聚合的原则是key一样的一定分发给一个reducetask，这个操作叫做shuffle

然后把相同key的数据作为一组进行处理。最后会把结果写到hdfs里面

每有几个reducetask，就会生成几个part-r-xxxx文件

reducetask里面的方法reduce(k,value迭代器，context),k就是key，迭代器遍历每一个key相同的value，然后context就是写入hdfs里的，也是一个key和value

 

入门样例：wordcount

设计思路，每个maptask读取文件，

map里面k  起始偏移量没用，我们每读一行v，产生就是key是每一个单词，然后value就定为1就行，把这个key，value放入context里面

在reduce阶段，每个key相同的就会作为一组，也就是单词相同的作为一组，就统计出现几次就行。

 

开始在esclipe写mapreduce的业务逻辑，首先我们需要一些jar包，相关的jar包在解压出来的hadoop下的share/hadoop文件夹下

这几个文件夹下的jar包和这几个文件夹下的lib下的jar包都拷贝到esclipe再buildPath

首先编写mapper方法

package test;

import java.io.IOException;

import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;

/**
 * KEYIN ：是map task读取到的数据的key的类型，是一行的起始偏移量Long
 * VALUEIN:是map task读取到的数据的value的类型，是一行的内容String
 * 
 * KEYOUT：是用户的自定义map方法要返回的结果kv数据的key的类型，在wordcount逻辑中，我们需要返回的是单词String
 * VALUEOUT:是用户的自定义map方法要返回的结果kv数据的value的类型，在wordcount逻辑中，我们需要返回的是整数Integer
 * 
 * 
 * 但是，在mapreduce中，map产生的数据需要传输给reduce，需要进行序列化和反序列化，而jdk中的原生序列化机制产生的数据量比较冗余，就会导致数据在mapreduce运行过程中传输效率低下
 * 所以，hadoop专门设计了自己的序列化机制，那么，mapreduce中传输的数据类型就必须实现hadoop自己的序列化接口
 * 
 * hadoop为jdk中的常用基本类型Long String Integer Float等数据类型封住了自己的实现了hadoop序列化接口的类型：LongWritable,Text,IntWritable,FloatWritable
 * 
 * 
 * 
 * 
 */
//第一个泛型为起始偏移量，没啥用，第二个为字符串，为读取到的一行内容，第三个，第四个为context中的key，和value，即发送给reduce的k，v对
public class WordcountMapper extends Mapper<LongWritable, Text, Text, IntWritable>{
    
    @Override
    //重写map方法
    protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
        // 切单词
        String line = value.toString();
        String[] words = line.split(" ");
        for(String word:words){
            context.write(new Text(word), new IntWritable(1));
        }
    }
}

接下来是reduce类

package test;

import java.io.IOException;
import java.util.Iterator;

import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;
//第一个，第二个为接收到的map的key，value，第三第四为写入到hdfs的key，value
public class WordcountReducer extends Reducer<Text, IntWritable, Text, IntWritable>{
    
    
    @Override
    //一个key，众多value的迭代器，一个context；
    protected void reduce(Text key, Iterable<IntWritable> values,Context context) throws IOException, InterruptedException {
    
        int count = 0; 
        Iterator<IntWritable> iterator = values.iterator();
        while(iterator.hasNext()){
            
            IntWritable value = iterator.next();
            count += value.get();
        }
        context.write(key, new IntWritable(count));
    }
    
    

}

 

 

 然而我们写的程序需要提交给我们的hadoop集群去运行，而管理这个事情的就是我们的yarn

yarn是一个分布式程序的运行调度平台

yarn中有两大核心角色：

1、Resource