这个很好的例子显示了on this blog,我尝试了它,它非常好.我引用了代码中最重要的部分.

首先,您必须创建一个表示您要存储的数据的类.该类必须实现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 {

protected void reduce(Text key, Iterable 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);

}

}

Logo

腾讯云面向开发者汇聚海量精品云计算使用和开发经验,营造开放的云计算技术生态圈。

更多推荐