使用Hadoop里面的MapReduce来处理海量数据 -

weitao1026

浏览: 1048265 次
性别:
来自: 上海

最近访客更多访客>>

vicento4

723499280

liuzidong

s1986q

博主相关

博客

微博

相册

留言

关于我

文章分类

社区版块

存档分类

使用Hadoop里面的MapReduce来处理海量数据

博客分类：

hadoop/spark/Hbase/Hive/pig/Zookeeper/HAWQ/cloudera/Impala/Oozie

hadoop

使用Hadoop里面的MapReduce来处理海量数据是非常简单方便的，但有时候我们的应用程序，往往需要多个MR作业，来计算结果，比如说一个最简单的使用MR提取海量搜索日志的TopN的问题，注意，这里面，其实涉及了两个MR作业，第一个是词频统计，第两个是排序求TopN，这显然是需要两个MapReduce作业来完成的。其他的还有，比如一些数据挖掘类的作业，常常需要迭代组合好几个作业才能完成，这类作业类似于DAG类的任务，各个作业之间是具有先后，或相互依赖的关系，比如说，这一个作业的输入，依赖上一个作业的输出等等。

在Hadoop里实际上提供了，JobControl类，来组合一个具有依赖关系的作业，在新版的API里，又新增了ControlledJob类，细化了任务的分配，通过这两个类，我们就可以轻松的完成类似DAG作业的模式，这样我们就可以通过一个提交来完成原来需要提交2次的任务，大大简化了任务的繁琐度。具有依赖式的作业提交后，hadoop会根据依赖的关系，先后执行的job任务，每个任务的运行都是独立的。

下面来看下散仙的例子，组合一个词频统计+排序的作业，测试数据如下：

Java代码

秦东亮;72
秦东亮;34
秦东亮;100
三劫;899
三劫;32
三劫;1
a;45
b;567
b;12

秦东亮;72
秦东亮;34
秦东亮;100
三劫;899
三劫;32
三劫;1
a;45
b;567
b;12

代码如下：

Java代码

