MapReduce是一个分布式运算程序的编程框架，是用户开发“基于Hadoop的数据分析应用”的核心框架。
MapReduce核心功能是将用户编写的业务逻辑代码和自带默认组件整合成一个完整的分布式运算程序，并发运行在一个Hadoop集群上。
(1)MapReduce易于编程
它简单的实现一些接口，就可以的完成一个分布式程序。
(2)良好的扩展性
可以通过简单的增加机器来扩展。
(3)高容错性
其中一台机器挂了，它可以把上面的计算任务转移到另外另外一个节点进行计算。
(4)适合PB级以上海量数据的离线处理
(1)不擅长实时计算
(2)不擅长流式计算
MapReduce的输入数据集是静态的
(3)不擅长DAG（有向无环图）计算
每个MapReduce作业的输出结果都会写入到磁盘，会造成大量的磁盘IO，导致性能非常的低下。
重点注意：
（1）map阶段数据是怎么切分的
（2）reduce阶段数据是怎么合并的
一个完整的MapReduce程序在分布式运行时有三类实例进程：
（1）MrAppMaster：负责整个程序的过程调度机状态协调
（2）MapTask：负责Map阶段的整个数据处理流程
（3）ReduceTask：负责Reduce阶段的整个数据处理流程。
| Java类型 | Hadoop Writer类型 |
|---|---|
| Boolean | BooleaWriter |
| Byte | ByteWriter |
| int | intWriter |
| Float | FloatWriter |
| Long | LongWriter |
| Double | DoubleWriter |
| String | Text |
| Map | MapWriter |
| Array | ArrayWriter |
| Null | NullWriter |
（1）Mapper阶段
①用户自定义的Mapper要继承自己的父类
②Mapper的输入数据是LongWriter和Text类型的。
③Mapper中的业务逻辑写在map方法中
④Mapper的输出数据是Text和IntWriter类型的。
⑤map方法对每一个<k,v>调用一次
（2）Reduce阶段
①用户自定义的Reducer要继承自己的父类
②Reducer的输入数据类型对应Mapper的输出数据类型，也是KV
③Reducer的业务逻辑写在reduce方法中
④ReduceTask进程对每一组相同k的<k，v>组调用一次reduce方法
（3）Driver阶段
相当于YARN集群的客户端，用于提交我们整个程序到YARN集群，提交的是封装了 MapReduce程序相关参数的job对象
1)需求
在给定的文本文件中统计输出每一个单词出现的总次数
（1)输入数据到hello.txtatguigu atguigu ss ss cls cls jiao banzhang xue hadoop（2)期望输出数据
atguigu 2 banzhang 1 cls 2 hadoop 1 jiao 1 ss 2 xue 1
2)需求分析
按照MapReduce编程规范，分别编写Mapper，Reducer，Driver。可参考下图：
3）环境准备
（1）创建maven工程
（2）在pom.xml文件中添加如下依赖<dependencies> <dependency> <groupId>junit</groupId> <artifactId>junit</artifactId> <version>4.12</version> </dependency> <dependency> <groupId>org.apache.logging.log4j</groupId> <artifactId>log4j-slf4j-impl</artifactId> <version>2.12.0</version> </dependency> <dependency> <groupId>org.apache.hadoop</groupId> <artifactId>hadoop-client</artifactId> <version>3.1.3</version> </dependency> </dependencies>在项目的src/main.resource目录下，新建一个文件，命名为“log4j2.xml”，在文件中填入。
<?xml version="1.0" encoding="UTF-8"?> <Configuration status="error" strict="true" name="XMLConfig"> <Appenders> <!-- 类型名为Console，名称为必须属性 --> <Appender type="Console" name="STDOUT"> <!-- 布局为PatternLayout的方式， 输出样式为[INFO] [2018-01-22 17:34:01][org.test.Console]I'm here --> <Layout type="PatternLayout" pattern="[%p] [%d{yyyy-MM-dd HH:mm:ss}][%c{10}]%m%n" /> </Appender> </Appenders> <Loggers> <!-- 可加性为false --> <Logger name="test" level="info" additivity="false"> <AppenderRef ref="STDOUT" /> </Logger> <!-- root loggerConfig设置 --> <Root level="info"> <AppenderRef ref="STDOUT" /> </Root> </Loggers> </Configuration>(3)编写程序
①编写Mapper类public class WcMapper extends Mapper<LongWritable, Text,Text, IntWritable> { private Text k = new Text(); private IntWritable v = new IntWritable(1); @Override protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException { //以空格分割字符 String[] words = value.toString().split(" "); //输出 for (String word : words) { k.set(word); context.write(k,v); } } }②编写Reducer类
public class WcReducer extends Reducer<Text, IntWritable,Text,IntWritable> { private int sum; private IntWritable v = new IntWritable(); @Override protected void reduce(Text key, Iterable<IntWritable> values, Context context) throws IOException, InterruptedException { //累加求和 sum = 0; for (IntWritable value : values) { sum += value.get(); } //输出 v.set(sum); context.write(key,v); } }③编写Driver类，这里我注释掉了Windows环境和集群交互的情况
public class WcDriver{ public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException { //获取job对象 Configuration configuration = new Configuration(); // //设置HDFS NameNode的地址 // configuration.set("fs.defaultFS", "hdfs://hadoop102:9820"); // // 指定MapReduce运行在Yarn上 // configuration.set("mapreduce.framework.name","yarn"); // // 指定mapreduce可以在远程集群运行 // configuration.set("mapreduce.app-submission.cross-platform","true"); // //指定Yarn resourcemanager的位置 // configuration.set("yarn.resourcemanager.hostname","hadoop103"); Job job = Job.getInstance(configuration); //设置driver的jar包 job.setJarByClass(WcDriver.class); >//job.setJar("D:\\code\\mapreduce\\target\\mapreduce-1.0-SNAPSHOT.jar"); //设置mapper和reducer的jar包 job.setMapperClass(WcMapper.class); job.setReducerClass(WcReducer.class); //设置mapper的输出类型 job.setMapOutputKeyClass(Text.class); job.setMapOutputValueClass(IntWritable.class); //设置最终输出kv类型 job.setOutputKeyClass(Text.class); job.setOutputValueClass(IntWritable.class); //设置输入路径和输出路径 FileInputFormat.setInputPaths(job,new Path(args[0])); FileOutputFormat.setOutputPath(job,new Path(args[1])); //提交job boolean result = job.waitForCompletion(true); System.exit(result? 0 : 1); } }④编辑任务配置
（a）检查第一个参数Main class是不是我们要运行的类的全类名，如果不是的话一定要修改！
（b）在VM options后面加上 ：-DHADOOP_USER_NAME=atguigu
（c）在Program arguments后面加上两个参数分别代表输入输出路径，两个参数之间用空格隔开。如：hdfs://hadoop102:9820/input hdfs://hadoop102:9820/output
序列化就是把内存中的对象，转换成字节序列（或其他数据传输协议）以便于存储到磁盘（持久化）和网络传输。
序列化可以存储“活的”对象，可以将“活的”对象发送到远程计算机。
Java的序列化框架在序列化过程中会产生很多额外的信息，不便于在网络中高效传输，所以Hadoop自己开发了一套序列化机制。
Hadoop序列化特点：
（1）紧凑：高效使用存储空间
（2）快速：读写数据的额外开销小
（3）可扩展：随着通信协议的升级而可升级
（4）互操作：支持多语言的交互
具体实现bean对象的步骤
（1）必须实现Writable接口
（2）重写tostring方法
（3）重写序列化方法
（4）重写反序列化方法
1)需求
统计每一个手机耗费的总上行流量、总下行流量、总流量
（1）输入数据（有需要的留言）
（2）输入数据格式
| 7 | 13560436666 | 120.196.100.99 | 1116 | 954 | 200 |
|---|---|---|---|---|---|
| id | 手机号码 | 网络ip | 上行流量 | 下行流量 | 网络状态码 |
期望输出数据格式
| 13560436666 | 1116 | 954 | 2070 |
|---|---|---|---|
| 手机号码 | 上行流量 | 下行流量 | 总流量 |
2）需求分析，参考下图：
3)编写MapReduce程序
（1）编写流量统计的bean对象public class FlowBean implements Writable { private long upFlow; private long downFlow; private long sumFlow; @Override public String toString() { return "FlowBean{" + "upFlow=" + upFlow + ", downFlow=" + downFlow + ", sumFlow=" + sumFlow + '}'; } public void set(long upFlow,long downFlow){ this.upFlow = upFlow; this.downFlow = downFlow; this.sumFlow = upFlow + downFlow; } public long getUpFlow() { return upFlow; } public void setUpFlow(long upFlow) { this.upFlow = upFlow; } public long getDownFlow() { return downFlow; } public void setDownFlow(long downFlow) { this.downFlow = downFlow; } public long getSumFlow() { return sumFlow; } public void setSumFlow(long sumFlow) { this.sumFlow = sumFlow; } /** *序列化 * @param dataOutput * @throws IOException */ public void write(DataOutput dataOutput) throws IOException { dataOutput.writeLong(upFlow); dataOutput.writeLong(downFlow); dataOutput.writeLong(sumFlow); } /** * 反序列化 * @param dataInput * @throws IOException */ public void readFields(DataInput dataInput) throws IOException { upFlow= dataInput.readLong(); downFlow = dataInput.readLong(); sumFlow = dataInput.readLong(); } }(2)编写Mapper类
public class FlowMapper extends Mapper<LongWritable,Text,Text,FlowBean> { private Text ph = new Text(); private FlowBean flowBean = new FlowBean(); @Override protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException { String[] filds = value.toString().split("\t"); ph.set(filds[1]); flowBean.set( Long.parseLong( filds[filds.length -3]), Long.parseLong(filds[filds.length - 2]) ); context.write(ph,flowBean); } }(3)编写Reducer类
public class FlowReducer extends Reducer<Text,FlowBean,Text,FlowBean> { private FlowBean result = new FlowBean(); @Override protected void reduce(Text key, Iterable<FlowBean> values, Context context) throws IOException, InterruptedException { long up = 0; long down = 0; for (FlowBean value : values) { up += value.getUpFlow(); down += value.getDownFlow(); } result.set(up,down); context.write(key,result); } }(4)编写driver类
public class FlowDriver { public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException { //获取job对象 Configuration configuration = new Configuration(); // //设置HDFS NameNode的地址 // configuration.set("fs.defaultFS", "hdfs://hadoop102:9820"); // // 指定MapReduce运行在Yarn上 // configuration.set("mapreduce.framework.name","yarn"); // // 指定mapreduce可以在远程集群运行 // configuration.set("mapreduce.app-submission.cross-platform","true"); // //指定Yarn resourcemanager的位置 // configuration.set("yarn.resourcemanager.hostname","hadoop103"); Job job = Job.getInstance(configuration); //设置driver的jar包 job.setJarByClass(FlowDriver.class); //job.setJar("D:\\code\\mapreduce\\target\\mapreduce-1.0-SNAPSHOT.jar"); //设置mapper和reducer的jar包 job.setMapperClass(FlowMapper.class); job.setReducerClass(FlowReducer.class); //设置map的输出类型 job.setMapOutputKeyClass(Text.class); job.setMapOutputValueClass(FlowBean.class); //设置最终kv的输出类型 job.setOutputKeyClass(Text.class); job.setOutputValueClass(FlowBean.class); //设置输入输出路径 FileInputFormat.setInputPaths(job,new Path(args[0])); FileOutputFormat.setOutputPath(job,new Path(args[1])); //提交 boolean b = job.waitForCompletion(true); System.exit(b? 0 : 1); } }
