疯狂Hadoop之MapReduce入门（五）

MapReduce计算模型框架

MapReduce计算模型介绍

Hadoop MapReduce设计构思

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

既然是做计算的框架，那么表现形式就是有一个输入（input），MapReduce操作这个输入（input），通过本身定义好的计算模型，得出一个输出（output）。

对许多开发者来说，自己完完全全实现一个并行计算程序难度太大，而MapReduce就是一种简化并行计算的编程模型，降低了开发并行应用的入门门槛。

Hadoop MapReduce构思体现在如下三个方面：

1. 如何对付大数据处理：分而治之

对相互间不具有计算依赖的大数据，实现并行最自然的办法就是采取分而治之的策略。并行计算的第一个重要问题是如何划分计算任务或者计算数据以便对划分的子任务或数据块同时进行计算。不可拆分的计算任务或相互间有依赖关系的数据无法进行并行计算

1. 构建抽象模型：Map和Reduce

MapReduce借鉴了函数式语言中的思想，用Map和Reduce两个函数提供了高 层的并行编程抽象模型

Map:对一组数据元素进行某种重复式的处理

Reduce:对Map的中间结果进行某种进一步的结果整理

MapReduce中定义了如下的Map和Reduce两个抽象的编程接口，由用户去编 程实现

1. 统一架构，隐藏系统层细节

如何统一的计算框架，如果没有统一封装底层细节，那么程序则需要考虑诸如数据存储、划分、分发、结果收集、错误恢复等诸多细节。为此，MapReduce设计并提供了统一的计算框架，为程序员隐藏了绝大多数系统层面的处理细节。

MapReduce框架结构

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

MRAppMaster:负责整个程序的过程调度及状态协调

MapTask：负责Map阶段的整个数据处理流程

ReduceTask:负责Reduce阶段的整个数据处理流程

编程规范

1. 用户编写的程序分成三个部分：Mapper、Reducer、Driver(提交运行MR程序的 客户端)
2. Mapper的输入数据是KV对形式（KV类型可自定义）
3. Mapper的输出数据是KV对形式（KV类型可自定义）
4. Mapper中的业务逻辑写在map()方法中
5. map()方法（MapTask进程）对每一个<K,V>调用一次
6. Reducer的输入数据类型对应Mapper的输出数据类型，也是KV
7. Reducer的业务逻辑写在reduce()方法中
8. Reducetask进程对每一组相同k的<k,v>组调用一次reduce()方法
9. 用户自定义的Mapper和Reducer都要继承各自父类
10. 整个程序需要一个Driver来进行提交，提交的是一个描述了各种必要信息的 job对象

理解MapReduce思想

MapReduce思想在生活中处处可见。或多或少都曾接触过这种思想。MapReduce的思想核心是“分而治之”，适用于大量复杂的任务处理场景（大规模数据处理场景）。即使是发布过论文实现分布式计算的谷歌也只是实现了这种思想，而不是原创。

Map负责“分”，即把复杂的任务分解为若干个“简单的任务”开并行处理。可以进行拆分的前提是这些小任务可以并行计算，彼此间几乎没有依赖关系。

Reduce负责“合”，即对map阶段的结果进行全局汇总

这两个阶段合起来正是MapReduce思想的体现

生活中的MapReduce思想

假设需要数图书馆的所有图书，那么可以找多个人，一人负责数一个书架上的图书，这就是Map，人越多，数书的时间就越少

每个人数完各自书架的图书之后，把所有人的数据统计到一起，这就是Reduce

WorldCount示例代码

案例1：在一堆给定的文本文件中统计输出每个单词出现的次数

定义一个mapper类

[code]package hdfsFirst;

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:表示mapper数据输入时候KEY的数据类型，在默认的读取数据组件下，叫InputFormat,它的行为是一行一行的读取待处理的数据
* 读取一行，返回一行给我们的MR程序。这种情况下，KEYIN就表示每一行的起始偏移量，因此数据类型为Long
*
* VALUEIN:表示mapper数据输入时候VALUE的数据类型，在默认的读取数据组件下，valuein就表示读取的这一行内容，因此数据类型为String
*
* KEYOUT:表示mapper数据输出时候KEY的数据类型，在本案例中，输出的key是单词，因此数据类型是String
*
* VALUEOUT:表示mapper数据输出时候VALUE的数据类型，在本案例中，输出的value是单词的次数，因此数据类型是Integer
*
* 在hadoop中拥有自己封装的数据类型
* long------LongWritable
* String----Text
* Integer----Intwritable
* null------Nullwritable
* @author gw
*
*/
public class WordCountMapper extends Mapper<LongWritable, Text, Text, IntWritable>{

@Override
protected void map(LongWritable key, Text value, Mapper<LongWritable, Text, Text, IntWritable>.Context context)
throws IOException, InterruptedException {
//拿到传入进来的一行内容，把数据类型转换为String
String line = value.toString();
//将这一行内容按照分隔符进行一行内容的切割，切割成一个单词数组
String[] words = line.split(" ");
//遍历数组，每出现一个单词，就标记一个数字1，如<单词，1>
for (String word : words) {
/*
* 使用MR程序的上下文context把mapper阶段处理的数据发送出去
* 作为reduce节点的输入数据
*/
context.write(new Text(word), new IntWritable(1));
}
}
}

定义一个reducer类

[code]package hdfsFirst;

import java.io.IOException;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;
/**
* KEYIN:表示reducer阶段输入数据key的类型,对应mapper的输出key类型，在本案例中就是单词--Text
* VALUEIN:表示reducer阶段输入数据value的类型,对应mapper的输出value类型，在本案例中就是单词次数--IntWritable
* KEYOUT:表示reducer阶段输出数据key的类型,在本案例中就是单词--Text
* VALUEOUT:表示reducer阶段输出数据value的类型，在本案例中就是单词总次数--IntWritable
* @author gw
*
*/
public class WordCountReducer extends Reducer<Text, IntWritable, Text, IntWritable>{

@Override
protected void reduce(Text key, Iterable<IntWritable> values,Context context) throws IOException, InterruptedException {
//定义一个计数器
int count = 0;
//遍历一组迭代器，把每一个数量1累加起来就构成单词总次数
for (IntWritable value : values) {
count += value.get();
}
//把最终结果输出
context.write(key, new IntWritable(count));
}
}

定义一个主类，用来描述job并提交job

[code]package hdfsFirst;

import java.io.IOException;

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 WordCountRunner {
public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException {
//通过job来封装本次mr的相关信息
Configuration conf = new Configuration();
Job job = Job.getInstance(conf);
//指定本次mr job jar包运行主类
job.setJarByClass(WordCountRunner.class);
//指定本次mr所用的mapper和reducer类
job.setMapperClass(WordCountMapper.class);
job.setReducerClass(WordCountReducer.class);

job.setMapOutputKeyClass(Text.class);
job.setMapOutputValueClass(IntWritable.class);

job.setOutputKeyClass(Text.class);
job.setOutputValueClass(IntWritable.class);

FileInputFormat.setInputPaths(job, "/nihao/bb.txt");
FileOutputFormat.setOutputPath(job, new Path("/wordcount/bboutput"));

//提交程序，并且监听打印程序情况
boolean b = job.waitForCompletion(true);
System.exit(b?0:1);
}
}

查看运行结果

命令：hadoop dfs -text /wordcount/output/part-r-00000