package com.qin.test.hadoop.jobctrol;
import java.io.IOException;
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.io.WritableComparator;
import org.apache.hadoop.mapred.JobConf;
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.jobcontrol.ControlledJob;
import org.apache.hadoop.mapreduce.lib.jobcontrol.JobControl;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
/**
* Hadoop的版本是1.2的
* JDK环境1.6
* 使用ControlledJob+JobControl新版API
* 完成组合式任务
* 第一个任务是统计词频
* 第二个任务是降序排序
*
* 如果使用MapReduce作业来完成的话，则需要跑2个MR任务
* 但是如果我们使用了JobControl+ControlledJob就可以在
* 一个类里面完成类型的DAG依赖式的作业
*
*
* @author qindongliang
*
*
*
* ***/
public class MyHadoopControl {
/***
*
*MapReduce作业1的Mapper
*
*LongWritable 1 代表输入的key值，默认是文本的位置偏移量
*Text 2 每行的具体内容
*Text 3 输出的Key类型
*Text 4 输出的Value类型
*
* */
private static class SumMapper extends Mapper<LongWritable, Text, Text, IntWritable>{
private Text t=new Text();
private IntWritable one=new IntWritable(1);
/**
*
* map阶段输出词频
*
*
* **/
@Override
protected void map(LongWritable key, Text value,Context context)
throws IOException, InterruptedException {
String data=value.toString();
String words[]=data.split(";");
if(words[0].trim()!=null){
t.set(" "+words[0]);//赋值K
one.set(Integer.parseInt(words[1]));
context.write(t, one);
}
}
}
/**
* MapReduce作业1的Reducer
* 负责词频累加，并输出
*
* **/
private static class SumReduce extends Reducer<Text, IntWritable, IntWritable, Text>{
//存储词频对象
private IntWritable iw=new IntWritable();
@Override
protected void reduce(Text key, Iterable<IntWritable> value,Context context)
throws IOException, InterruptedException {
int sum=0;
for(IntWritable count:value){
sum+=count.get();//累加词频
}
iw.set(sum);//设置词频
context.write(iw, key);//输出数据
}
}
/**
* MapReduce作业2排序的Mapper
*
* **/
private static class SortMapper extends Mapper<LongWritable, Text, IntWritable, Text>{
IntWritable iw=new IntWritable();//存储词频
private Text t=new Text();//存储文本
@Override
protected void map(LongWritable key, Text value,Context context)throws IOException, InterruptedException {
String words[]=value.toString().split(" ");
System.out.println("数组的长度: "+words.length);
System.out.println("Map读入的文本: "+value.toString());
System.out.println("=====> "+words[0]+" =====>"+words[1]);
if(words[0]!=null){
iw.set(Integer.parseInt(words[0].trim()));
t.set(words[1].trim());
context.write(iw, t);//map阶段输出，默认按key排序
}
}
}
/**
* MapReduce作业2排序的Reducer
*
* **/
private static class SortReduce extends Reducer<IntWritable, Text, Text, IntWritable>{
/**
*
* 输出排序内容
*
* **/
@Override
protected void reduce(IntWritable key, Iterable<Text> value,Context context)
throws IOException, InterruptedException {
for(Text t:value){
context.write(t, key);//输出排好序后的K,V
}
}
}
/***
* 排序组件，在排序作业中，需要使用
* 按key的降序排序
*
* **/
public static class DescSort extends WritableComparator{
public DescSort() {
super(IntWritable.class,true);//注册排序组件
}
@Override
public int compare(byte[] arg0, int arg1, int arg2, byte[] arg3,
int arg4, int arg5) {
return -super.compare(arg0, arg1, arg2, arg3, arg4, arg5);//注意使用负号来完成降序
}
@Override
public int compare(Object a, Object b) {
return -super.compare(a, b);//注意使用负号来完成降序
}
}
/**
* 驱动类
*
* **/
public static void main(String[] args)throws Exception {
JobConf conf=new JobConf(MyHadoopControl.class);
conf.set("mapred.job.tracker","192.168.75.130:9001");
conf.setJar("tt.jar");
System.out.println("模式： "+conf.get("mapred.job.tracker"));;
/**
*
*作业1的配置
*统计词频
*
* **/
Job job1=new Job(conf,"Join1");
job1.setJarByClass(MyHadoopControl.class);
job1.setMapperClass(SumMapper.class);
job1.setReducerClass(SumReduce.class);
job1.setMapOutputKeyClass(Text.class);//map阶段的输出的key
job1.setMapOutputValueClass(IntWritable.class);//map阶段的输出的value
job1.setOutputKeyClass(IntWritable.class);//reduce阶段的输出的key
job1.setOutputValueClass(Text.class);//reduce阶段的输出的value
//加入控制容器
ControlledJob ctrljob1=new ControlledJob(conf);
ctrljob1.setJob(job1);
FileInputFormat.addInputPath(job1, new Path("hdfs://192.168.75.130:9000/root/input"));
FileSystem fs=FileSystem.get(conf);
Path op=new Path("hdfs://192.168.75.130:9000/root/op");
if(fs.exists(op)){
fs.delete(op, true);
System.out.println("存在此输出路径，已删除！！！");
}
FileOutputFormat.setOutputPath(job1, op);
/**========================================================================*/
/**
*
*作业2的配置
*排序
*
* **/
Job job2=new Job(conf,"Join2");
job2.setJarByClass(MyHadoopControl.class);
//job2.setInputFormatClass(TextInputFormat.class);
job2.setMapperClass(SortMapper.class);
job2.setReducerClass(SortReduce.class);
job2.setSortComparatorClass(DescSort.class);//按key降序排序
job2.setMapOutputKeyClass(IntWritable.class);//map阶段的输出的key
job2.setMapOutputValueClass(Text.class);//map阶段的输出的value
job2.setOutputKeyClass(Text.class);//reduce阶段的输出的key
job2.setOutputValueClass(IntWritable.class);//reduce阶段的输出的value
//作业2加入控制容器
ControlledJob ctrljob2=new ControlledJob(conf);
ctrljob2.setJob(job2);
/***
*
* 设置多个作业直接的依赖关系
* 如下所写：
* 意思为job2的启动，依赖于job1作业的完成
*
* **/
ctrljob2.addDependingJob(ctrljob1);
//输入路径是上一个作业的输出路径
FileInputFormat.addInputPath(job2, new Path("hdfs://192.168.75.130:9000/root/op/part*"));
FileSystem fs2=FileSystem.get(conf);
Path op2=new Path("hdfs://192.168.75.130:9000/root/op2");
if(fs2.exists(op2)){
fs2.delete(op2, true);
System.out.println("存在此输出路径，已删除！！！");
}
FileOutputFormat.setOutputPath(job2, op2);
// System.exit(job2.waitForCompletion(true) ? 0 : 1);
/**====================================================================***/
/**
*
* 主的控制容器，控制上面的总的两个子作业
*
* **/
JobControl jobCtrl=new JobControl("myctrl");
//ctrljob1.addDependingJob(ctrljob2);// job2在job1完成后，才可以启动
//添加到总的JobControl里，进行控制
jobCtrl.addJob(ctrljob1);
jobCtrl.addJob(ctrljob2);
//在线程启动
Thread t=new Thread(jobCtrl);
t.start();
while(true){
if(jobCtrl.allFinished()){//如果作业成功完成，就打印成功作业的信息
System.out.println(jobCtrl.getSuccessfulJobList());
jobCtrl.stop();
break;
}
if(jobCtrl.getFailedJobList().size()>0){//如果作业失败，就打印失败作业的信息
System.out.println(jobCtrl.getFailedJobList());
jobCtrl.stop();
break;
}
}
}
}

