大数据-Hadoop-MapReduce(一)：离线数据编程分析框架【Java语言】【运算效率低；被SparkCore的RDD编程取代】【不擅长实时计算、不擅长流式计算、不擅长DAG计算】

灰太狼 2023-10-08 23:19 20阅读 0赞

## ![da4f273ff0b54fb8b79c6c0cbec5d630.png][] ##

## 第1章 MapReduce概述 ##

### 1.1 MapReduce定义 ###

MapReduce是一个分布式运算程序的编程框架，是用户开发“基于Hadoop的数据分析应用”的核心框架。

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

### 1.2 MapReduce优缺点 ###

#### 1.2.1 优点 ####

**1****）MapReduce****易于编程**

它简单的实现一些接口，就可以完成一个分布式程序，这个分布式程序可以分布到大量廉价的PC机器上运行。也就是说你写一个分布式程序，跟写一个简单的串行程序是一模一样的。就是因为这个特点使得MapReduce编程变得非常流行。

**2****）良好的扩展性**

当你的计算资源不能得到满足的时候，你可以通过简单的增加机器来扩展它的计算能力。

**3****）高容错性**

MapReduce设计的初衷就是使程序能够部署在廉价的PC机器上，这就要求它具有很高的容错性。比如其中一台机器挂了，它可以把上面的计算任务转移到另外一个节点上运行，不至于这个任务运行失败，而且这个过程不需要人工参与，而完全是由Hadoop内部完成的。

**4****）适合PB****级以上海量数据的离线处理**

可以实现上千台服务器集群并发工作，提供数据处理能力。

#### 1.2.2 缺点 ####

**1****）不擅长实时计算**

MapReduce无法像MySQL一样，在毫秒或者秒级内返回结果。

**2****）不擅长流式计算**

流式计算的输入数据是动态的，而MapReduce的输入数据集是静态的，不能动态变化。这是因为MapReduce自身的设计特点决定了数据源必须是静态的。

**3****）不擅长DAG****（有向无环图）计算**

多个应用程序存在依赖关系，后一个应用程序的输入为前一个的输出。在这种情况下，MapReduce并不是不能做，而是使用后，每个MapReduce作业的输出结果都会写入到磁盘，会造成大量的磁盘IO，导致性能非常的低下。

### 1.3 MapReduce核心思想 ###

![1dfcfbc034c145c4bc285b39ef47e47d.png][]

（1）分布式的运算程序往往需要分成至少2个阶段。

（2）第一个阶段的MapTask并发实例，完全并行运行，互不相干。

（3）第二个阶段的ReduceTask并发实例互不相干，但是他们的数据依赖于上一个阶段的所有MapTask并发实例的输出。

（4）MapReduce编程模型只能包含一个Map阶段和一个Reduce阶段，如果用户的业务逻辑非常复杂，那就只能多个MapReduce程序，串行运行。

总结：分析WordCount数据流走向深入理解MapReduce核心思想。

### 1.4 MapReduce进程 ###

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

（1）**MrAppMaster**：负责整个程序的过程调度及状态协调。

（2）**MapTask**：负责Map阶段的整个数据处理流程。

（3）**ReduceTask**：负责Reduce阶段的整个数据处理流程。

### 1.5 官方WordCount源码 ###

采用反编译工具反编译源码，发现WordCount案例有Map类、Reduce类和驱动类。且数据的类型是Hadoop自身封装的序列化类型。

### 1.6 常用数据序列化类型 ###

