Hadoop详解(三)——MapReduce原理和执行过程，远程Debug，Writable序列化接口，MapReduce程序编写

青旅半醒 2022-06-08 20:37 159阅读 0赞

# MapReduce概述 #

MapReduce是一种分布式计算模型，由Google提出，主要用于搜索领域，解决海量数据的计算问题。

MR由两个阶段组成：Map和Reduce，用户只需要实现map()和Reduce()两个函数，即可实现分布式计算。

这两个函数的形参是key、Value对，表示函数的输入信息。

MapReduce的老大是JobTracker 小弟叫TaskTracker相当于小组长 执行具体任务的是Map任务和reduce任务  
在Hadoop 0.23版本之后 JobTracker—>ResourceManager(RM)   TaskTracker—>NodeManager(NM)  
RM和NM只存在于Hadoop 2.0之后的版本中 JobTracker和TaskTracker只存在Hadoop1.0以下 除了0.23版本  
有了Yarn之后 完全转为RM和NM YARN不仅仅可以运行MapReduce

# MapReduce的大致流程 #

![Center][]

① Client将业务逻辑代码(计算逻辑)打成jar包 上传到Linux运行。

② client然后向RM发出提交作业的请求（使用RPC协议），RM根据任务请求 获取jar包的信息，然后向客户端发送元数据信息(在HDFS上的jar包位置即存放jar包的路径 和一个jobid) ,

③ 客户端Client拿到RM返回的信息 为避免重复，将信息里存放jar包的路径作为前缀，jobid作为后缀进行拼接作为唯一标识在HDFS上存放此jar包的路径，得到唯一路径后，将此jar包写入到HDFS中(Client中持有FileSystem对象的引用)。一般手动上传文件默认是保存3份(伪分布式配置为1份)，为了减轻计算压力 jar包(map和reducer)会在HSDF上保存10份，运行完成一段时间后就会删除。

④ 写入成功完成后，客户端会将jar包的存储位置，jobid 作为参数以RPC的方式传递给RM。

⑤、⑥ RM就会存储此作业的描述信息。然后RM就会使用任务调度器(默认的调度器是队列调度器(job queue) 以先进先出(FIFO)的方式执行)来执行任务。

⑦ 一旦作业被放入调度器内，NM就会通过心跳机制领取任务，由RM根据NM的机器状况决定是否将任务分配给NM。

⑧、⑨NM领取到任务后就会从HDFS上下载jar包，并启动独立于自己的子进程(MapTask从HDFS读取数据。

10、ReduceTask计算结果并写入到HDFS中)执行任务

注意：这里所谓的心跳机制就是每隔一段时间，NM会向RM回报信息并自动向RM询问申请任务，然后RM就会将元数据信息发送给NM，NM根据元数据信息从HDFS上现在对应的jar包数据并运行。

Hadoop1.0版本中任务执行的详细步骤如下：

![Center 1][]

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当中

# MapReduce原理 #

![Center 2][]

JobTracker 功能是资源分配 由它根据机器性能决定在哪台机器上运行几个Mapper和几个Reducer  
同时它还进行任务的监控，如果检测到某执行任务的机器宕机了 JobTracker会将该任务进行转移  
InputSplit 输入切片  一个InputSplit对应一个Mapper  
Mapper的任务执行完成后  Mapper的输出会作为Reducer的输入进行运算，Reducer任务完成后会将结果输出到HDFS中。

![Center 3][]

任务逻辑代码(jar) 上传到HDFS 被物理切分成多个Block  
通过记录文件偏移量将任务逻辑切分成多个任务切片(split)   
每个TaskSplit对应一个Mapper  
Mapper和Reducer和输入输出都是以<key,value>的形式存在的  
Shuffle (\*\*\*\*)非常重要 主要完成Mapper输出的排序和分组 进行合并  
框架已经完成了分区排序和分组，如果想实现MapReduce模型，只需重写Map方法和Reduce方法实现具体的业务逻辑即可。