package com.qin.test.hadoop.jobctrol;

import java.io.IOException;

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.io.WritableComparator;
import org.apache.hadoop.mapred.JobConf;
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.jobcontrol.ControlledJob;
import org.apache.hadoop.mapreduce.lib.jobcontrol.JobControl;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;




/**
 * Hadoop的版本是1.2的
 * JDK环境1.6
 * 使用ControlledJob+JobControl新版API
 * 完成组合式任务 
 * 第一个任务是统计词频
 * 第二个任务是降序排序
 * 
 * 如果使用MapReduce作业来完成的话，则需要跑2个MR任务
 * 但是如果我们使用了JobControl+ControlledJob就可以在
 * 一个类里面完成类型的DAG依赖式的作业
 * 
 * 
 * @author qindongliang
 * 
 * 
 * 
 * ***/
public class MyHadoopControl {
	
	
	
	/***
	 *
	 *MapReduce作业1的Mapper
	 *
	 *LongWritable 1  代表输入的key值，默认是文本的位置偏移量
	 *Text 2          每行的具体内容
	 *Text 3          输出的Key类型
	 *Text 4          输出的Value类型
	 * 
	 * */
	private static class SumMapper extends Mapper<LongWritable, Text, Text, IntWritable>{
		
		private Text t=new Text();
        private IntWritable one=new IntWritable(1);
		
        /**
         * 
         * map阶段输出词频
         * 
         * 
         * **/
		@Override
		protected void map(LongWritable key, Text value,Context context)
				throws IOException, InterruptedException {
			String data=value.toString();
			String words[]=data.split(";");
		  if(words[0].trim()!=null){
			  t.set(" "+words[0]);//赋值K
			  one.set(Integer.parseInt(words[1]));
			  context.write(t, one);
		  }	
		}
		
	}
	