<table style="margin-left:-.25pt;width:556px;"> 
 <tbody> 
 <tr> 
 <td style="border-color:#000000;vertical-align:top;width:212.9pt;"> Java类型 </td> 
 <td style="border-color:#000000;vertical-align:top;width:203.85pt;"> Hadoop Writable类型 </td> 
 </tr> 
 <tr> 
 <td style="border-color:#000000;vertical-align:top;width:212.9pt;"> Boolean </td> 
 <td style="vertical-align:top;width:203.85pt;"> BooleanWritable </td> 
 </tr> 
 <tr> 
 <td style="border-color:#000000;vertical-align:top;width:212.9pt;"> Byte </td> 
 <td style="vertical-align:top;width:203.85pt;"> ByteWritable </td> 
 </tr> 
 <tr> 
 <td style="border-color:#000000;vertical-align:top;width:212.9pt;"> Int </td> 
 <td style="vertical-align:top;width:203.85pt;"> IntWritable </td> 
 </tr> 
 <tr> 
 <td style="border-color:#000000;vertical-align:top;width:212.9pt;"> Float </td> 
 <td style="vertical-align:top;width:203.85pt;"> FloatWritable </td> 
 </tr> 
 <tr> 
 <td style="border-color:#000000;vertical-align:top;width:212.9pt;"> Long </td> 
 <td style="vertical-align:top;width:203.85pt;"> LongWritable </td> 
 </tr> 
 <tr> 
 <td style="border-color:#000000;vertical-align:top;width:212.9pt;"> Double </td> 
 <td style="vertical-align:top;width:203.85pt;"> DoubleWritable </td> 
 </tr> 
 <tr> 
 <td style="border-color:#000000;vertical-align:top;width:212.9pt;"> String </td> 
 <td style="vertical-align:top;width:203.85pt;"> Text </td> 
 </tr> 
 <tr> 
 <td style="border-color:#000000;vertical-align:top;width:212.9pt;"> Map </td> 
 <td style="vertical-align:top;width:203.85pt;"> MapWritable </td> 
 </tr> 
 <tr> 
 <td style="border-color:#000000;vertical-align:top;width:212.9pt;"> Array </td> 
 <td style="vertical-align:top;width:203.85pt;"> ArrayWritable </td> 
 </tr> 
 <tr> 
 <td style="border-color:#000000;vertical-align:top;width:212.9pt;"> Null </td> 
 <td style="vertical-align:top;width:203.85pt;"> NullWritable </td> 
 </tr> 
 </tbody> 
</table>

### 1.7 MapReduce编程规范 ###

用户编写的程序分成三个部分：Mapper、Reducer和Driver。

![8ee484396d894f43bc373eb3913cf4d0.png][]

![c858e300be5c40acbdbdcc605df8eb73.png][]

### 1.8 WordCount案例实操 ###

#### 1.8.1 本地测试 ####

**1****）需求**

在给定的文本文件中统计输出每一个单词出现的总次数。

（1）输入数据

![b3f55f17be134035ba1128235c1154a5.png][]

（2）期望输出数据

atguigu 2

banzhang 1

cls 2

hadoop 1

jiao 1

ss 2

xue 1

**2****）需求分析**

按照MapReduce编程规范，分别编写Mapper，Reducer，Driver。

![ba758ad73524459ab4e5b3a3fd1b0991.png][]

**3****）环境准备**

（1）创建maven工程，MapReduceDemo

（2）在pom.xml文件中添加如下依赖

<dependencies>
 <dependency>
 <groupId>org.apache.hadoop</groupId>
 <artifactId>hadoop-client</artifactId>
 <version>3.1.3</version>
 </dependency>
 <dependency>
 <groupId>junit</groupId>
 <artifactId>junit</artifactId>
 <version>4.12</version>
 </dependency>
 <dependency>
 <groupId>org.slf4j</groupId>
 <artifactId>slf4j-log4j12</artifactId>
 <version>1.7.30</version>
 </dependency>
 </dependencies>

（2）在项目的src/main/resources目录下，新建一个文件，命名为“log4j.properties”，在文件中填入。

log4j.rootLogger=INFO, stdout  
    log4j.appender.stdout=org.apache.log4j.ConsoleAppender  
    log4j.appender.stdout.layout=org.apache.log4j.PatternLayout  
    log4j.appender.stdout.layout.ConversionPattern=%d %p [%c] - %m%n  
    log4j.appender.logfile=org.apache.log4j.FileAppender  
    log4j.appender.logfile.File=target/spring.log  
    log4j.appender.logfile.layout=org.apache.log4j.PatternLayout  
    log4j.appender.logfile.layout.ConversionPattern=%d %p [%c] - %m%n

（3）创建包名：com.atguigu.mapreduce.wordcount

**4****）编写程序**

（1）编写Mapper类

package com.atguigu.mapreduce.wordcount;
 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;
 
 public class WordCountMapper extends Mapper<LongWritable, Text, Text, IntWritable>{
 	
 	Text k = new Text();
 	IntWritable v = new IntWritable(1);
 	
 	@Override
 	protected void map(LongWritable key, Text value, Context context)	throws IOException, InterruptedException {
 		
 		// 1 获取一行
 		String line = value.toString();
 		
 		// 2 切割
 		String[] words = line.split(" ");
 		
 		// 3 输出
 		for (String word : words) {
 			
 			k.set(word);
 			context.write(k, v);
 		}
 	}
 }

