PostgreSQL

當兩隻大象結合的時候... 說實在的，這個標題我想很難讓人聯想到這意含為何.. 加上Goolge搜尋引擎對於網頁中的「title元素」所佔的權重又比較高... 嗯~ 所以本文似乎不容易被搜尋得到... 天曉得這是一篇探討Hadoop結合PostgreSQL的文章.. 不過我還是想這麼做...

以往要將資料庫中的資料抓出來當作MapReduce的輸入/輸出都必須先自行處理這當中的轉換工作，而本文要探討的是直接採用資料庫當作MapReduce的輸入/輸出資料，因為這在Hadoop 0.19版(目前為0.19.1)就納入支援了「MapReduce for MySQL(Hadoop-2536)」，底下是一個簡單的測試範例，下述是筆者自行建立的「wordcount」資料表：

CREATE TABLE wordcount
(
  id serial NOT NULL,
  word character varying(20) NOT NULL,
  count integer NOT NULL DEFAULT 1,
  CONSTRAINT wc_id PRIMARY KEY (id)
)
WITH (OIDS=FALSE);
ALTER TABLE wordcount OWNER TO postgres;

預設的資料內容如下：

基本上就是先透過DBConfiguration去設定資料庫相關的組態工作，然後交由DBInputFormat和DBOutputFormat來處理相對應資料表的輸入和輸出，並且撰寫一個實作DBWritable介面的Class，用來作為資料庫讀/寫工作的橋梁，在這個範例中為「WordRecord」Class，詳細請參考附檔。

P.S. 請拷貝一份「JDBC」driver放置在「HADOOP_HOME/lib」底下，另外您執行的jar檔也需要一同打包這個driver。

import java.io.IOException;
import java.util.Iterator;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapred.JobClient;
import org.apache.hadoop.mapred.JobConf;
import org.apache.hadoop.mapred.MapReduceBase;
import org.apache.hadoop.mapred.Mapper;
import org.apache.hadoop.mapred.OutputCollector;
import org.apache.hadoop.mapred.Reducer;
import org.apache.hadoop.mapred.Reporter;
import org.apache.hadoop.mapred.lib.db.DBConfiguration;
import org.apache.hadoop.mapred.lib.db.DBInputFormat;
import org.apache.hadoop.mapred.lib.db.DBOutputFormat;

import org.apache.hadoop.conf.Configured;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;

public class WordCount extends Configured implements Tool
{

    public int run(String[] arg0) throws Exception
    {
        JobConf job = new JobConf(getConf(), WordCount.class);

        job.setJobName("DBWordCount");

        job.setMapperClass(WordCountMapper.class);
        job.setReducerClass(WordCountReducer.class);

        DBConfiguration.configureDB(job, "org.postgresql.Driver", "jdbc:postgresql://localhost/WordCount", "帳號", "密碼");

        String[] fields = { "word", "count" };

        DBInputFormat.setInput(job, WordRecord.class, "wordcount", null, "id", fields);
        DBOutputFormat.setOutput(job, "wordcount(word,count)", fields);

        job.setMapOutputKeyClass(Text.class);
        job.setMapOutputValueClass(IntWritable.class);

        job.setOutputKeyClass(WordRecord.class);
        job.setOutputValueClass(NullWritable.class);

        JobClient.runJob(job);

        return 0;
    }

    static class WordCountMapper extends MapReduceBase implements
            Mapper<LongWritable, WordRecord, Text, IntWritable>
    {

        public void map(LongWritable key, WordRecord value,
                OutputCollector<Text, IntWritable> output, Reporter reporter) throws IOException
        {
            output.collect(new Text(value.word), new IntWritable(value.count));
        }
    }

    static class WordCountReducer extends MapReduceBase implements
            Reducer<Text, IntWritable, WordRecord, NullWritable>
    {

        public void reduce(Text key, Iterator<IntWritable> values,
                OutputCollector<WordRecord, NullWritable> output,
                Reporter reporter) throws IOException
        {
            int sum = 0;
            while (values.hasNext())
            {
                sum += values.next().get();
            }
            output.collect(new WordRecord(key.toString(), sum), NullWritable.get());
        }
    }

    public static void main(String args[]) throws Exception
    {
        int ret = ToolRunner.run(new WordCount(), args);
        System.exit(ret);
    }
}

結果：(直接寫回wordcount資料表)

詳細的內部實作可以參考DBInputFormat和DBOutputFormat，會發現DBInputFormat中的「getSelectQuery()」方法裡面用了select... order by、limit、offset去串起來這樣的SQL語法(所以目前尚不支援某些資料庫，如：Oracle)，相反的DBOutputFormat當然就是用insert into tablename values(fields name)，而在此範例中雖然有一個serial number當作Primary Key(id)，不過筆者所撰寫的「WordRecord」並沒有去操作這個ID，所以在「setOutput」的方法中筆者明確地告知資料表名稱為「wordcount(word,count)」，如此在輸出到資料表時才不會出錯。

．原始檔

參考資源

．Database Access with Hadoop

．DBInputFormat (Hadoop 0.19.1 API)

．2008-12-15 新增Apache Commons DBCP範例

