文档章节

hadoop2.2.0定制mapreduce输出到数据库

ihaolin
 ihaolin
发布于 2014/03/19 18:31
字数 666
阅读 1679
收藏 5

hadoop2.2.0定制mapreduce输出到数据库:

这里以redis数据库为例。

这里的例子是,我想统计日志文件中的某天各个小时的访问量,日志格式为:

2014-02-10 04:52:34 127.0.0.1 xxx

我们知道在写mapreduce job时,要配置输入输出,然后编写mapper和reducer类,hadoop默认输出是到hdfs的文件中,例如:

job.setOutputFormatClass(FileOutputFormat.class);

现在我们想要将任务计算结果输出到数据库(redis)中,怎么做呢?可以继承FileOutputFormat类,定制自己的类,看代码:

public class LoginLogOutputFormat<K, V> extends FileOutputFormat<K, V> {
	/**
	 * 重点也是定制一个RecordWriter类,每一条reduce处理后的记录,我们便可将该记录输出到数据库中
	 */
	protected static class RedisRecordWriter<K, V> extends RecordWriter<K, V>{
		private Jedis jedis; //redis的client实例
		
		public RedisRecordWriter(Jedis jedis){
			this.jedis = jedis;
		}
		
		@Override
		public void write(K key, V value) throws IOException,
				InterruptedException {
			
			boolean nullKey = key == null;
			boolean nullValue = value == null;
			if (nullKey || nullValue) return;
			
			String[] sKey = key.toString().split("-");
			String outKey = sKey[0]+"-"+sKey[1]+"-"+sKey[2]+"_login_stat"; //zset key为yyyy-MM-dd_login_stat
			jedis.zadd(outKey.getBytes("UTF-8"), -1, 
						(sKey[3]+":"+value).getBytes("UTF-8")); //zadd, 其值格式为: 时刻:访问量
		}

		@Override
		public void close(TaskAttemptContext context) throws IOException,
				InterruptedException {
			if (jedis != null) jedis.disconnect(); //关闭链接
		}
	}
	
	@Override
	public RecordWriter<K, V> getRecordWriter(TaskAttemptContext job)
			throws IOException, InterruptedException {
		Jedis jedis = RedisClient.newJedis(); //构建一个redis,这里你可以自己根据实际情况来构建数据库连接对象
		//System.out.println("构建RedisRecordWriter");
		return new RedisRecordWriter<K, V>(jedis);
	}
}
下面就是整个job实现:
public class LoginLogStatTask extends Configured implements Tool {
	public static class MyMapper extends Mapper<LongWritable, Text, Text, IntWritable>{
		@Override
		protected void map(LongWritable key, Text value, Context context)
				throws IOException, InterruptedException {
			if (value == null || "".equals(value)) return;
			// 解析value,如: 2014-02-10 04:52:34 127.0.0.1 xxx
			String[] fields = value.toString().split(" ");
			String date = fields[0];
			String time = fields[1];
			String hour = time.split(":")[0];
			String outKey = date+"-"+hour;
			context.write(new Text(outKey), new IntWritable(1));
		}
	}
	
	public static class MyReducer extends Reducer<Text, IntWritable, Text, IntWritable>{
		@Override
		protected void reduce(Text key, Iterable<IntWritable> values,
				Context context)
				throws IOException, InterruptedException {
			int count = 0;
			while (values.iterator().hasNext()){ //统计数量
				count ++;
				values.iterator().next(); 
			}
			context.write(key, new IntWritable(count));
		}
	}

	@Override
	public int run(String[] args) throws Exception {
		Configuration conf = getConf();
		List<Path> inputs = new ArrayList<>();
		String inputPath = args[0];
		if (inputPath.endsWith("/")){ //如果是目录
			inputs.addAll(HdfsUtil.listFiles(inputPath, conf));
		} else{ //如果是文件
			inputs.add(new Path(inputPath));
		}
		long ts = System.currentTimeMillis();
		String jobName = "login_logs_stat_job_" + ts;
		Job job = Job.getInstance(conf, jobName);
		job.setJarByClass(LoginLogStatTask.class);
		//添加输入文件路径
		for (Path p : inputs){
			FileInputFormat.addInputPath(job, p);
		}
		//设置输出路径
		Path out = new Path(jobName + ".out"); //以jobName.out作为输出
		FileOutputFormat.setOutputPath(job, out);
		//设置mapper
		job.setMapperClass(MyMapper.class);
		//设置reducer
		job.setReducerClass(MyReducer.class);
		
		//设置输入格式
		job.setInputFormatClass(TextInputFormat.class);
		//设置输出格式
		job.setOutputFormatClass(LoginLogOutputFormat.class);
		//设置输出key类型
		job.setOutputKeyClass(Text.class);
		//设置输出value类型
		job.setOutputValueClass(IntWritable.class);
		job.waitForCompletion(true);
		return job.isSuccessful()?0:1;
	}
	 