（2）编写Reducer类

package com.atguigu.mapreduce.wordcount;
 import java.io.IOException;
 import org.apache.hadoop.io.IntWritable;
 import org.apache.hadoop.io.Text;
 import org.apache.hadoop.mapreduce.Reducer;
 
 public class WordCountReducer extends Reducer<Text, IntWritable, Text, IntWritable>{
 
 int sum;
 IntWritable v = new IntWritable();
 
 	@Override
 	protected void reduce(Text key, Iterable<IntWritable> values,Context context) throws IOException, InterruptedException {
 		
 		// 1 累加求和
 		sum = 0;
 		for (IntWritable count : values) {
 			sum += count.get();
 		}
 		
 		// 2 输出
 v.set(sum);
 		context.write(key,v);
 	}
 }

（3）编写Driver驱动类

package com.atguigu.mapreduce.wordcount;
    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 WordCountDriver {
    
    	public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException {
    
    		// 1 获取配置信息以及获取job对象
    		Configuration conf = new Configuration();
    		Job job = Job.getInstance(conf);
    
    		// 2 关联本Driver程序的jar
    		job.setJarByClass(WordCountDriver.class);
    
    		// 3 关联Mapper和Reducer的jar
    		job.setMapperClass(WordCountMapper.class);
    		job.setReducerClass(WordCountReducer.class);
    
    		// 4 设置Mapper输出的kv类型
    		job.setMapOutputKeyClass(Text.class);
    		job.setMapOutputValueClass(IntWritable.class);
    
    		// 5 设置最终输出kv类型
    		job.setOutputKeyClass(Text.class);
    		job.setOutputValueClass(IntWritable.class);
    		
    		// 6 设置输入和输出路径
    		FileInputFormat.setInputPaths(job, new Path(args[0]));
    		FileOutputFormat.setOutputPath(job, new Path(args[1]));
    
    		// 7 提交job
    		boolean result = job.waitForCompletion(true);
    		System.exit(result ? 0 : 1);
    	}
    }

**5****）本地测试**

（1）需要首先配置好HADOOP\_HOME变量以及Windows运行依赖

（2）在IDEA/Eclipse上运行程序

#### 1.8.2 提交到集群测试 ####

**1****）集群上测试**

（1）用maven打jar包，需要添加的打包插件依赖

<build>
 <plugins>
 <plugin>
 <artifactId>maven-compiler-plugin</artifactId>
 <version>3.6.1</version>
 <configuration>
 <source>1.8</source>
 <target>1.8</target>
 </configuration>
 </plugin>
 <plugin>
 <artifactId>maven-assembly-plugin</artifactId>
 <configuration>
 <descriptorRefs>
 <descriptorRef>jar-with-dependencies</descriptorRef>
 </descriptorRefs>
 </configuration>
 <executions>
 <execution>
 <id>make-assembly</id>
 <phase>package</phase>
 <goals>
 <goal>single</goal>
 </goals>
 </execution>
 </executions>
 </plugin>
 </plugins>
 </build>

注意：如果工程上显示红叉。在项目上右键->maven->Reimport刷新即可。

（2）将程序打成jar包

![ce284b833cf549778488119c0a273732.png][]

（3）修改不带依赖的jar包名称为wc.jar，并拷贝该jar包到Hadoop集群的/opt/module/hadoop-3.1.3路径。

（4）启动Hadoop集群

[wyr@hadoop102 hadoop-3.1.3]sbin/start-dfs.sh
    [wyr@hadoop103 hadoop-3.1.3]$ sbin/start-yarn.sh

（5）执行WordCount程序

[wyr@hadoop102 hadoop-3.1.3]$ hadoop jar  wc.jar
     com.wyr.mapreduce.wordcount.WordCountDriver /user/wyr/input /user/wyr/output

#### 1.8.3在Windows上向集群提交任务 ####

（1）添加必要配置信息

public class WordcountDriver {
    