# MapReduce相关的类简介： #

(一) FIleInputFormat：

FileInputFormat是所有以文件为数据源的InputFormat实现的基类，FileInputFormat保存作为job输入的所有文件，并实现了对输入文件计算splits的方法。至于获得记录的方法是有不同的子类——TextInputFormat进行实现的。

(二) InputFormat：

InputFormat负责处理MR的输入部分。

有三个作用：

① 验证作业的输入是否规范。

② 将输入文件切分成InputSplit。

③ 提供RecordReader的实现类，将InputSplit读到Mapper中进行处理。

(三) InputSplit

① 在执行MapReduce之前，原始数据被切割成若干个split，每个split作为一个map任务的输入，在map执行过程中split会被分解成一个个记录(key-value对)，map会依次处理每一个记录。

② FileInputFormat只划分比HDFS Block大的文件，所以FileInputFormat划分的结果就是这个文件或者是这个文件中的一部分。

③ 如果一个文件的大小比Block小，将不会被划分，这也就是Hadoop处理大文件的效率要比处理很多小文件的效率高的原因。

④ 当Hadoop处理很多小文件(文件大小小于HDFS Block大小)的时候，由于FileInputFormat不会对小文件进行划分，所以每一个小文件都会被当做一个split并分配一个map任务，导致效率低下。

例如：一个G的文件，会被划分成8个128MB的split，并分配8个map任务处理，而10000个100KB的文件会被10000个map任务处理。

(四) TextInputFormat

TextInputFormat是默认的处理类，处理普通文本文件。

文件每一行作为一个记录，它将每一行在文件中的起始偏移量作为key，每一行的内容作为Value。

默认以\\n或者回车键作为一行记录

TextInputFormat继承了FileInputFormat

(五)其它输入类

◆ CombineFileInputFormat  
相对于大量的小文件来说，hadoop更合适处理少量的大文件。  
CombineFileInputFormat可以缓解这个问题，它是针对小文件而设计的。  
◆ KeyValueTextInputFormat  
当输入数据的每一行是两列，并用tab分离的形式的时候，KeyValueTextInputformat处理这种格式的文件非常适合。  
◆ NLineInputformat NLineInputformat可以控制在每个split中数据的行数。  
◆ SequenceFileInputformat    
当输入文件格式是sequencefile的时候，要使用SequenceFileInputformat作为输入。  
(六) 自定义输入格式

1）继承FileInputFormat基类。  
2）重写里面的getSplits(JobContext context)方法。  
3）重写createRecordReader(InputSplit split,TaskAttemptContext context)方法。  
(七) Hadoop的输出

◆ TextOutputformat   
默认的输出格式，key和value中间值用tab隔开的。   
◆ SequenceFileOutputformat   
将key和value以sequencefile格式输出。   
◆ SequenceFileAsOutputFormat   
将key和value以原始二进制的格式输出。   
◆ MapFileOutputFormat   
将key和value写入MapFile中。由于MapFile中的key是有序的，所以写入的时候必须保证记录是按key值顺序写入的。   
◆ MultipleOutputFormat   
默认情况下一个reducer会产生一个输出，但是有些时候我们想一个reducer产生多个输出，MultipleOutputFormat和MultipleOutputs可以实现这个功能。

# Hadoop序列化 #

## 序列化概念 ##

序列化(Serialization) 是指将结构化对象转化为字节流的过程。

反序列化(Deserization) 是指将字节流转化为结构化对象的过程。

Java序列化(java.io.Serializable)

## Hadoop序列化特点 ##

Hadoop序列化特点

① 紧凑：高效使用存储空间。

② 快速：读取数据时的额外开销小

③ 可扩展：可透明地读取老格式的数据

④ 互操作：支持多语言的交互

Hadoop的序列化格式：Writable

Java序列化的不足：

