hadoop记录篇5-eclipse开发mapreduce

一。 MAP REDUCE执行原理

    MapReduce主要分为三个阶段 Map阶段 Shuffle阶段 Reduce阶段   

1》Map阶段：

Hadoop Map/Reduce框架为每一个InputSplit产生一个map任务，而每个InputSplit是由该作业的InputFormat产生的。

1》InputSplit的大小算法 

通过查看FileInputFormat

protected long computeSplitSize(long blockSize, long minSize,
                                  long maxSize) {
    return Math.max(minSize, Math.min(maxSize, blockSize));
  }
blockSize hadoop2之后默认是128M （http://hadoop.apache.org/docs/r2.6.5/hadoop-project-dist/hadoop-hdfs/hdfs-default.xml） 参数名：dfs.blocksize

minsize 找到调用computeSplitSize的方法同类的

public List<InputSplit> getSplits(JobContext job) throws IOException {
    StopWatch sw = new StopWatch().start();
    long minSize = Math.max(getFormatMinSplitSize(), getMinSplitSize(job));
    long maxSize = getMaxSplitSize(job);

查看过getFormatMinSplitSize() 直接返回 1 

protected long getFormatMinSplitSize() {
    return 1;
  }
getMinSplitSize(job)

获取 mapreduce.input.fileinputformat.split.minsize（切片的最小个数）

参数值 如果没有设置默认 1  查看官方文档该参数确实没有 值是0 所以这里代码就是1
getMaxSplitSize(job)

获取参数mapreduce.input.fileinputformat.split.maxsize值 官方文档没有 取默认值Long.MAX_VALUE

数据计算 InputSplit个数 如果参数都是默认值 =Block快的个数

假设文件大小大于 128M 比如 130M 被会分为2个block
运算结果是
long minSize = Math.max(1, 1);  =1
long maxSize = getMaxSplitSize(job);=Long.maxValue =9,223,372,036,854,775,807

Math.max(minSize, Math.min(maxSize, blockSize));
Math.max(1, Math.min(9,223,372,036,854,775,807, 2)); =2个切片

假设 是100M 会被切成 1个block
Math.max(1, Math.min(9,223,372,036,854,775,807, 1)); =1个切片

输出键值对不需要与输入键值对的类型一致。一个给定的输入键值对可以映射成0个或多个输出键值对（该过程用户实现）

2》Shuffle阶段：

  每个map 对应的输出数据会被存储在内存中 通过key排序 然后产生分区号 该分区号 key的hash%reduce个数 也就是将来进入哪个reduce task

默认reudce只有一个 所以就是 %1 永远是0 就是第一个 当排序好后 最后合并到磁盘中 等待其他的map程序 执行完成 

 reduce程序等待所有的map执行完后  同步多个map的数据（通过分区号获取对应map中键值对） 进行合并 合并后 再次进行排序 通过key进行分组

将分组数据调用reduce程序 


3》Reduce阶段：

  reduce将每组的数据处理后 输出到hdfs中  多个reduce就会存在多个输出





二。 插件安装

 hadoop2.x的插件源代码地址  https://github.com/winghc/hadoop2x-eclipse-plugin 必须通过ant安装

apache官方下载ant 

设置解压后目录 bin到 path环境变量下 配置好后测试 

C:Usersjiaozi>ant
Buildfile: build.xml does not exist!
Build failed

具体编译步骤 github中有参考

下载的插件源代码 假如解压在 D:learnhadoophadoop2x-eclipse-plugin-master

插件安装需要下载对应hadoop的jar包所以需要maven仓库 默认是repo1.apache.org 非常慢 修改为aliyun

修改 D:learnhadoophadoop2x-eclipse-plugin-masterivyivysettings.xml

修改为
<property name="repo.maven.org"
    value="http://maven.aliyun.com/nexus/content/groups/public"
    override="false"/>
该插件编译指定hadoop安装包 注意jdk版本和eclipse版本 

hadoop2.7.4 jdk1.8  eclipse mars2不支持jdk1.8
hadoop2.6.5. jdk1.7  eclipse mars2支持jdk1.7
我这里 下载了hadoop2.6.5 解压在D:learnhadoophadoop-2.6.5下 

执行命令

ant jar -Dversion=2.6.5 -Dhadoop.version=2.6.5 -Declipse.home=E:/开发软件/eclipse -Dhadoop.home=D:/learn/hadoop/hadoop-2.6.5
插件最终生成在 D:learnhadoophadoop2x-eclipse-plugin-masterbincontribeclipse-plugin
碰到问题1：
缺少某个版本的jar包 比如hadoop-2.6.5sharehadoopcommonlibcommons-collections-3.2.1.jar
  结果发现该目录下 有个 commons-collections-3.2.2.jar 复制一个改成 commons-collections-3.2.1.jar即可
其他包都这样处理
碰到问题2：

插件丢到 eclipse/dropings目录 发现无法安装 
调试eclipse

 eclipse的快捷方式上 添加  -clean -consolelog -debug

首选项 会发现插件 但是点击时出现 unsupported major.minor version 52.0
 说明eclipse加载时使用的jdk版本 比该插件的编译版本 52.0要低  

此时 重新使用 eclipse加载的jdk（path和java_home中定义的jdk）来重新编译hadoop插件 重新丢到eclipse/dropings
删除eclipse目录configuration下的org.eclipse.update 重启eclipse 发现成功安装

安装后 发现插件在Project Explorer下多了一个 DFS location的列表项
 切换到Map/Reduce视图 添加上对应dfs主机和端口


但是dfs无论如何都连接不上 死命重连
  hadoop java.lang.IllegalStateException
后来自己使用maven 使用java客户端 发现都可以正常操作 
看eclipse的日志 comgooglecommonasePreconditions validate失败 
后来找到了这个jar包时guava-11.0.2.jar
maven添加依赖
<dependency>
 <groupId>org.apache.hadoop</groupId>
 <artifactId>hadoop-client</artifactId>
 <version>2.6.5</version>
 <scope>compile</scope>
</dependency>
的 guava-11.0.2.jar里面却没有这个类 
我尝试将maven引入的所有jar包替换掉eclipse-plugin-2.6.5.jar下的lib目录 不是删除替换 是直接替换（该有的还是保留）
重新发布到eclipse 发现成功 

三。简单mapreduce例子

   mapreduce官网提供的统计词频的例子 一个文本文件统计总共多少词每个词出现的次数

添加maven依赖

<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
  <modelVersion>4.0.0</modelVersion>
  <groupId>cn.et</groupId>
  <artifactId>remote_hadoop</artifactId>
  <version>0.0.1-SNAPSHOT</version>
  <dependencies>
  	<dependency>
		  <groupId>org.apache.hadoop</groupId>
		  <artifactId>hadoop-client</artifactId>
		  <version>2.7.4</version>
		  <scope>compile</scope>
		</dependency>
  </dependencies>
</project>
添加mapreudce处理类 

package hadoop;

import java.io.IOException;
import java.util.Iterator;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
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 ReduceTest {

	public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException {
		// 设置当前机器的hadoop目录
		System.setProperty("hadoop.home.dir", "D:\learn\hadoop\hadoop-2.7.4");
		// 设置操作使用的用户 如果不设置为root 和 hadoop服务的相同 出现异常 本机账号是window账号
		System.setProperty("HADOOP_USER_NAME", "root");
		
		Configuration conf = new Configuration();
		conf.set("mapreduce.framework.name", "yarn");
		//window运行需要设置这行 linux运行不能添加这样
		conf.set("mapreduce.job.jar", "target\remote_hadoop-0.0.1-SNAPSHOT.jar");
		FileSystem fs = FileSystem.get(conf);
		if(fs.exists(new Path("/user/root/output")))
			fs.delete(new Path("/user/root/output"),true);
		Job job =Job.getInstance(conf, "WordCount");
		//不设置运行的主类 使用 hadoop jar jar包 主类 无法找到 Mapper和Reduce的类
		job.setJarByClass(ReduceTest.class);
		job.setMapperClass(MyMapper.class);
		job.setMapOutputKeyClass(Text.class);
		job.setMapOutputValueClass(IntWritable.class);
		job.setReducerClass(MyReduce.class);
		FileInputFormat.addInputPath(job,new Path("/user/root/input"));
		FileOutputFormat.setOutputPath(job, new Path("/user/root/output"));
		job.waitForCompletion(true);
	}
	public static class MyMapper extends Mapper<LongWritable,Text, Text, IntWritable>{
    	public MyMapper() {
		}
    	@Override
    	protected void map(LongWritable key, Text value, Mapper<LongWritable, Text, Text, IntWritable>.Context context)
    			throws IOException, InterruptedException {
    		String[] split=value.toString().split(" ");
    		for(String s:split){
    			context.write(new Text(s),new IntWritable(1));
    		}
    	}
    }//内部类必须是静态的
    public static class MyReduce extends Reducer<Text, IntWritable, Text, IntWritable>{
    	public MyReduce() {
		}
    	@Override
    	protected void reduce(Text arg0, Iterable<IntWritable> arg1,
    			Reducer<Text, IntWritable, Text, IntWritable>.Context arg2) throws IOException, InterruptedException {
    		Iterator<IntWritable> ite=arg1.iterator();
    		int i=0;
    		while(ite.hasNext()){
    			i++;
    			ite.next();
    		}
    		arg2.write(arg0, new IntWritable(i));
    	}
    }
}

使用命令 mvn package打包 （我这里包名为remote_hadoop-0.0.1-SNAPSHOT.jar）上传到namenode机器node1

执行命令 运行

 hadoop jar ./remote_hadoop-0.0.1-SNAPSHOT.jar hadoop.ReduceTest
结果为： 

[root@node1 ~]# hadoop jar ./remote_hadoop-0.0.1-SNAPSHOT.jar hadoop.ReduceTest
17/10/27 00:51:40 WARN mapreduce.JobResourceUploader: Hadoop command-line option parsing not performed. Implement the Tool interface and execute your application with ToolRunner to remedy this.
17/10/27 00:51:41 INFO input.FileInputFormat: Total input paths to process : 1
17/10/27 00:51:41 INFO mapreduce.JobSubmitter: number of splits:1
17/10/27 00:51:41 INFO mapreduce.JobSubmitter: Submitting tokens for job: job_1509078980537_0006
17/10/27 00:51:42 INFO impl.YarnClientImpl: Submitted application application_1509078980537_0006
17/10/27 00:51:42 INFO mapreduce.Job: The url to track the job: http://node1:8088/proxy/application_1509078980537_0006/
17/10/27 00:51:42 INFO mapreduce.Job: Running job: job_1509078980537_0006
17/10/27 00:51:57 INFO mapreduce.Job: Job job_1509078980537_0006 running in uber mode : false
17/10/27 00:51:57 INFO mapreduce.Job:  map 0% reduce 0%
17/10/27 00:52:08 INFO mapreduce.Job:  map 100% reduce 0%
17/10/27 00:52:19 INFO mapreduce.Job:  map 100% reduce 100%
17/10/27 00:52:19 INFO mapreduce.Job: Job job_1509078980537_0006 completed successfully
17/10/27 00:52:20 INFO mapreduce.Job: Counters: 49

这里我发现 在window上 代码中添加 

conf.set("mapreduce.job.jar", "target\remote_hadoop-0.0.1-SNAPSHOT.jar");
将所有的hadoop配置文件放在classpath下



虽然可以提交任务 一直卡住

17/10/27 00:51:57 INFO mapreduce.Job: Job job_1509078980537_0006 running in uber mode : false

一直都不出现 map 0% reduce 0%  奇怪 如果直接打包到linux执行命令 没有问题  

四。window调试mapreduce

  下载hadoop安装包 我解压在 D:/learn/hadoop/hadoop-2.6.5下 window环境下存在一些 exe和dll文件才能启动hadoop

下载地址   https://github.com/SweetInk/hadoop-common-2.7.1-bin  拷贝下载文件下的所有文件到 hadoop下的bin目录

尝试启动bin目录下的start-dfs.cmd看是否能正常启动 

c盘下创建目录 c:/user/input

创建项目 maven配置同上

 处理类LocalReduceTest

package hadoop;

import java.io.IOException;
import java.util.Iterator;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
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 LocalReduceTest {

	public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException {
		// 设置当前机器的hadoop目录
		System.setProperty("hadoop.home.dir", "D:\learn\hadoop\hadoop-2.7.4");
		Configuration conf = new Configuration();
		FileSystem fs = FileSystem.get(conf);
		if(fs.exists(new Path("c:/user/output")))
			fs.delete(new Path("c:/user/output"),true);
		Job job =Job.getInstance(conf, "WordCount");
		job.setMapperClass(MyMapper.class);
		job.setMapOutputKeyClass(Text.class);
		job.setMapOutputValueClass(IntWritable.class);
		job.setReducerClass(MyReduce.class);
		FileInputFormat.addInputPath(job,new Path("c:/user/input"));
		FileOutputFormat.setOutputPath(job, new Path("c:/user/output"));
		job.waitForCompletion(true);
	}
	public static class MyMapper extends Mapper<LongWritable,Text, Text, IntWritable>{
    	public MyMapper() {
		}
    	@Override
    	protected void map(LongWritable key, Text value, Mapper<LongWritable, Text, Text, IntWritable>.Context context)
    			throws IOException, InterruptedException {
    		String[] split=value.toString().split(" ");
    		for(String s:split){
    			context.write(new Text(s),new IntWritable(1));
    		}
    	}
    }
    public static class MyReduce extends Reducer<Text, IntWritable, Text, IntWritable>{
    	public MyReduce() {
		}
    	@Override
    	protected void reduce(Text arg0, Iterable<IntWritable> arg1,
    			Reducer<Text, IntWritable, Text, IntWritable>.Context arg2) throws IOException, InterruptedException {
    		Iterator<IntWritable> ite=arg1.iterator();
    		int i=0;
    		while(ite.hasNext()){
    			i++;
    			ite.next();
    		}
    		arg2.write(arg0, new IntWritable(i));
    	}
    }
}
直接点击main方法run 发现报错 

Exception in thread "main" java.lang.UnsatisfiedLinkError: org.apache.hadoop.io.nativeio.NativeIO$Windows.access0(Ljava/lang/String;I)Z
	at org.apache.hadoop.io.nativeio.NativeIO$Windows.access0(Native Method)
	at org.apache.hadoop.io.nativeio.NativeIO$Windows.access(NativeIO.java:557)
	at org.apache.hadoop.fs.FileUtil.canRead(FileUtil.java:977)
NativeIO Window子类中access0这个方法没有办法链接到本地动态链接库 该方法是判断是否有进入权限的修改源代码  下载2.65.源代码找到该类



再次运行 main方法 发现成功  并且 c:/user/out输出了结果  而且速度快于linux的集群（因为本机不用远程资源调度） 该中方式可以直接断点调试


【推广】 免费学中医，健康全家人
原文地址：https://www.cnblogs.com/liaomin416100569/p/9331178.html