    	public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException {
    
    		// 1 获取配置信息以及封装任务
    		Configuration conf = new Configuration();
    
           //设置在集群运行的相关参数-设置HDFS,NAMENODE的地址
            conf.set("fs.defaultFS", "hdfs://hadoop102:8020");
            //指定MR运行在Yarn上
            conf.set("mapreduce.framework.name","yarn");
            //指定MR可以在远程集群运行
            conf.set("mapreduce.app-submission.cross-platform",
    "true");
            //指定yarn resourcemanager的位置
            conf.set("yarn.resourcemanager.hostname",
    "hadoop103")
    		Job job = Job.getInstance(configuration);
    		// 2 设置jar加载路径
    		job.setJarByClass(WordcountDriver.class);
    
    		// 3 设置map和reduce类
    		job.setMapperClass(WordcountMapper.class);
    		job.setReducerClass(WordcountReducer.class);
    
    		// 4 设置map输出
    		job.setMapOutputKeyClass(Text.class);
    		job.setMapOutputValueClass(IntWritable.class);
    
    		// 5 设置最终输出kv类型
    		job.setOutputKeyClass(Text.class);
    		job.setOutputValueClass(IntWritable.class);
    		
    		// 6 设置输入和输出路径
    		FileInputFormat.setInputPaths(job, new Path(args[0]));
    		FileOutputFormat.setOutputPath(job, new Path(args[1]));
    
    		// 7 提交
    		boolean result = job.waitForCompletion(true);
    
    		System.exit(result ? 0 : 1);
    	}
    }

（2）编辑任务配置

![021b3397123840a0b128cf29f76aecda.png][]（3） 打包

（4） 将Jar包设置到Driver中

public class WordcountDriver {
    
    	public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException {
    
    		// 1 获取配置信息以及封装任务
    		Configuration configuration = new Configuration();
    
           configuration.set("fs.defaultFS", "hdfs://hadoop102:8020");
           configuration.set("mapreduce.framework.name","yarn");
           configuration.set("mapreduce.app-submission.cross-platform","true");
        configuration.set("yarn.resourcemanager.hostname","hadoop103");
    
    		Job job = Job.getInstance(configuration);
    
    		// 2 设置jar加载路径
    		job.setJar("C:\\Users\\skiin\\IdeaProjects\\mapreduce1021\\target\\mapreduce1021-1.0-SNAPSHOT.jar");
    
    		// 3 设置map和reduce类
    		job.setMapperClass(WordcountMapper.class);
    		job.setReducerClass(WordcountReducer.class);
    
    		// 4 设置map输出
    		job.setMapOutputKeyClass(Text.class);
    		job.setMapOutputValueClass(IntWritable.class);
    
    		// 5 设置最终输出kv类型
    		job.setOutputKeyClass(Text.class);
    		job.setOutputValueClass(IntWritable.class);
    		
    		// 6 设置输入和输出路径
    		FileInputFormat.setInputPaths(job, new Path(args[0]));
    		FileOutputFormat.setOutputPath(job, new Path(args[1]));
    
    		// 7 提交
    		boolean result = job.waitForCompletion(true);
    
    		System.exit(result ? 0 : 1);
    	}
    }

（4）提交并查看结果

## 第2章 Hadoop序列化 ##

### 2.1 序列化概述 ###

**1****）什么是序列化**

**序列化**就是把内存中的对象，转换成字节序列（或其他数据传输协议）以便于存储到磁盘（持久化）和网络传输。

**反序列化**就是将收到字节序列（或其他数据传输协议）或者是磁盘的持久化数据，转换成内存中的对象。

**2****）为什么要序列化**

一般来说，“活的”对象只生存在内存里，关机断电就没有了。而且“活的”对象只能由本地的进程使用，不能被发送到网络上的另外一台计算机。 然而序列化可以存储“活的”对象，可以将“活的”对象发送到远程计算机。

**3****）为什么不用Java****的序列化**

Java的序列化是一个重量级序列化框架（Serializable），一个对象被序列化后，会附带很多额外的信息（各种校验信息，Header，继承体系等），不便于在网络中高效传输。所以，Hadoop自己开发了一套序列化机制（Writable）。

**4****）Hadoop****序列化特点**

*  **紧凑** **：**高效使用存储空间。
 *  **快速：**读写数据的额外开销小。
 *  **互操作：**支持多语言的交互

### 2.2 自定义bean对象实现序列化接口（Writable） ###

在企业开发中往往常用的基本序列化类型不能满足所有需求，比如在Hadoop框架内部传递一个bean对象，那么该对象就需要实现序列化接口。

具体实现bean对象序列化步骤如下7步。

（1）必须实现Writable接口

（2）反序列化时，需要反射调用空参构造函数，所以必须有空参构造

public FlowBean() {
    	super();
    }

（3）重写序列化方法

@Override
    public void write(DataOutput out) throws IOException {
    	out.writeLong(upFlow);
    	out.writeLong(downFlow);
    	out.writeLong(sumFlow);
    }