① 不精简。附加信息多。不大适合随机访问。  
② 存储空间大。递归地输出类的超类描述直到不再有超类。序列化图对象，反序列化时为每个对象新建一个实例。相反。Writable对象可以重用。  
③ 扩展性差。而Writable方便用户自定义

总结： 因为JDK自带的序列化接口序列化对象时需要保存其继承关系，会导致网络传输非常慢  
Hadoop有两套序列化机制：一是自己的序列化机制，二是Google提供的Protobuf(protocol buffer)  
Hadoop 框架中标已经使用自己的序列化机制对一些类型进行了包装  
String —>Text   Long—>LongWritable

## Hadoop序列化的作用 ##

① 序列化在分布式环境的两大作用：进程间通讯，永久存储。

② Hadoop节点间通讯，如下图所示：

![Center 4][]

## Writable接口 ##

① Writable接口，是根据 DataInput和 DataOutput实现的简单的、有效的序列化对象。

②  MR任意的key和Value必须实现Writable,官方源码如下：

package org.apache.hadoop.io;
    import java.io.DataOutput;
    import java.io.DataInput;
    import java.io.IOException;
    import org.apache.hadoop.classification.InterfaceAudience;
    import org.apache.hadoop.classification.InterfaceStability;
    
    @InterfaceAudience.Public
    @InterfaceStability.Stable
    public interface Writable {
       void write(DataOutput out) throws IOException;
       void readFields(DataInput in) throws IOException;
    }

③ MR任意的key必须实现WritableComparable接口

官方源码如下：

package org.apache.hadoop.io;
    import org.apache.hadoop.classification.InterfaceAudience;
    import org.apache.hadoop.classification.InterfaceStability;
    @InterfaceAudience.Public
    @InterfaceStability.Stable
    public interface WritableComparable<T> extends Writable, Comparable<T> {
    }

# WordCount测试和编写 #

## 测试Hadoop自带的wordCount程序 ##

测试运行MapReduce  
测试步骤1：首先启动HDFS 再启动Yarn  
![Center 5][]  
测试步骤2：新建一个words文档，内容如下：  
![Center 6][]  
测试步骤3：将words上传到HDFS  
![Center 7][]  
测试步骤4：进入到/cloud/hadoop-2.7.4/share/hadoop/mapreduce 使用示例jar进行测试  
\[root@hadoop1 mapreduce\]\# hadoop jar hadoop-mapreduce-examples-2.7.4.jar wordcount /word.txt /wcount  其中/word.txt 是输入内容 /wcount是输出内容  
![Center 8][]  
测试步骤5：查看输出内容

![Center 9][]

## 分析Wordcount执行过程 ##

![Center 10][]

## 根据执行过程编写WordCount ##

① 编写Mapper类WCMapper

package liuxun.test.hadoop.mr;
    
    import java.io.IOException;
    
    import org.apache.hadoop.io.LongWritable;
    import org.apache.hadoop.io.Text;
    import org.apache.hadoop.mapreduce.Mapper;
    
    public class WCMapper extends Mapper<LongWritable, Text, Text, LongWritable> {
    
    	@Override
    	protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
    		// 接收数据
    		String line = value.toString();
    		// 切分数据
    		String[] words = line.split(" ");
    		// 循环
    		for (String w : words) {
    			// 出现一次，记一个1
    			context.write(new Text(w), new LongWritable(1));
    		}
    		
    	}
    
    }

② 编写Reducer类WCReducer

package liuxun.test.hadoop.mr;
    
    import java.io.IOException;
    
    import org.apache.hadoop.io.LongWritable;
    import org.apache.hadoop.io.Text;
    import org.apache.hadoop.mapreduce.Reducer;
    
    public class WCReducer extends Reducer<Text, LongWritable, Text, LongWritable> {
    
    	@Override
    	protected void reduce(Text k2, Iterable<LongWritable> v2s, Context context)
    			throws IOException, InterruptedException {
    		// 接收数据
    		Text k3 = k2;  // 这里仍将接收的key作为输出的key
    		// 定义一个计数器
    		long counter = 0;
    		// 循环迭代v2s
    		for (LongWritable i : v2s) {
    			counter+= i.get();
    		}
    		// 输出
    		context.write(k3, new LongWritable(counter));
    	}
    }