	public static void main(String[] args) throws Exception {
		Configuration conf = new Configuration();
		int res = ToolRunner.run(conf, new LoginLogStatTask(), args);
		System.exit(res);
	}

运行job后,就会在redis数据库中有对应的key:

不吝指正。

© 著作权归作者所有

ihaolin
粉丝 263
博文 164
码字总数 106524
作品 4
朝阳
高级程序员
私信 提问
hadoop 2.2.0 编译运行wordcount

hadoop2.2.0 编译运行wordcount,因为hadoop2.2.0不支持eclipse的插件,所以运行wordcount,需要手动编译并将wordcount打包成jar包来运行,下面记录一下编译运行的过程,希望能给大家有些帮助...

Adel
2014/09/05
693
1
国内最全最详细的hadoop2.2.0集群的MapReduce的最简单配置

简介 hadoop2的中的MapReduce不再是hadoop1中的结构已经没有了JobTracker,而是分解成ResourceManager和ApplicationMaster。这次大变革被称为MRv2或者YARN,是一次革命性的变化。 配置 在前面...

吴超沉思录
2014/01/26
277
0
mapreduce第三方jar

@iDestiny 你好,想跟你请教个问题:hadoop2.2.0,在mapreduce将结果写入hbase中,报错找不到HbaseConfigureation ,我试过在-libjars中添加相关jar和将相关的jar打包到我的jar中,都不能解决...

xp19901010
2014/03/31
353
0
MapReduce: 一个巨大的倒退

前言 databasecolumn 的数据库大牛们(其中包括PostgreSQL的最初伯克利领导:Michael Stonebraker)最近写了一篇评论当前如日中天的MapReduce技术的文章,引发剧烈的讨论。我抽空在这儿翻译一...

ddatsh
2011/11/04
4.6K
7
MapReduce 算法 - 反序模式 (Order Inversion)

这一篇其它段落的一系列MapReduce算法在" Data-Intensive Text Processing with MapReduce"这本书上呈现。以前分别是 Local Aggregation , Local Aggregation PartII 和 Creating a Co-Occu......

可观
2013/01/25
3.1K
5

没有更多内容

加载失败,请刷新页面

加载更多

Phpstorm2018 永久激活

1、安装phpstorm,安装包请自行官网下载 http://www.jetbrains.com/phpstorm/download/ 2、下载JetbrainsCrack.jar文件,存放至你的phpstorm执行文件同级目录下 下载JetbrainsCrack.jar 提取...

happyfish319
38分钟前
7
0
谈一谈Android进程间通信的几种方式

###来看一下Android中除了AIDL还有哪些进程间通信的方式: 1、Bundle Bundle实现了Parcelable,所以在Android中我们可以通过Intent在不同进程间传递Bundle数据。 但是在Intent 传输数据的过程...

二营长的意大利炮手
38分钟前
7
0
互联网薪资“高开低走”,你的能力是否真的可以匹配高薪?

对于国内外主流互联网大厂,技术出身似乎已经成为各大掌门人的必备标签。谷歌 CEO 桑达尔·皮查伊、马克·扎克伯格、李彦宏、马化腾、雷军等等皆为技术人出身,都曾参与了公司内部重要产品的...

Java技术剑
40分钟前
11
0
java 多线程

线程声明周期 线程的五个状态:新建,就绪,运行,阻塞,死亡。 其中就绪和运行两个状态客户互相转换,但运行到阻塞,阻塞到就绪,只能单向转换。 刚new出的线程就是【新建】状态,调用start...

雷开你的门
42分钟前
14
0
构造器Constructor是否可被overrid

构造器不能被重写,不能用static修饰构造器,只能用public private protected这三个权限修饰符,且不能有返回语句。

无名氏的程序员
46分钟前
7
0

没有更多内容

加载失败,请刷新页面

加载更多

返回顶部
顶部