（4）重写反序列化方法

@Override
    public void readFields(DataInput in) throws IOException {
    	upFlow = in.readLong();
    	downFlow = in.readLong();
    	sumFlow = in.readLong();
    }

（5）注意反序列化的顺序和序列化的顺序完全一致。

（6）要想把结果显示在文件中，需要重写toString()，可用"\\t"分开，方便后续用。

（7）如果需要将自定义的bean放在key中传输，则还需要实现Comparable接口，因为MapReduce框中的Shuffle过程要求对key必须能排序。详见后面排序案例。

@Override
    public int compareTo(FlowBean o) {
    	// 倒序排列，从大到小
    	return this.sumFlow > o.getSumFlow() ? -1 : 1;
    }

### 2.3 序列化案例实操 ###

**1****）需求**

统计每一个手机号耗费的总上行流量、总下行流量、总流量

（1）输入数据

![63c9bc861c8d4bc5a83c9bdfa0a16a26.png][]

（2）输入数据格式：

<table style="margin-left:21.05pt;width:527px;"> 
 <tbody> 
 <tr> 
 <td style="border-color:#000000;vertical-align:top;width:395.45pt;"> 7 &nbsp;&nbsp;&nbsp;&nbsp; 13560436666&nbsp;&nbsp;&nbsp;&nbsp; 120.196.100.99&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; 1116&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; 954&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; 200 id&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; 手机号码&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; 网络ip&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; 上行流量 &nbsp;下行流量&nbsp;&nbsp;&nbsp;&nbsp; 网络状态码 </td> 
 </tr> 
 </tbody> 
</table>

（3）期望输出数据格式

**2****）需求分析**

![041f002161b142d98599e237c5647dbe.png][]

**3****）编写MapReduce****程序**

（1）编写流量统计的Bean对象

package com.wyr.mapreduce.writable;
    
    import org.apache.hadoop.io.Writable;
    import java.io.DataInput;
    import java.io.DataOutput;
    import java.io.IOException;
    
    //1 继承Writable接口
    public class FlowBean implements Writable {
    
        private long upFlow; //上行流量
        private long downFlow; //下行流量
        private long sumFlow; //总流量
    
        //2 提供无参构造
        public FlowBean() {
        }
    
        //3 提供三个参数的getter和setter方法
        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;
        }
    
        public void setSumFlow() {
            this.sumFlow = this.upFlow + this.downFlow;
        }
    
        //4 实现序列化和反序列化方法,注意顺序一定要保持一致
        @Override
        public void write(DataOutput dataOutput) throws IOException {
            dataOutput.writeLong(upFlow);
            dataOutput.writeLong(downFlow);
            dataOutput.writeLong(sumFlow);
        }
    
        @Override
        public void readFields(DataInput dataInput) throws IOException {
            this.upFlow = dataInput.readLong();
            this.downFlow = dataInput.readLong();
            this.sumFlow = dataInput.readLong();
        }
    
        //5 重写ToString
        @Override
        public String toString() {
            return upFlow + "\t" + downFlow + "\t" + sumFlow;
        }
    }

（2）编写Mapper类

package com.wyr.mapreduce.writable;
 
 import org.apache.hadoop.io.LongWritable;
 import org.apache.hadoop.io.Text;
 import org.apache.hadoop.mapreduce.Mapper;
 import java.io.IOException;
 
 public class FlowMapper extends Mapper<LongWritable, Text, Text, FlowBean> {
 private Text outK = new Text();
 private FlowBean outV = new FlowBean();
 
 @Override
 protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
 
 //1 获取一行数据,转成字符串
 String line = value.toString();
 
 //2 切割数据
 String[] split = line.split("\t");
 
 //3 抓取我们需要的数据:手机号,上行流量,下行流量
 String phone = split[1];
 String up = split[split.length - 3];
 String down = split[split.length - 2];
 
 //4 封装outK outV
 outK.set(phone);
 outV.setUpFlow(Long.parseLong(up));
 outV.setDownFlow(Long.parseLong(down));
 outV.setSumFlow();
 
 //5 写出outK outV
 context.write(outK, outV);
 }
 }

（3）编写Reducer类