③ 编写主程序组装Mapper和Reducer

package liuxun.test.hadoop.mr;
    
    import java.io.IOException;
    
    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;
    /**
     * 
     * @author liuxun
     * 1.分析具体的业务逻辑，确定输入和输出数据的样式
     * 2.自定义一个类继承自org.apache.hadoop.mapreduce.Mapper类，重写map方法，实现具体业务逻辑，将新的key-Value输出
     * 3.自定义一个类继承自org.apache.hadoop.mapreduce.Reducer类，重写reduce方法，实现具体业务逻辑，将新的key-Value输出
     * 4.将自定义的mapper和reducer通过job对象组装起来
     */
    public class WordCount {
    
    	public static void main(String[] args) throws Exception {
    		// 构建一个Job对象
    		Job job = Job.getInstance(new Configuration());
    		
    		// 注意：一定要将main方法所在的类设置进去
    		job.setJarByClass(WordCount.class);
    		
    		// 设置Mapper相关属性
    		job.setMapperClass(WCMapper.class);
    		job.setMapOutputKeyClass(Text.class); //设置Map的输出参数key的类型
    		job.setMapOutputValueClass(LongWritable.class); //设置Map输出参数Value的类型
    		FileInputFormat.setInputPaths(job, new Path(args[0])); //设置输入的路径
    		
    		// 设置Reducer相关属性
    		job.setReducerClass(WCReducer.class);
    		job.setOutputKeyClass(Text.class);
    		job.setOutputValueClass(LongWritable.class);
    		FileOutputFormat.setOutputPath(job, new Path(args[1]));
    		
    		// 提交任务
    		// 参数 表示在执行任务的过程中是否打印进程信息
    		job.waitForCompletion(true);
    	}
    }

程序打包并指定主方法，然后将jar包上传至Linux主机，使用命令进行测试

hadoop jar /root/Desktop/WC.jar /words /wcResult

查看运行结果

\[root@hadoop0 ~\]\# hadoop fs -cat  /wcResult/part-r-00000  
hello   5  
jetty   1  
kitty   1  
tom     2  
world   1  
注意事项：

① 在编写job进行组装时FileInputFormat需要使用新的接口

org.apache.hadoop.mapred 是过时的老接口  
org.apache.hadoop.mapreduce.lib.input  是最新的接口

② 在打包时需要注意：

![Center 11][]

如果导出的是Runnable JAR file  运行时才可以使用java -jar <jar\_name>  
如果导出的是JAR file  运行时hadoop  jar  <jar\_name>  
在涉及到Hadoop中Mapper和Reducer自己的逻辑代码进行打包时 选择JAR file

③ 打包时尽量指定Main函数所在的类

如果不指定，那么在运行时需要指定Main方法所在类的全路径，格式如下：

hadoop jar <jar在linux的路径> <main方法所在的类的全类名> <参数>

# Maven开发自定义Bean实现MR #

日志内容如下：

HTTP\_20130313143750.dat

