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

今天就跟大家聊聊有关 hadoop2.2.0 如何定制 mapreduce 输出到数据库，可能很多人都不太了解，为了让大家更加了解，丸趣 TV 小编给大家总结了以下内容，希望大家根据这篇文章可以有所收获。

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:

看完上述内容，你们对 hadoop2.2.0 如何定制 mapreduce 输出到数据库有进一步的了解吗？如果还想了解更多知识或者相关内容，请关注丸趣 TV 行业资讯频道，感谢大家的支持。