package com.wyr.mapreduce.writable;
 
 import org.apache.hadoop.io.Text;
 import org.apache.hadoop.mapreduce.Reducer;
 import java.io.IOException;
 
 public class FlowReducer extends Reducer<Text, FlowBean, Text, FlowBean> {
 private FlowBean outV = new FlowBean();
 @Override
 protected void reduce(Text key, Iterable<FlowBean> values, Context context) throws IOException, InterruptedException {
 
 long totalUp = 0;
 long totalDown = 0;
 
 //1 遍历values,将其中的上行流量,下行流量分别累加
 for (FlowBean flowBean : values) {
 totalUp += flowBean.getUpFlow();
 totalDown += flowBean.getDownFlow();
 }
 
 //2 封装outKV
 outV.setUpFlow(totalUp);
 outV.setDownFlow(totalDown);
 outV.setSumFlow();
 
 //3 写出outK outV
 context.write(key,outV);
 }
 }

（4）编写Driver驱动类

package com.wyr.mapreduce.writable;
    
    import org.apache.hadoop.conf.Configuration;
    import org.apache.hadoop.fs.Path;
    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;
    import java.io.IOException;
    
    public class FlowDriver {
        public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException {
    
            //1 获取job对象
            Configuration conf = new Configuration();
            Job job = Job.getInstance(conf);
    
            //2 关联本Driver类
            job.setJarByClass(FlowDriver.class);
    
            //3 关联Mapper和Reducer
            job.setMapperClass(FlowMapper.class);
            job.setReducerClass(FlowReducer.class);
            
    //4 设置Map端输出KV类型
            job.setMapOutputKeyClass(Text.class);
            job.setMapOutputValueClass(FlowBean.class);
            
    //5 设置程序最终输出的KV类型
            job.setOutputKeyClass(Text.class);
            job.setOutputValueClass(FlowBean.class);
            
    //6 设置程序的输入输出路径
            FileInputFormat.setInputPaths(job, new Path("D:\\inputflow"));
            FileOutputFormat.setOutputPath(job, new Path("D:\\flowoutput"));
            
    //7 提交Job
            boolean b = job.waitForCompletion(true);
            System.exit(b ? 0 : 1);
        }
    }

## 第3章 MapReduce框架原理 ##

![e2366bcb1336405a983e45aa26507c29.png][]

### **3.1 InputFormat数据输入** ###

#### **3.1.1 切片与MapTask并行度决定机制** ####

**1****）问题引出**

MapTask的并行度决定Map阶段的任务处理并发度，进而影响到整个Job的处理速度。

思考：1G的数据，启动8个MapTask，可以提高集群的并发处理能力。那么1K的数据，也启动8个MapTask，会提高集群性能吗？MapTask并行任务是否越多越好呢？哪些因素影响了MapTask并行度？

**2****）MapTask****并行度决定机制**

**数据块：**Block是HDFS物理上把数据分成一块一块。数据块是HDFS存储数据单位。

**数据切片：**数据切片只是在逻辑上对输入进行分片，并不会在磁盘上将其切分成片进行存储。数据切片是MapReduce程序计算输入数据的单位，一个切片会对应启动一个MapTask。

![10b8e0672b904f4b8c4f0d10f5a1774b.png][]

#### **3.1.2 Job提交流程源码和切片源码详解** ####

**1****）Job****提交流程源码详解**

waitForCompletion()
    
    submit();
    
    // 1建立连接
    	connect();	
    		// 1）创建提交Job的代理
    		new Cluster(getConfiguration());
    			// （1）判断是本地运行环境还是yarn集群运行环境
    			initialize(jobTrackAddr, conf); 
    
    // 2 提交job
    submitter.submitJobInternal(Job.this, cluster)
    
    	// 1）创建给集群提交数据的Stag路径
    	Path jobStagingArea = JobSubmissionFiles.getStagingDir(cluster, conf);
    
    	// 2）获取jobid ，并创建Job路径
    	JobID jobId = submitClient.getNewJobID();
    
    	// 3）拷贝jar包到集群
    copyAndConfigureFiles(job, submitJobDir);	
    	rUploader.uploadFiles(job, jobSubmitDir);
    
    	// 4）计算切片，生成切片规划文件
    writeSplits(job, submitJobDir);
    		maps = writeNewSplits(job, jobSubmitDir);
    		input.getSplits(job);
    
    	// 5）向Stag路径写XML配置文件
    writeConf(conf, submitJobFile);
    	conf.writeXml(out);
    
    	// 6）提交Job,返回提交状态
    status = submitClient.submitJob(jobId, submitJobDir.toString(), job.getCredentials());

![84231973d2934bb6a234c005023beff9.png][]