大約六、七年前剛開始初學Web Application的時候~ 那時常聽到的是Tomcat、Resin~ 至於Jetty... 嗯~ 我聞不見其名...XD

Jetty．是一個100%以Java撰寫而成並開放源碼的HTTP Server & Servlet Container~ 從三年前的「Jetty vs. Tomcat vs. Resin: A Performance Comparison」這篇文章來看~ 它的效率似乎表現不錯~ 好了~ 回歸重點~ 為什麼我對它燃起了興趣？... 二個原因，其一，到目前Hadoop 0.19.0版所內建的Servlet Container就是採用Jetty 5.1.4~ 不過根據「(#HADOOP-1650) Upgrade Jetty to 6.x - ASF JIRA」這個issue來看~ 0.20版就會變成Jetty 6.x版了~

另一個原因，由於今年年初有協助老師將一個「English Collocations」的雛型系統改寫成Web版~ 而這樣的系統其實只需要「唯讀」資料庫的內容即可~ 想將整個這樣的系統都做成「DVD-ROM」版本~ 這樣就變成可以帶著跑的Web Application了~ 方便性大大的提高~ 所以Jetty相當適合這樣的應用!!

下面簡單地記錄一下一些常用的設定及步驟：

啟動Jetty Sever

java -jar start.jar etc/jetty.xml

手動增加一個新的Web Application (含VirualHost設定)

請修改「/etc/jetty.xml」。

<New class="org.mortbay.jetty.webapp.WebAppContext">
	<Arg><Ref id="Contexts"/></Arg>
	<Arg><SystemProperty name="jetty.home"/>/webapps/webapp</Arg>
	<Arg>/webapp</Arg>
	<Set name="defaultsDescriptor"><SystemProperty name="jetty.home" default="."/>/etc/webdefault.xml</Set>
	<Set name="VirtualHosts">
	<Array type="java.lang.String">
		<Item>localhost</Item>
	</Array>
	</Set>
</New>

測試Servlet - Hello World

import java.io.*;
import javax.servlet.*;
import javax.servlet.http.*;

public class Test extends HttpServlet
{
	public void doGet(HttpServletRequest req, HttpServletResponse res) throws ServletException, IOException
	{
		res.setContentType("text/html");
		PrintWriter out = res.getWriter();
		out.println("<html>");
		out.println("<head><title>Hello</title></head>");
		out.println("<body>");
		out.println("HIHI");
		out.println("</body>");
		out.println("</html>");
	}
}

修改「web.xml」

<servlet>
	<servlet-name>Hello</servlet-name>
	<servlet-class>Test</servlet-class>
</servlet>
<servlet-mapping>
	<servlet-name>Hello</servlet-name>
	<url-pattern>/Hello.do</url-pattern>
</servlet-mapping>

最後開啟「http://localhost:8080/webapp/Hello.do」即可。

PostgreSQL - ConnectionPoolDataSource

請修改「/etc/jetty.xml」。

<New id="DSTest" class="org.mortbay.jetty.plus.naming.Resource">
	<Arg>jdbc/DSTest</Arg>
	<Arg>
	<New class="org.postgresql.ds.PGConnectionPoolDataSource">
		<Set name="User">postgres</Set>
		<Set name="Password">xxx</Set>
		<Set name="DatabaseName">test</Set>
		<Set name="ServerName">localhost</Set>
		<Set name="PortNumber">5432</Set>
	</New>
	</Arg>
</New>

修改「web.xml」。

<resource-ref>
	<description>My DataSource Reference</description>
	<res-ref-name>jdbc/DSTest</res-ref-name>
	<res-type>javax.sql.DataSource</res-type>
	<res-auth>Container</res-auth>
</resource-ref>

一個簡單的測試程式如下：

import java.io.*;
import java.sql.*;
import javax.sql.*;
import javax.naming.*;
import javax.servlet.*;
import javax.servlet.http.*;

public class PostgresDBTest extends HttpServlet
{
	public void doGet (HttpServletRequest req, HttpServletResponse res) throws ServletException, IOException
	{
		res.setContentType("text/html");
		PrintWriter out = res.getWriter();
		
		Connection con = null;
		Statement stmt = null;
		ResultSet rs = null;
		
		try
		{
			ConnectionPoolDataSource source = (ConnectionPoolDataSource)new InitialContext().lookup("jdbc/DSTest");
    			con = source.getPooledConnection().getConnection();  
    			stmt = con.createStatement();
    			rs = stmt.executeQuery("select * from test");
    			while(rs.next())
    			{
    				out.println("Title:"+rs.getString(1)+"<br/>");
    			}
		} catch(Exception e){
    			e.printStackTrace();
		} finally {
    			if(con != null)
    			{
        			try {
       	 				con.close();
        			}catch(SQLException e)
        			{
        				e.printStackTrace();
        			}

    			}
		}		
	}	
}

修改「web.xml」。

<servlet>
	<servlet-name>PostgresDBTest</servlet-name>
	<servlet-class>PostgresDBTest</servlet-class>
</servlet>
<servlet-mapping>
	<servlet-name>PostgresDBTest</servlet-name>
	<url-pattern>/PostgresDBTest.do</url-pattern>
</servlet-mapping>

最後開啟「http://localhost:8080/webapp/PostgresDBTest.do」即可。

Apache Commons DBCP - PostgreSQL

下載下述這三個Library，並複製到「Jetty_Home/lib」底下。

commons-dbcp.jar

commons-pool.jar

commons-collections.jar

請修改「/etc/jetty.xml」。

<New id="pgsqldbcp" class="org.mortbay.jetty.plus.naming.Resource">
	<Arg>jdbc/Blog</Arg>
	<Arg>
	<New class="org.apache.commons.dbcp.BasicDataSource">
	<Set name="driverClassName">org.postgresql.Driver</Set>
	<Set name="url">jdbc:postgresql://localhost/Blog</Set>
	<Set name="username">postgres</Set>
	<Set name="password">1234</Set>
	<Set name="maxActive">10</Set>
	</New>
	</Arg>
</New>

修改「web.xml」。

<resource-ref>
	<description>My DataSource Reference</description>
	<res-ref-name>jdbc/DSTest</res-ref-name>
	<res-type>javax.sql.DataSource</res-type>
	<res-auth>Container</res-auth>
</resource-ref>

測試程式如下：

import java.io.*;
import java.sql.*;
import javax.sql.*;
import javax.naming.*;
import javax.servlet.*;
import javax.servlet.http.*;

public class PostgresDBTest extends HttpServlet
{
	public void doGet (HttpServletRequest req, HttpServletResponse res) throws ServletException, IOException
	{
		res.setContentType("text/html");
		PrintWriter out = res.getWriter();
		
		Connection con = null;
		Statement stmt = null;
		ResultSet rs = null;
		
		try
		{
			InitialContext ic = new InitialContext();
			DataSource source = (DataSource)ic.lookup("jdbc/Blog");
			con = source.getConnection();  
			stmt = con.createStatement();
			rs = stmt.executeQuery("select * from test");
			while(rs.next())
			{
    				out.println("Title:"+rs.getString(1)+"<br/>");
			}
		} catch(Exception e){
			e.printStackTrace();
		} finally {
			if(con != null)
			{
				try {
					con.close();
				}catch(SQLException e)
				{
					e.printStackTrace();
				}
			}
		}
	}		
}

自從上禮拜移機完成後~ 就沒有管理我這個Blog的資料了~

昨天想說來遠端試試~ 就在設定之中.... 呃... 事情發生了~

我key錯了指令... 導致資料庫Run不起來了~

今天早上才發覺事情大條了~

本來想到用「pg_resetxlog」來修復~ 結果還是不行...

無奈地重建資料庫~ 還好我上個禮拜有備份一次~

不然就欲哭無淚了...

所以有幾篇po文就此消失於雲煙之中~

也順勢地提醒我~ 要趕緊寫個shell script來自動地備份呀~ XD

當我們的程式和資料庫是位於不同的主機時~ 就必須透過TCP/IP來遠端連線資料庫~

或是你的PostgreSQL是裝在VMware的Linux之中，而你想透過Windows的pgAdmin III資料庫管理程式來操作的話~ 都需要做這設定來遠端連線~

允許遠端TCP/IP連線

修改「/usr/local/pgsql/data/pg_hba.conf」，並加入下列格式的設定值：

host    all     all     [ip位址]/[subnet mask]  trust

設定方式如下：

vi /usr/local/pgsql/data/pg_hba.conf
host    all     all     192.168.0.0/16  trust

接著修改「/usr/local/pgsql/data/postgresql.conf」

vi /usr/local/pgsql/data/postgresql.conf

找出「listen_addresses」和「port」，把這兩行的註解取消並修改成下列：

listen_addresses = '*'
port = 5432

最後重新啟動你的PostgreSQL即可。

在開始安裝PostgreSQL之前，請先安裝GNU Readline Library~ 因為編譯PostgreSQL時會用到這個函式庫~

apt-get install libreadline-dev

軟體取得

．PostgreSQL - postgresql-8.2.6.tar.gz

安裝步驟

先將「postgresql-8.2.6.tar.gz」解壓縮~ 然後進行一連串的編譯與安裝~

預設的安裝目錄在「/usr/local/pgsql」

tar zxvf postgresql-8.2.6.tar.gz
cd postgresql-8.2.6
./configure
make
make install

大致上這樣就安裝成功了~ 接著要進行一些設定及初始化資料庫的步驟~

設定步驟

在Linux新增一個名為「postgres」帳號~

adduser postgres

建立欲存放資料庫的目錄夾

mkdir /usr/local/pgsql/data
chown postgres /usr/local/pgsql/data
su - postgres

初始化資料庫並設定為UTF8編碼，如果你需要其它的編碼可參考Character Set Support

/usr/local/pgsql/bin/initdb -D /usr/local/pgsql/data -E UTF8

啟動PostgreSQL資料庫

/usr/local/pgsql/bin/postgres -D /usr/local/pgsql/data &

關閉PostgreSQL資料庫

/usr/local/pgsql/bin/pg_ctl stop -D /usr/local/pgsql/data