MapReduce2.0处理机制

MapReduce（分布式计算模型）作为Hadoop家族一重要的家庭成员主要用于搜素领域，海量数据计算等问题。 内部模型采用分而治之的思想。MapReduce分为两部分(Map和Reduce)。其中Shuffler是对Reduce的预处理。 map和reduce的数据处理方式均采取键对的方式:即 [k1

                      mapreduce（分布式计算模型）作为hadoop家族一重要的家庭成员主要用于搜素领域，海量数据计算等问题。

                 内部模型采用"分而治之"的思想。MapReduce分为两部分(Map和Reduce)。其中Shuffler是对Reduce的预处理。

map和reduce的数据处理方式均采取键值对的方式:即  [k1,v1]->MAP->[K2,V2]->Reduce->[k3,v3]。

MR执行流程
 (1).客户端提交一个mr的jar包给JobClient(提交方式：hadoop jar ...)
 (2).JobClient通过RPC和JobTracker进行通信，返回一个存放jar包的地址（HDFS）和jobId
 (3).client将jar包写入到HDFS当中(path = hdfs上的地址 + jobId)
 (4).开始提交任务(任务的描述信息，不是jar, 包括jobid，jar存放的位置，配置信息等等)
 (5).JobTracker进行初始化任务
 (6).读取HDFS上的要处理的文件，开始计算输入分片，每一个分片对应一个MapperTask
 (7).TaskTracker通过心跳机制领取任务（任务的描述信息）
 (8).下载所需的jar，配置文件等
 (9).TaskTracker启动一个java child子进程，用来执行具体的任务（MapperTask或ReducerTask）
 (10).将结果写入到HDFS当中

在hadoop2.0以上版本中JobTracker取名为RM(resourceManage)  TastTracker取名为NM(nodeManage)

mapReduce操作实现wordcount功能(即从文本中读取内容，计算出每个单词出现的次数)

程序分为3个类(自定义MAP方法功能实现，自定义REDUCE方法功能实现，最后类拼凑成mapreduce模式导成jar包，在HDFS分布式功能中实现)

1.WCMapper类（实现map）

import java.io.IOException;

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

/*
 * 给wordcount写mapper
 * 定义mapper
 * KEYIN:k1的类型
 * VALUEIN：v1的类型
 *
 * 重写map方法
 * hadoop没有使用jdk默认的序列化机制(long->longwriteable String->Text)
 */
public class WCMapper extends Mapper {

 @Override
 protected void map(LongWritable key, Text value,
   Mapper.Context context)
   throws IOException, InterruptedException {
  // TODO Auto-generated method stub

  // 接收信息V1
  String line = value.toString();
  // 切分数据
  String[] words = line.split(" ");
  // 循环
  for (String w : words) {
   // 出现一次记一个1,输出
   // 构一个新的key,value
   context.write(new Text(w), new LongWritable(1));
  }
 }

}

2.WCReducer类实现reduce功能

import java.io.IOException;

import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;

/*
 * KEYIN k2的类型
 * VALUEIN v2的类型
 *
 * 重写reducer方法
 */
public class WCReducer extends Reducer {

 @Override
 protected void reduce(Text k2, Iterable v2s,
   Reducer.Context context)
   throws IOException, InterruptedException {
  // 接收数据
  Text k3 = k2;
  // 定义一个计数器
  Long count = (long) 0;
  // 循环v2s
  for (LongWritable i : v2s) {
   count += i.get();
  }
  // 输出
  context.write(k3, new LongWritable(count));
 }

}

3.wordCount类。拼凑前两个类，符合mapreduce格式

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
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;

/*
 * mapReduce
 *
 * 组装自定义的map和reduce
 */
public class wordCount {
 public static void main(String[] args) throws Exception {
  // Job job=Job.instance(new Configuration()); //版本hadoop2
  Job job = new Job(new Configuration()); // 版本hadoop1

  // 4.注意---将main方法中的类设进去
  job.setJarByClass(wordCount.class);

  // 1.设置自定义Mapper
  job.setMapperClass(WCMapper.class);
  job.setMapOutputKeyClass(Text.class);
  job.setMapOutputValueClass(LongWritable.class);

  // 设置mapper读入的path(hdfs路径)
  FileInputFormat.setInputPaths(job, new Path("/words.txt"));

  // 2.设置reduce
  job.setReducerClass(WCReducer.class);
  job.setOutputKeyClass(Text.class);
  job.setOutputValueClass(LongWritable.class);

  FileOutputFormat.setOutputPath(job, new Path("/WcountResult"));

  // 3.提交
  job.waitForCompletion(true); // 打印进度和详情
 }
}