	/**
	 * MapReduce作业1的Reducer
	 * 负责词频累加，并输出
	 * 
	 * **/
	private static class SumReduce extends Reducer<Text, IntWritable, IntWritable, Text>{
		
		//存储词频对象
		private IntWritable iw=new IntWritable();
		
		@Override
		protected void reduce(Text key, Iterable<IntWritable> value,Context context)
				throws IOException, InterruptedException {
		 
			
			int sum=0;
			for(IntWritable count:value){
				sum+=count.get();//累加词频
			}
			iw.set(sum);//设置词频
			context.write(iw, key);//输出数据
			
			
			
			
			
		}
		
	}
	
	
	/**
	 * MapReduce作业2排序的Mapper
	 * 
	 * **/
	private static class SortMapper  extends Mapper<LongWritable, Text, IntWritable, Text>{
		
		
		IntWritable iw=new IntWritable();//存储词频
		private Text t=new Text();//存储文本
		@Override
		protected void map(LongWritable key, Text value,Context context)throws IOException, InterruptedException {
			
			String words[]=value.toString().split(" ");
		   System.out.println("数组的长度: "+words.length);
			System.out.println("Map读入的文本: "+value.toString());
			System.out.println("=====>  "+words[0]+"  =====>"+words[1]);
			 if(words[0]!=null){
				 iw.set(Integer.parseInt(words[0].trim()));
				 t.set(words[1].trim());
				 context.write(iw, t);//map阶段输出，默认按key排序
			 }
			
			 
			
		}
		
		
		
	}
	
	
	/**
	 * MapReduce作业2排序的Reducer
	 * 
	 * **/
	private static class SortReduce extends Reducer<IntWritable, Text, Text, IntWritable>{
		
		
		
		/**
		 * 
		 * 输出排序内容
		 * 
		 * **/
		@Override
		protected void reduce(IntWritable key, Iterable<Text> value,Context context)
				throws IOException, InterruptedException {
		 
			 for(Text t:value){
				 context.write(t, key);//输出排好序后的K,V
			 }
			
		}
		
	}
	
	
	
	
	/***
	 * 排序组件，在排序作业中，需要使用
	 * 按key的降序排序
	 * 
	 * **/
		public static class DescSort extends  WritableComparator{

			 public DescSort() {
				 super(IntWritable.class,true);//注册排序组件
			}
			 @Override
			public int compare(byte[] arg0, int arg1, int arg2, byte[] arg3,
					int arg4, int arg5) {
				return -super.compare(arg0, arg1, arg2, arg3, arg4, arg5);//注意使用负号来完成降序
			}
			 
