Articles

她．從小帶著我長大... 換個說法，也只有我讓她帶著長大~

小時候只要她一句話，我從來不會敢有任何過多的要求~ 那時候的我乖的像隻忠狗~

當我過的不是很好的時候，她... 曾塞了一千元到我的口袋~

當我不懂事的時候，她... 從來沒有放棄過我~

當我生日的時候，她... 會準備火鍋等我回去幫我慶生~

另外每次過年的時候，回去鹿港幫忙打掃貼春聯一定有我的份! 不過以後很難再有了~

她．總是希望我多賺點錢... 但我總是回話給她說~ 我是要來做事的，不是要來賺錢的~ (也許這是她的時代環境背景下的影響)

她．已經不記得我是誰了... 也不會說話了... 現在只會對著人微笑...

她．是我的阿嬤。

Laconica - The Open Microblogging Tool．是一套免費且開放原始碼(PHP)的微型部落格平臺，可以用在一個社群或群組的短訊(140 character)溝通平臺，類似於當下熱門的Twitter、Plurk。

如果您是Twitter的愛用者，對於「@」符號應該不會太陌生，下述是從Laconica的原始碼抓出它的Regular Expression來看:

import re
msg = "@account This is a test."
p = r'(?:^|\s)@([a-z0-9]{1,64})'
m = re.findall(p,msg)  
for i in m:
    print i

從這個簡單的小程式來看，重點在於「(?:^|\s)@([a-z0-9]{1,64})」這一行Regular Expression，它處理的方式是在「@」之後的「帳號名稱」只能是小寫的字母和數字所組成，且範圍限制在1到64個字元之間，而在「@」之前的處理方式則是必須是訊息的起始位置或是空白字元，不過因為它是用「(?:^|\s)」括弧涵蓋起來，所以加上「?:」來排除這個Group，如此就是一個類似處理Twitter訊息輸入欄的Regular Expression。

下述這個範例就是一個簡單的利用「#」符號來擴展成標記Tag的方法：

import re
msg = "Today is a sunny day. #weather #life"
p = r'(?:^|\s)#([A-Za-z0-9_\-\.]{1,64})'
m = re.findall(p,msg)  
for i in m:
    print i

參考資源

．8.2. re — Regular expression operations

．How @replies work on Twitter (and how they might)

writeAhead．是一個論文摘要輔助寫作工具，它俱備11種不同學科領域的參考字詞、片語和轉折語，而此系統最大的特色在於當您在文字編輯區進行摘要的撰寫工作時，每當按下「空白鍵」之後，系統會自動地取出您剛所輸入的前兩個單字作為查詢參考片語的依據。

另外系統也提供支援「*(wildcard)」的功能，例如：在文字編輯區中輸入「* algorithm」按下空白鍵後，系統便會自動幫您找出參考片語中最後一個單字為「algorithm」的參考片語。

如果您有任何的建議，也非常歡迎您在此留言。

writeAhead - 簡易說明

您可以點選「電腦科學(預設)」此連結來切換不同學科領域的參考資訊，另外出現在括弧內的「this paper」兩個單字，是此系統會依賴您所輸入的前兩個單字作為查詢參考片語的依據，提示您目前出現在參考片語中是依賴那兩個單字。

此字詞功能主要提供類似Google Suggest的「補字」功能，例如：當您在文字編輯區輸入「pa」兩個單字之後，大約在0.5秒後系統會自動提示「補字」的功能，您可以直接點選「字詞功能」所列出的單字，它會幫您補齊先前所輸入「pa」開頭的單字。

參考片語功能主要根據您所輸入的前兩個單字作為依據，來自動分析出連接著出現較高機率的參考片語。

另外此系統也針對不同學科領域提供不同的轉折語資訊，您可以直接點選加到文字編輯區中。

當兩隻大象結合的時候... 說實在的，這個標題我想很難讓人聯想到這意含為何.. 加上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)

當大象遇上PDFBox... 這個標題看起來蠻有趣的，相反的當「Hadoop + PDFBox」就太正式了.. XD

兩個月前筆者曾po「PDFBox - 擷取PDF檔案中的純文字」，現在一樣請多台幾器一起來做這件事~ 如果沒機會體驗的話~ 看看「Self-service, Prorated Super Computing Fun!」這篇描述NYT在兩年前用Hadoop將1100萬份文章的TIFF影像檔轉成PDF檔案，重點在於只花了一天的時間就搞定了... = =" 而本文要做的就是分散式的將這些PDF檔案擷取出純文字~ 當然會比一台機器快多了~ (不過話說我也是在一台機器上測試...)

P.S. third party library 請記得放在「lib」資料夾一同打包

import java.io.ByteArrayInputStream;
import java.io.IOException;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.conf.Configured;
import org.apache.hadoop.fs.FSDataOutputStream;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.BytesWritable;
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.Reporter;
import org.apache.hadoop.mapred.lib.NullOutputFormat;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;
import org.pdfbox.pdmodel.PDDocument;
import org.pdfbox.util.PDFTextStripper;

public class PDF2TXT extends Configured implements Tool
{

    public static class Map extends MapReduceBase implements
            Mapper<NullWritable, BytesWritable, Text, Text>
    {

        private JobConf conf;

        @Override
        public void configure(JobConf conf)
        {
            this.conf = conf;
        }

        public void map(NullWritable key, BytesWritable value,
                OutputCollector<Text, Text> output, Reporter reporter)
                throws IOException
        {
            String filename = conf.get("map.input.file");
            String output_dir = conf.get("output.dir");
            filename = getFileName(filename);

            FileSystem fs = FileSystem.get(conf);
            FSDataOutputStream dos = fs.create(new Path(output_dir + filename + ".txt"));
            PDDocument document = PDDocument.load(new ByteArrayInputStream(value.getBytes()));
            PDFTextStripper stripper = new PDFTextStripper();
            stripper.setStartPage(1);
            stripper.setEndPage(document.getNumberOfPages());
            String s = stripper.getText(document);
            dos.write(s.getBytes("UTF-8"));
            dos.close();
        }

        public String getFileName(String s)
        {
            return s.substring(s.lastIndexOf("/"), s.lastIndexOf("."));
        }
    }

    public int run(String[] args) throws Exception
    {
        JobConf conf = new JobConf(getConf(), PDF2TXT.class);
        conf.set("output.dir", args[1]);

        conf.setJobName("PDF2TXT");
        conf.setMapperClass(Map.class);

        conf.setInputFormat(WholeFileInputFormat.class);
        conf.setOutputFormat(NullOutputFormat.class);

        conf.set("mapred.child.java.opts", "-Xmx256m");
        conf.setNumReduceTasks(0);

        WholeFileInputFormat.setInputPaths(conf, new Path(args[0]));
        JobClient.runJob(conf);
        return 0;
    }

    public static void main(String[] args)
    {
        try
        {
            int res = ToolRunner.run(new Configuration(), new PDF2TXT(), args);
            System.exit(res);
        } catch (Exception e)
        {
            e.printStackTrace();
        }
    }
}

．原始檔