将Apache Hadoop数据输出存储到MySQL数据库中

Question

我需要将Map-Reduce程序的输出存储到数据库中，是否有方法可以实现？

如果可以，是否可以根据需求将输出存储到多个列和表中？

请给我一些解决方案。

谢谢。

- HarshaKP

---

看这个：http://blog.cloudera.com/blog/2009/03/database-access-with-hadoop/ 特别是“将结果写回数据库”一节。 - Amar

您可以使用 Sqoop 来进行数据的传输。 - Suvarna Pattayil

感谢Amar和SuvP提供的链接。 - HarshaKP

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();
     }
   }
}

最后，您必须配置与您的数据库的连接（不要忘记将您的数据库连接器添加到类路径中），并注册您的Mapper和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);
   }
}

- Michal