			 @Override
			public int compare(Object a, Object b) {
		 
				return   -super.compare(a, b);//注意使用负号来完成降序
			}
			
		}
	
	
	
	
	
	
    /**
     * 驱动类
     * 
     * **/
 	public static void main(String[] args)throws Exception {
	
		
		   JobConf conf=new JobConf(MyHadoopControl.class); 
		   conf.set("mapred.job.tracker","192.168.75.130:9001");
		   conf.setJar("tt.jar");
		 
		 System.out.println("模式：  "+conf.get("mapred.job.tracker"));;
		
		
		/**
		 * 
		 *作业1的配置
		 *统计词频
		 * 
		 * **/
		Job job1=new Job(conf,"Join1");
	    job1.setJarByClass(MyHadoopControl.class);
		
	    job1.setMapperClass(SumMapper.class);
		job1.setReducerClass(SumReduce.class);
		
		job1.setMapOutputKeyClass(Text.class);//map阶段的输出的key
		job1.setMapOutputValueClass(IntWritable.class);//map阶段的输出的value
		
		job1.setOutputKeyClass(IntWritable.class);//reduce阶段的输出的key
		job1.setOutputValueClass(Text.class);//reduce阶段的输出的value
		
	
		
		//加入控制容器
		ControlledJob ctrljob1=new  ControlledJob(conf);
		ctrljob1.setJob(job1);
		
		
		FileInputFormat.addInputPath(job1, new Path("hdfs://192.168.75.130:9000/root/input"));
        FileSystem fs=FileSystem.get(conf);
		 
		 Path op=new Path("hdfs://192.168.75.130:9000/root/op");
		 
		 if(fs.exists(op)){
			 fs.delete(op, true);
			 System.out.println("存在此输出路径，已删除！！！");
		 }
		FileOutputFormat.setOutputPath(job1, op);
		
	/**========================================================================*/
		
		/**
		 * 
		 *作业2的配置
		 *排序
		 * 
		 * **/
		Job job2=new Job(conf,"Join2");
	    job2.setJarByClass(MyHadoopControl.class);
		
	    //job2.setInputFormatClass(TextInputFormat.class);
	    
	    
	    job2.setMapperClass(SortMapper.class);
		job2.setReducerClass(SortReduce.class);
		
		job2.setSortComparatorClass(DescSort.class);//按key降序排序
		
		job2.setMapOutputKeyClass(IntWritable.class);//map阶段的输出的key
		job2.setMapOutputValueClass(Text.class);//map阶段的输出的value
		
		job2.setOutputKeyClass(Text.class);//reduce阶段的输出的key
		job2.setOutputValueClass(IntWritable.class);//reduce阶段的输出的value
		
		

		//作业2加入控制容器
		ControlledJob ctrljob2=new ControlledJob(conf);
		ctrljob2.setJob(job2);
		
		/***
		 * 
		 * 设置多个作业直接的依赖关系
		 * 如下所写：
		 * 意思为job2的启动，依赖于job1作业的完成
		 * 
		 * **/
		ctrljob2.addDependingJob(ctrljob1);
		
		
		
		//输入路径是上一个作业的输出路径
		FileInputFormat.addInputPath(job2, new Path("hdfs://192.168.75.130:9000/root/op/part*"));
        FileSystem fs2=FileSystem.get(conf);
		 
		 Path op2=new Path("hdfs://192.168.75.130:9000/root/op2");
		 if(fs2.exists(op2)){
			 fs2.delete(op2, true);
			 System.out.println("存在此输出路径，已删除！！！");
		 }
		FileOutputFormat.setOutputPath(job2, op2);
		
		// System.exit(job2.waitForCompletion(true) ? 0 : 1);
		
		
		
		/**====================================================================***/
		
		
		
		
		
		
		/**
		 * 
		 * 主的控制容器，控制上面的总的两个子作业
		 * 
		 * **/
		JobControl jobCtrl=new JobControl("myctrl");
		//ctrljob1.addDependingJob(ctrljob2);// job2在job1完成后，才可以启动
		//添加到总的JobControl里，进行控制
		
		jobCtrl.addJob(ctrljob1); 
		jobCtrl.addJob(ctrljob2);
		
 
		//在线程启动
		Thread  t=new Thread(jobCtrl);
		t.start();
		
		while(true){
			
			if(jobCtrl.allFinished()){//如果作业成功完成，就打印成功作业的信息
				System.out.println(jobCtrl.getSuccessfulJobList());
				
				jobCtrl.stop();
				break;
			}
			
			if(jobCtrl.getFailedJobList().size()>0){//如果作业失败，就打印失败作业的信息
				System.out.println(jobCtrl.getFailedJobList());
				
				jobCtrl.stop();
				break;
			}
			
		}
		
		
		
		
	 
		
		
		
		
		
		
		
		
		
		
		
		
		
		
 
		
		
	}
	
	
	
	

}

运行日志如下：

Java代码

