Hadoop DBOutputFormat的使用

最近在研究数据在HDFS和关系型数据库之间的迁移，主要使用了两种方式：一是，按照数据库要求的文件格式生成文件，而后由数据库提供的导入工具进行导入；二是采用JDBC的方式进行导入。MapReduce默认提供了DBInputFormat和DBOutputFormat，分别用于数据库的读取和数据库的写入。为了使用DBOutputFormat咱们须要完成如下工做。

首先，对于每个数据库表编写对应的bean类，该类须要实现WritableComparable接口和DBWritable接口（若是是DBInputFormat，则须要实现Writable和DBWritable。之因此是这样是由于DBOutputFormat在输出的时候是将key写入到数据库而不是value。根据Hadoop的规定Key须要可以进行比较排序，因此须要实现WritableComparable）。Bean类的实现如下面的代码为例：

　　public void readFields(ResultSet result) throws SQLException {
　　		int index = 1;
　　		this.setTestId(result.getString(index++));
　　		this.setTestName(result.getString(index++));
　　		this.setAge(result.getInt(index++));
　　	}
　　	public void write(PreparedStatement statement) throws SQLException {
　　		int index = 1;
　　		statement.setString(index++, this.getTestId());
　　		statement.setString(index++, this.getTestName());
　　		statement.setInt(index, this.getAge());
　　
　　	}

上面两个方法对应着DBWriteable接口。readFields方法负责从结果集中读取数据库数据（注意ResultSet的下标是从1开始的），一次读取查询SQL中筛选的某一列。Write方法负责将数据写入到数据库，将每一行的每一列依次写入。

完成bean的定义后，进行Mapper的编写，主要是解析数据库的每一行数据而后将每一列赋值给bean对应的属性，这里再也不作详细的介绍。

最后进行Job的一些配置，具体以下面代码所示：

　　Configuration conf = new Configuration();
　　conf.set(DBConfiguration.DRIVER_CLASS_PROPERTY, 			                                                       "com.mysql.jdbc.Driver");
　　conf.set(DBConfiguration.URL_PROPERTY,
　　				"jdbc:mysql://localhost:3306/htestdb");
　　conf.set(DBConfiguration.USERNAME_PROPERTY, "root");
　　conf.set(DBConfiguration.PASSWORD_PROPERTY, "");
　　job.setNumReduceTasks(0);
　　DBOutputFormat.setOutput(job, "test", "testid","testname","age");
　　job.setOutputFormatClass(DBOutputFormat.class);

上面的配置主要包括如下几项：

l 数据库驱动的名称：com.mysql.jdbc.Driver

l 数据库URL：jdbc:mysql://localhost:3306/htestdb

l 用户名：root

l 密码：空

l 数据库表以及每列的名称：DBOutputFormat.setOutput(job, "test", "testid","testname","age")

除此以外还有Hadoop基础设置，好比reduce的个数、输入输出方式、输入输出路径等，这里再也不作详细介绍。

须要提醒的是DBOutputFormat以MapReduce的方式运行，会并行的链接数据库。在这里须要合适的设置map活着reduce的个数，以便将并行链接的数量控制在合理的范围以内。