今天就跟大家聊聊有关hadoop2.2.0如何定制mapreduce输出到数据库，可能很多人都不太了解，为了让大家更加了解，小编给大家总结了以下内容，希望大家根据这篇文章可以有所收获。
成都创新互联是专业的丰顺网站建设公司，丰顺接单;提供做网站、网站建设,网页设计,网站设计,建网站,PHP网站建设等专业做网站服务;采用PHP框架,可快速的进行丰顺网站开发网页制作和功能扩展;专业做搜索引擎喜爱的网站,专业的做网站团队,希望更多企业前来合作!
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 extends 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输出到数据库有进一步的了解吗？如果还想了解更多知识或者相关内容，请关注创新互联行业资讯频道，感谢大家的支持。
网站名称：hadoop2.2.0如何定制mapreduce输出到数据库
URL链接：
http://bjjierui.cn/article/gcgphg.html
