将Apache Hadoop数据输出存储到Mysql数据库

Question

我需要将map reduce程序的输出存储到数据库中，所以有什么方法吗？

如果是这样的话，是否可以将输出存储到多列&基于需求的表格？？

请给我一些解决办法。

谢谢。。

Answer 1

这个博客上展示了一个很棒的例子，我试过了，效果非常好。我引用代码中最重要的部分。

首先，必须创建一个表示要存储的数据的类。该类必须实现DBWritable接口：

public class DBOutputWritable implements Writable, DBWritable
{
   private String name;
   private int count;
   public DBOutputWritable(String name, int count) {
     this.name = name;
     this.count = count;
   }
   public void readFields(DataInput in) throws IOException {   }
   public void readFields(ResultSet rs) throws SQLException {
     name = rs.getString(1);
     count = rs.getInt(2);
   }
   public void write(DataOutput out) throws IOException {    }
   public void write(PreparedStatement ps) throws SQLException {
     ps.setString(1, name);
     ps.setInt(2, count);
   }
}

在Reducer:中创建以前定义的类的对象

public class Reduce extends Reducer<Text, IntWritable, DBOutputWritable, NullWritable> {
   protected void reduce(Text key, Iterable<IntWritable> values, Context ctx) {
     int sum = 0;
     for(IntWritable value : values) {
       sum += value.get();
     }
     try {
       ctx.write(new DBOutputWritable(key.toString(), sum), NullWritable.get());
     } catch(IOException e) {
       e.printStackTrace();
     } catch(InterruptedException e) {
       e.printStackTrace();
     }
   }
}

最后，您必须配置到数据库的连接（不要忘记在类路径上添加数据库连接器），并注册映射器和reducer的输入/输出数据类型。

public class Main
{
   public static void main(String[] args) throws Exception
   {
     Configuration conf = new Configuration();
     DBConfiguration.configureDB(conf,
     "com.mysql.jdbc.Driver",   // driver class
     "jdbc:mysql://localhost:3306/testDb", // db url
     "user",    // username
     "password"); //password
     Job job = new Job(conf);
     job.setJarByClass(Main.class);
     job.setMapperClass(Map.class); // your mapper - not shown in this example
     job.setReducerClass(Reduce.class);
     job.setMapOutputKeyClass(Text.class); // your mapper - not shown in this example
     job.setMapOutputValueClass(IntWritable.class); // your mapper - not shown in this example
     job.setOutputKeyClass(DBOutputWritable.class); // reducer's KEYOUT
     job.setOutputValueClass(NullWritable.class);   // reducer's VALUEOUT
     job.setInputFormatClass(...);
     job.setOutputFormatClass(DBOutputFormat.class);
     DBInputFormat.setInput(...);
     DBOutputFormat.setOutput(
     job,
     "output",    // output table name
     new String[] { "name", "count" }   //table columns
     );
     System.exit(job.waitForCompletion(true) ? 0 : 1);
   }
}