模式： 192.168.75.130:9001
存在此输出路径，已删除！！！
存在此输出路径，已删除！！！
WARN - JobClient.copyAndConfigureFiles(746) | Use GenericOptionsParser for parsing the arguments. Applications should implement Tool for the same.
INFO - FileInputFormat.listStatus(237) | Total input paths to process : 1
WARN - NativeCodeLoader.<clinit>(52) | Unable to load native-hadoop library for your platform... using builtin-java classes where applicable
WARN - LoadSnappy.<clinit>(46) | Snappy native library not loaded
WARN - JobClient.copyAndConfigureFiles(746) | Use GenericOptionsParser for parsing the arguments. Applications should implement Tool for the same.
INFO - FileInputFormat.listStatus(237) | Total input paths to process : 1
[job name: Join1
job id: myctrl0
job state: SUCCESS
job mapred id: job_201405092039_0001
job message: just initialized
job has no depending job:
, job name: Join2
job id: myctrl1
job state: SUCCESS
job mapred id: job_201405092039_0002
job message: just initialized
job has 1 dependeng jobs:
depending job 0: Join1
]

模式：  192.168.75.130:9001
存在此输出路径，已删除！！！
存在此输出路径，已删除！！！
WARN - JobClient.copyAndConfigureFiles(746) | Use GenericOptionsParser for parsing the arguments. Applications should implement Tool for the same.
INFO - FileInputFormat.listStatus(237) | Total input paths to process : 1
WARN - NativeCodeLoader.<clinit>(52) | Unable to load native-hadoop library for your platform... using builtin-java classes where applicable
WARN - LoadSnappy.<clinit>(46) | Snappy native library not loaded
WARN - JobClient.copyAndConfigureFiles(746) | Use GenericOptionsParser for parsing the arguments. Applications should implement Tool for the same.
INFO - FileInputFormat.listStatus(237) | Total input paths to process : 1
[job name:	Join1
job id:	myctrl0
job state:	SUCCESS
job mapred id:	job_201405092039_0001
job message:	just initialized
job has no depending job:	
, job name:	Join2
job id:	myctrl1
job state:	SUCCESS
job mapred id:	job_201405092039_0002
job message:	just initialized
job has 1 dependeng jobs:
	 depending job 0:	Join1
]

处理的结果如下：

Java代码

三劫 932
b 579
秦东亮 206
a 45

三劫	932
b	579
秦东亮	206
a	45

可以看出，结果是正确的。程序运行成功，上面只是散仙测的2个MapReduce作业的组合，更多的组合其实和上面的一样。
总结：在配置多个作业时，Job的配置尽量分离单独写，不要轻易拷贝修改，这样很容易出错的，散仙在配置的时候，就是拷贝了一个，结果因为少修改了一个地方，在运行时候一直报错，最后才发现，原来少改了某个地方，这一点需要注意一下。

分享到：

外部CSS样式实现hover鼠标悬停改变样式 | AngularJS的select

2016-07-11 19:07
浏览 507
评论(0)
分类:企业架构
查看更多

发表评论

您还没有登录,请您登录后再发表评论

最近访客更多访客>>

博主相关

文章分类

社区版块

存档分类

最新评论

使用Hadoop里面的MapReduce来处理海量数据

评论

发表评论

相关推荐

最近访客 更多访客>>

博主相关

文章分类

社区版块

存档分类

最新评论

使用Hadoop里面的MapReduce来处理海量数据

评论

发表评论

相关推荐

CDH与原生态hadoop之间的区别

Cloudera的CDH和Apache的Hadoop的区别

大数据、云计算系统高级架构师课程学习路线图

Oozie简介

清理ambari安装的hadoop集群

hawk大数据基础知识总结（2）

hawk大数据基础知识总结（1）

ambari是什么

数据仓库中的Inmon与Kimball架构之争

Hive的meta 数据支持以下三种存储方式

大数据要学习知识

Spark Streaming简介

pentaho套件

Impala：新一代开源大数据分析引擎

Weka是什么

解密Apache HAWQ ——功能强大的SQL-on-Hadoop引擎

Kettle的使用

clouder manager端口7180没有打开为什么

Impala与Hive的比较

Cloudera Manager、CDH零基础入门、线路指导

最近访客更多访客>>