**2****）FileInputFormat****切片源码解析（input.getSplits(job)****）**

![5462fb11aa4c41b0817b1a635edc2e20.png][]

#### **3.1.3 FileInputFormat切片机制** ####

![11ff211b2f8741e285dea126bb0aaf61.png][]

![6ad4f5f3c8cf4fe8a08eef8e483986b7.png][]

#### **3.1.4 TextInputFormat** ####

**1****）FileInputFormat****实现类**

思考：在运行MapReduce程序时，输入的文件格式包括：基于行的日志文件、二进制格式文件、数据库表等。那么，针对不同的数据类型，MapReduce是如何读取这些数据的呢？

FileInputFormat常见的接口实现类包括：TextInputFormat、KeyValueTextInputFormat、NLineInputFormat、CombineTextInputFormat和自定义InputFormat等。

**2****）TextInputFormat**

TextInputFormat是默认的FileInputFormat实现类。按行读取每条记录。key是存储该行在整个文件中的起始字节偏移量， LongWritable类型。value是这行的内容，不包括任何行终止符（换行符和回车符），Text类型。

以下是一个示例，比如，一个分片包含了如下4条文本记录。

Rich learning form
    Intelligent learning engine
    Learning more convenient
    From the real demand for more close to the enterprise

每条记录表示为以下键/值对：

(0,Rich learning form)
    (20,Intelligent learning engine)
    (49,Learning more convenient)
    (74,From the real demand for more close to the enterprise)

#### **3.1.6 CombineTextInputFormat案例实操** ####

**1****）需求**

将输入的大量小文件合并成一个切片统一处理。

（1）输入数据

准备4个小文件

![e0200796ec60416dac27fc5572d31ae1.png][]

（2）期望

期望一个切片处理4个文件

**2****）实现过程**

（1）不做任何处理，运行1.8节的WordCount案例程序，观察切片个数为4。

number of splits:4

（2）在WordcountDriver中增加如下代码，运行程序，并观察运行的切片个数为3。

①驱动类中添加代码如下：

// 如果不设置InputFormat，它默认用的是TextInputFormat.class
    job.setInputFormatClass(CombineTextInputFormat.class);
    
    //虚拟存储切片最大值设置4m
    CombineTextInputFormat.setMaxInputSplitSize(job, 4194304);

②运行如果为3个切片。

number of splits:3

（3）在WordcountDriver中增加如下代码，运行程序，并观察运行的切片个数为1。

①驱动中添加代码如下：

// 如果不设置InputFormat，它默认用的是TextInputFormat.class
    job.setInputFormatClass(CombineTextInputFormat.class);
    
    //虚拟存储切片最大值设置20m
    CombineTextInputFormat.setMaxInputSplitSize(job, 20971520);

②运行如果为1个切片。

number of splits:1

### **3.2 MapReduce工作流程** ###

![fec440c7ab6d4c8d931c8233e5809896.png][]

![ce2a3fd57b2f4095ac446d990c0e5f62.png][]

上面的流程是整个MapReduce最全工作流程，但是Shuffle过程只是从第7步开始到第16步结束，具体Shuffle过程详解，如下：

（1）MapTask收集我们的map()方法输出的kv对，放到内存缓冲区中

（2）从内存缓冲区不断溢出本地磁盘文件，可能会溢出多个文件

（3）多个溢出文件会被合并成大的溢出文件

（4）在溢出过程及合并的过程中，都要调用Partitioner进行分区和针对key进行排序

（5）ReduceTask根据自己的分区号，去各个MapTask机器上取相应的结果分区数据

（6）ReduceTask会抓取到同一个分区的来自不同MapTask的结果文件，ReduceTask会将这些文件再进行合并（归并排序）

（7）合并成大文件后，Shuffle的过程也就结束了，后面进入ReduceTask的逻辑运算过程（从文件中取出一个一个的键值对Group，调用用户自定义的reduce()方法）

**注意：**

（1）Shuffle中的缓冲区大小会影响到MapReduce程序的执行效率，原则上说，缓冲区越大，磁盘io的次数越少，执行速度就越快。

（2）缓冲区的大小可以通过参数调整，参数：mapreduce.task.io.sort.mb默认100M。

### **3.3 Shuffle机制** ###

#### **3.3.1 Shuffle机制** ####

Map方法之后，Reduce方法之前的数据处理过程称之为Shuffle。

![5eb2be9202684efc8438c1726f8e0939.png][]

