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的個數，以便將並行鏈接的數量控制在合理的範圍以內。