大橙子网站建设,新征程启航
为企业提供网站建设、域名注册、服务器等服务
今天就跟大家聊聊有关hadoop2.2.0如何定制mapreduce输出到数据库,可能很多人都不太了解,为了让大家更加了解,小编给大家总结了以下内容,希望大家根据这篇文章可以有所收获。
成都创新互联是专业的丰顺网站建设公司,丰顺接单;提供做网站、网站建设,网页设计,网站设计,建网站,PHP网站建设等专业做网站服务;采用PHP框架,可快速的进行丰顺网站开发网页制作和功能扩展;专业做搜索引擎喜爱的网站,专业的做网站团队,希望更多企业前来合作!
这里以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 LoginLogOutputFormatextends FileOutputFormat { /** * 重点也是定制一个RecordWriter类,每一条reduce处理后的记录,我们便可将该记录输出到数据库中 */ protected static class RedisRecordWriter extends RecordWriter { 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 getRecordWriter(TaskAttemptContext job) throws IOException, InterruptedException { Jedis jedis = RedisClient.newJedis(); //构建一个redis,这里你可以自己根据实际情况来构建数据库连接对象 //System.out.println("构建RedisRecordWriter"); return new RedisRecordWriter (jedis); } }
下面就是整个job实现:
public class LoginLogStatTask extends Configured implements Tool { public static class MyMapper extends Mapper{ @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 { @Override protected void reduce(Text key, Iterable 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 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:
看完上述内容,你们对hadoop2.2.0如何定制mapreduce输出到数据库有进一步的了解吗?如果还想了解更多知识或者相关内容,请关注创新互联行业资讯频道,感谢大家的支持。