1363157985066 	13726230503	00-FD-07-A4-72-B8:CMCC	120.196.100.82	i02.c.aliimg.com		24	27	2481	24681	200
    1363157995052 	13826544101	5C-0E-8B-C7-F1-E0:CMCC	120.197.40.4			4	0	264	0	200
    1363157991076 	13926435656	20-10-7A-28-CC-0A:CMCC	120.196.100.99			2	4	132	1512	200
    1363154400022 	13926251106	5C-0E-8B-8B-B1-50:CMCC	120.197.40.4			4	0	240	0	200
    1363157993044 	18211575961	94-71-AC-CD-E6-18:CMCC-EASY	120.196.100.99	iface.qiyi.com	视频网站	15	12	1527	2106	200
    1363157995074 	84138413	5C-0E-8B-8C-E8-20:7DaysInn	120.197.40.4	122.72.52.12		20	16	4116	1432	200
    1363157993055 	13560439658	C4-17-FE-BA-DE-D9:CMCC	120.196.100.99			18	15	1116	954	200
    1363157995033 	15920133257	5C-0E-8B-C7-BA-20:CMCC	120.197.40.4	sug.so.360.cn	信息安全	20	20	3156	2936	200
    1363157983019	13719199419	68-A1-B7-03-07-B1:CMCC-EASY	120.196.100.82			4	0	240	0	200
    1363157984041 	13660577991	5C-0E-8B-92-5C-20:CMCC-EASY	120.197.40.4	s19.cnzz.com	站点统计	24	9	6960	690	200
    1363157973098 	15013685858	5C-0E-8B-C7-F7-90:CMCC	120.197.40.4	rank.ie.sogou.com	搜索引擎	28	27	3659	3538	200
    1363157986029 	15989002119	E8-99-C4-4E-93-E0:CMCC-EASY	120.196.100.99	www.umeng.com	站点统计	3	3	1938	180	200
    1363157992093 	13560439658	C4-17-FE-BA-DE-D9:CMCC	120.196.100.99			15	9	918	4938	200
    1363157986041 	13480253104	5C-0E-8B-C7-FC-80:CMCC-EASY	120.197.40.4			3	3	180	180	200
    1363157984040 	13602846565	5C-0E-8B-8B-B6-00:CMCC	120.197.40.4	2052.flash2-http.qq.com	综合门户	15	12	1938	2910	200
    1363157995093 	13922314466	00-FD-07-A2-EC-BA:CMCC	120.196.100.82	img.qfc.cn		12	12	3008	3720	200
    1363157982040 	13502468823	5C-0A-5B-6A-0B-D4:CMCC-EASY	120.196.100.99	y0.ifengimg.com	综合门户	57	102	7335	110349	200
    1363157986072 	18320173382	84-25-DB-4F-10-1A:CMCC-EASY	120.196.100.99	input.shouji.sogou.com	搜索引擎	21	18	9531	2412	200
    1363157990043 	13925057413	00-1F-64-E1-E6-9A:CMCC	120.196.100.55	t3.baidu.com	搜索引擎	69	63	11058	48243	200
    1363157988072 	13760778710	00-FD-07-A4-7B-08:CMCC	120.196.100.82			2	2	120	120	200
    1363157985066 	13726238888	00-FD-07-A4-72-B8:CMCC	120.196.100.82	i02.c.aliimg.com		24	27	2481	24681	200
    1363157993055 	13560436666	C4-17-FE-BA-DE-D9:CMCC	120.196.100.99			18	15	1116	954	200

日志描述如下：

![Center 12][]

要求计算出每个手机用户的上行总流量，下行总流量和总流量。

需要自定义Bean实现