#### **3.3.2 Partition分区** ####

#### **3.3.3 Partition分区案例实操** ####

**1****）需求**

将统计结果按照手机归属地不同省份输出到不同文件中（分区）

（1）输入数据

（2）期望输出数据

手机号136、137、138、139开头都分别放到一个独立的4个文件中，其他开头的放到一个文件中。

**2****）需求分析**

**3****）在案例2.3****的基础上，增加一个分区类**

package com.wyr.mapreduce.partitioner;

import org.apache.hadoop.io.Text;

import org.apache.hadoop.mapreduce.Partitioner;

public class ProvincePartitioner extends Partitioner<Text, FlowBean> \{

@Override

public int getPartition(Text text, FlowBean flowBean, int numPartitions) \{

//获取手机号前三位prePhone

String phone = text.toString();

String prePhone = phone.substring(0, 3);

//定义一个分区号变量partition,根据prePhone设置分区号

int partition;

if("136".equals(prePhone))\{

partition = 0;

\}else if("137".equals(prePhone))\{

partition = 1;

\}else if("138".equals(prePhone))\{

partition = 2;

[da4f273ff0b54fb8b79c6c0cbec5d630.png]: https://img-blog.csdnimg.cn/da4f273ff0b54fb8b79c6c0cbec5d630.png
[1dfcfbc034c145c4bc285b39ef47e47d.png]: https://img-blog.csdnimg.cn/1dfcfbc034c145c4bc285b39ef47e47d.png
[8ee484396d894f43bc373eb3913cf4d0.png]: https://img-blog.csdnimg.cn/8ee484396d894f43bc373eb3913cf4d0.png
[c858e300be5c40acbdbdcc605df8eb73.png]: https://img-blog.csdnimg.cn/c858e300be5c40acbdbdcc605df8eb73.png
[b3f55f17be134035ba1128235c1154a5.png]: https://img-blog.csdnimg.cn/b3f55f17be134035ba1128235c1154a5.png
[ba758ad73524459ab4e5b3a3fd1b0991.png]: https://img-blog.csdnimg.cn/ba758ad73524459ab4e5b3a3fd1b0991.png
[ce284b833cf549778488119c0a273732.png]: https://img-blog.csdnimg.cn/ce284b833cf549778488119c0a273732.png
[021b3397123840a0b128cf29f76aecda.png]: https://img-blog.csdnimg.cn/021b3397123840a0b128cf29f76aecda.png
[63c9bc861c8d4bc5a83c9bdfa0a16a26.png]: https://img-blog.csdnimg.cn/63c9bc861c8d4bc5a83c9bdfa0a16a26.png
[041f002161b142d98599e237c5647dbe.png]: https://img-blog.csdnimg.cn/041f002161b142d98599e237c5647dbe.png
[e2366bcb1336405a983e45aa26507c29.png]: https://img-blog.csdnimg.cn/e2366bcb1336405a983e45aa26507c29.png
[10b8e0672b904f4b8c4f0d10f5a1774b.png]: https://img-blog.csdnimg.cn/10b8e0672b904f4b8c4f0d10f5a1774b.png
[84231973d2934bb6a234c005023beff9.png]: https://img-blog.csdnimg.cn/84231973d2934bb6a234c005023beff9.png
[5462fb11aa4c41b0817b1a635edc2e20.png]: https://img-blog.csdnimg.cn/5462fb11aa4c41b0817b1a635edc2e20.png
[11ff211b2f8741e285dea126bb0aaf61.png]: https://img-blog.csdnimg.cn/11ff211b2f8741e285dea126bb0aaf61.png
[6ad4f5f3c8cf4fe8a08eef8e483986b7.png]: https://img-blog.csdnimg.cn/6ad4f5f3c8cf4fe8a08eef8e483986b7.png
[e0200796ec60416dac27fc5572d31ae1.png]: https://img-blog.csdnimg.cn/e0200796ec60416dac27fc5572d31ae1.png
[fec440c7ab6d4c8d931c8233e5809896.png]: https://img-blog.csdnimg.cn/fec440c7ab6d4c8d931c8233e5809896.png
[ce2a3fd57b2f4095ac446d990c0e5f62.png]: https://img-blog.csdnimg.cn/ce2a3fd57b2f4095ac446d990c0e5f62.png
[5eb2be9202684efc8438c1726f8e0939.png]: https://img-blog.csdnimg.cn/5eb2be9202684efc8438c1726f8e0939.png