安装Maven插件  [http://blog.csdn.net/u013087513/article/details/71716667][http_blog.csdn.net_u013087513_article_details_71716667]

① 在Eclipse下新建maven project，如果出错，在当前项目的pom.xml中添加如下内容，刷新即可（需要有网络）

<dependencies> 
    	<dependency>
          <groupId>org.apache.maven.plugins</groupId>
          <artifactId>maven-resources-plugin</artifactId>
          <version>2.4.3</version>
    	</dependency>
    </dependencies>

② HDFS程序需要引入如下依赖

<dependencies>
    	<dependency>
    	  <groupId>junit</groupId>
    	  <artifactId>junit</artifactId>
    	  <version>4.8.2</version>
    	  <scope>test</scope>
    	</dependency>
    
    	<dependency>
    		<groupId>org.apache.hadoop</groupId>
    		<artifactId>hadoop-common</artifactId>
    		<version>2.4.1</version>
    	</dependency>
    
    	<dependency>
    		<groupId>org.apache.hadoop</groupId>
    		<artifactId>hadoop-hdfs</artifactId>
    		<version>2.4.1</version>
    	</dependency>
    </dependencies>

③mr程序需要引入依赖：

<dependencies>
    	<dependency>
    		<groupId>junit</groupId>
    		<artifactId>junit</artifactId>
    		<version>4.8.2</version>
    		<scope>test</scope>
    	</dependency>
    
    	<dependency>
    		<groupId>org.apache.hadoop</groupId>
    		<artifactId>hadoop-common</artifactId>
    		<version>2.4.1</version>
    	</dependency>
    
    	<dependency>
    		<groupId>org.apache.hadoop</groupId>
    		<artifactId>hadoop-mapreduce-client-core</artifactId>
    		<version>2.4.1</version>
    	</dependency>
    </dependencies>

开始程序的编写

一、自定义Bean DataBean实现Writable接口

package liuxun.hadoop.mr.dc;
    
    import java.io.DataInput;
    import java.io.DataOutput;
    import java.io.IOException;
    
    import org.apache.hadoop.io.Writable;
    
    public class DataBean implements Writable {
    
    	private String tel;
    
    	private long upPayLoad;
    
    	private long downPayLoad;
    
    	private long totalPayLoad;
    
    	public DataBean() {
    	}
    
    	public DataBean(String tel, long upPayLoad, long downPayLoad) {
    		this.tel = tel;
    		this.upPayLoad = upPayLoad;
    		this.downPayLoad = downPayLoad;
    		this.totalPayLoad = upPayLoad + downPayLoad;
    	}
    
    	@Override
    	public String toString() {
    		return this.upPayLoad + "\t" + this.downPayLoad + "\t" + this.totalPayLoad;
    	}
    
    	public void readFields(DataInput in) throws IOException {
    		this.tel = in.readUTF();
    		this.upPayLoad = in.readLong();
    		this.downPayLoad = in.readLong();
    		this.totalPayLoad = in.readLong();
    
    	}
    
    	// 注意两点：写入的顺序和写入的类型
    	public void write(DataOutput out) throws IOException {
    		out.writeUTF(tel);
    		out.writeLong(upPayLoad);
    		out.writeLong(downPayLoad);
    		out.writeLong(totalPayLoad);
    	}
    
    	public String getTel() {
    		return tel;
    	}
    
    	public void setTel(String tel) {
    		this.tel = tel;
    	}
    
    	public long getUpPayLoad() {
    		return upPayLoad;
    	}
    
    	public void setUpPayLoad(long upPayLoad) {
    		this.upPayLoad = upPayLoad;
    	}
    
    	public long getDownPayLoad() {
    		return downPayLoad;
    	}
    
    	public void setDownPayLoad(long downPayLoad) {
    		this.downPayLoad = downPayLoad;
    	}
    
    	public long getTotalPayLoad() {
    		return totalPayLoad;
    	}
    
    	public void setTotalPayLoad(long totalPayLoad) {
    		this.totalPayLoad = totalPayLoad;
    	}
    
    }

二、编写MapReduce程序 使用job进行组装

DataCount

package liuxun.hadoop.mr.dc;
    
    import java.io.IOException;
    
    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.Mapper;
    import org.apache.hadoop.mapreduce.Reducer;
    import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
    import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
    
    public class DataCount {
    
    	public static class DCMapper extends Mapper<LongWritable, Text, Text, DataBean>{
    
    		@Override
    		protected void map(LongWritable key, Text value, Context context)
    				throws IOException, InterruptedException {
    			//accept 
    			String line = value.toString();
    			//split
    			String[] fields = line.split("\t");
    			String tel = fields[1];
    			long up = Long.parseLong(fields[8]);
    			long down = Long.parseLong(fields[9]);	
    			DataBean bean = new DataBean(tel, up, down);
    			//send
    			context.write(new Text(tel), bean);
    		}
    		
    	}
    	
    	public static class DCReducer extends Reducer<Text, DataBean, Text, DataBean>{
    
    		@Override
    		protected void reduce(Text key, Iterable<DataBean> values, Context context)
    				throws IOException, InterruptedException {
    			long up_sum = 0;
    			long down_sum = 0;
    			for(DataBean bean : values){
    				up_sum += bean.getUpPayLoad();
    				down_sum += bean.getDownPayLoad();
    			}
    			DataBean bean = new DataBean("", up_sum, down_sum);
    			context.write(key, bean);
    		}
    		
    		
    	}
    	
    	
    	public static void main(String[] args) throws Exception {
    		Configuration conf = new Configuration();
    		Job job = Job.getInstance(conf);
    		
    		job.setJarByClass(DataCount.class);
    		
    		job.setMapperClass(DCMapper.class);
    		job.setMapOutputKeyClass(Text.class);
    		job.setMapOutputValueClass(DataBean.class);
    		FileInputFormat.setInputPaths(job, new Path(args[0]));
    		
    		job.setReducerClass(DCReducer.class);
    		job.setOutputKeyClass(Text.class);
    		job.setOutputValueClass(DataBean.class);
    		FileOutputFormat.setOutputPath(job, new Path(args[1]));
    		
    		job.waitForCompletion(true);
    		
    	}
    	
    
    }

打包上传测试运行 最后查看运行结果

![Center 13][]

# Hadoop远程Debug #

注意：  
如果没有打包在Linux上运行，而是在Eclipse上调试运行是按本地模式启动的 只会启动一个Mapper和一个Reducer  看不到真正的集群效果   
JPDA 简介  
Sun Microsystem 的 Java Platform Debugger Architecture (JPDA) 技术是一个多层架构，使您能够在各种环境中轻松调试 Java 应用程序。JPDA 由两个接口（分别是 JVM Tool Interface 和 JDI）、一个协议（Java Debug Wire Protocol）和两个用于合并它们的软件组件（后端和前端）组成。它的设计目的是让调试人员在任何环境中都可以进行调试。  
更详细的介绍，可以参考使用 Eclipse 远程调试 Java 应用程序  
JDWP 设置  
JVM本身就支持远程调试，Eclipse也支持JDWP，只需要在各模块的JVM启动时加载以下参数：  
dt\_socket表示使用套接字传输。  
address=8000  
JVM在8000端口上监听请求，这个设定为一个不冲突的端口即可。  
server=y   
y表示启动的JVM是被调试者。如果为n，则表示启动的JVM是调试器。  
suspend=y  
y表示启动的JVM会暂停等待，直到调试器连接上才继续执行。suspend=n，则JVM不会暂停等待。  
  
需要在$HADOOP\_HOME/etc/hadoop/hadoop-env.sh文件的最后添加需要debug的进程  
\#远程调试namenode  
export HADOOP\_NAMENODE\_OPTS="-agentlib:jdwp=transport=dt\_socket,address=8888,server=y,suspend=y"  
\#远程调试datanode  
export HADOOP\_DATANODE\_OPTS="-agentlib:jdwp=transport=dt\_socket,address=9888,server=y,suspend=y"  
\#远程调试RM  
export YARN\_RESOURCEMANAGER\_OPTS="-agentlib:jdwp=transport=dt\_socket,address=10888,server=y,suspend=y"  
\#远程调试NM  
export YARN\_NODEMANAGER\_OPTS="-agentlib:jdwp=transport=dt\_socket,address=10888,server=y,suspend=y"

测试：

单个启动： daemon 守护进程，后台程序  
hadoop-daemon.sh start namenode  
hadoop-daemon.sh start datanode  
测试案例：测试NameNode和DataNode  
编辑$HADOOP\_HOME/etc/hadoop/hadoop-env.sh 添加如下配置  
\#远程调试namenode  
export HADOOP\_NAMENODE\_OPTS="-agentlib:jdwp=transport=dt\_socket,address=8888,server=y,suspend=y"  
\#远程调试datanode  
export HADOOP\_DATANODE\_OPTS="-agentlib:jdwp=transport=dt\_socket,address=9888,server=y,suspend=y"  
单个启动：

![Center 14][]

然后在Eclipse中Ctrl+shift+t（Mac上是command+shift+t）查找NameNode  
然后关联源码：再使用command+o 查找Main方法

选中main方法名 右键Debug as—>Debug Configurations

然后在弹窗内使用Remote Java Application

![Center 15][]

在SecureCRT 输入jps 发现 NameNode已经启动了   
 ![Center 16][]  
注意：Debug时如果不打断点 直接就运行了 是不会停住的  
接下来断点调试DataNode  
在main方法中打断点   
 ![Center 17][]  
以下步骤上同   
 ![Center 18][]  
然后apply  ， debug 就OK了  
 ![Center 19][]  
远程调试一般在分布式上用的最多。  
例如在Eclipse调试时 普通调试只能调试客户端的代码，当通过RPC协议与云主机上的Server进行通讯时，即使导入全部的源码 也无法走到服务器端的程序，因为当与服务器交互时走的是服务器上的程序。只有通过远程调试才可进入Server端程序进行查看。  
远程调试技巧：  
客户端与服务端进行交互一般都是通过某种协议进行通讯，例如Hadoop是通过RPC通讯的。共同点是服务端和实现了协议接口，客户端通过协议接口代理服务端的程序，调用协议中的方法，进行通讯。  
首先找到通讯的协议接口 查找协议接口的实现类(协议接口的实现是服务器端实现的)  
在需要调试的部分查找客户端调用的是协议接口中的哪个方法，选中右键查看其方法的实现  
然后打上断点调试即可。

[Center]: /images/20220609/232b730a15054e1bb21e12b80f6d66a6.png
[Center 1]: /images/20220609/faaa1ce92f614936a76d21a24788d4c7.png
[Center 2]: /images/20220609/dcb01797df544cc6825cc773072d8a3c.png
[Center 3]: /images/20220609/e7759cff8f7b471d90ddf81a704149fc.png
[Center 4]: /images/20220609/f4bb9f553fc14220ab03fb01a7441226.png
[Center 5]: /images/20220609/c79f0087e7a744228ebfba47c4b4a879.png
[Center 6]: /images/20220609/f3002bb7a7e34571939f2374c7188a69.png
[Center 7]: /images/20220609/0353ae7446c14aadb6a6a6debfb45dc7.png
[Center 8]: /images/20220609/7ec3932a3df34ffb90cae78b9b03abcb.png
[Center 9]: /images/20220609/05111d547b604d16951387652b331356.png
[Center 10]: /images/20220609/6c18ceafd5ac4c859480dfed76ca4c22.png
[Center 11]: /images/20220609/5b762d4d8c04494bb7bcbf98eede30ea.png
[Center 12]: /images/20220609/61d2ea1a7e8849f6b63a90d731a4c66a.png
[http_blog.csdn.net_u013087513_article_details_71716667]: http://blog.csdn.net/u013087513/article/details/71716667
[Center 13]: /images/20220609/c7f8c57387b34e86bcaac5f2d561bc33.png
[Center 14]: /images/20220609/fe796df2ca0a44dca012102d066d9bf0.png
[Center 15]: /images/20220609/701cdab0fa6649c9ac31f0f55c1f756d.png
[Center 16]: /images/20220609/dbdcffda71334295997c547bf81cc198.png
[Center 17]: /images/20220609/c6683541638f4c50a371193404c13e21.png
[Center 18]: /images/20220609/4cde2051380940e998c8c264156c9307.png
[Center 19]: /images/20220609/f60c5f0e49544294843c8b27fc4f15d1.png