DataX插件開(kāi)發(fā)的指南

1、 .DOC資料. Linux公社（ HYPERLINK / LinuxIDC.com）于2006年9月25日注冊(cè)并開(kāi)通網(wǎng)站，Linux現(xiàn)在已經(jīng)成為一種廣受關(guān)注和支持的一種操作系統(tǒng)，IDC是互聯(lián)網(wǎng)數(shù)據(jù)中心，LinuxIDC就是關(guān)于Linux的數(shù)據(jù)中心。 HYPERLINK / LinuxIDC.com提供包括Ubuntu，F(xiàn)edora，SUSE技術(shù)，以及最新IT資訊等Linux專(zhuān)業(yè)類(lèi)網(wǎng)站。DataX插件開(kāi)發(fā)指南版本號(hào)修改內(nèi)容修改日期修改人V0.1創(chuàng)建2011-09-08何健超目錄 TOC o 1-3 h z u HYPERLINK l _Toc303275627 一、概述 PAGEREF _To

2、c303275627 h 4HYPERLINK l _Toc303275628 二、Reader插件開(kāi)發(fā)（以httpreader為例） PAGEREF _Toc303275628 h 5HYPERLINK l _Toc303275629 1、確定插件所需配置的參數(shù) PAGEREF _Toc303275629 h 5HYPERLINK l _Toc303275630 2、構(gòu)建相應(yīng)包和類(lèi)結(jié)構(gòu) PAGEREF _Toc303275630 h 5HYPERLINK l _Toc303275631 3、實(shí)現(xiàn)重載方法 PAGEREF _Toc303275631 h 7HYPERLINK l _Toc303

3、275632 4、自定義split方法 PAGEREF _Toc303275632 h 10HYPERLINK l _Toc303275633 三、Writer插件開(kāi)發(fā)（以streamwriter為例） PAGEREF _Toc303275633 h 13HYPERLINK l _Toc303275634 1、確定插件參數(shù)、構(gòu)建相應(yīng)包和類(lèi)結(jié)構(gòu) PAGEREF _Toc303275634 h 13HYPERLINK l _Toc303275635 2、實(shí)現(xiàn)重載方法 PAGEREF _Toc303275635 h 14HYPERLINK l _Toc303275636 四、插件運(yùn)行配置（以http

4、reader為例） PAGEREF _Toc303275636 h 16HYPERLINK l _Toc303275637 1、注冊(cè)插件 PAGEREF _Toc303275637 h 16HYPERLINK l _Toc303275638 2、修改build.xml文件，打包 PAGEREF _Toc303275638 h 16概述DataX是一個(gè)在不同類(lèi)型的數(shù)據(jù)庫(kù)(文件系統(tǒng))之間交換數(shù)據(jù)的工具，采用“框架+插件”的結(jié)構(gòu)，框架相當(dāng)于一個(gè)數(shù)據(jù)中轉(zhuǎn)平臺(tái)，而插件則為訪(fǎng)問(wèn)不同類(lèi)型的數(shù)據(jù)庫(kù)(文件系統(tǒng))提供實(shí)現(xiàn)。DataX插件分為Reader和Writer兩類(lèi)。Reader負(fù)責(zé)從數(shù)據(jù)源端讀取數(shù)據(jù)到Sto

5、rage(交換空間)，Writer負(fù)責(zé)將Storage中的數(shù)據(jù)寫(xiě)入到數(shù)據(jù)目的端。Storage可以適配不同種類(lèi)的Reader和Writer，從而實(shí)現(xiàn)數(shù)據(jù)同步。目前DataX版本已經(jīng)提供的Reader插件如下：hdfsreader : 支持從hdfs文件系統(tǒng)獲取數(shù)據(jù)。mysqlreader: 支持從mysql數(shù)據(jù)庫(kù)獲取數(shù)據(jù)。sqlserverreader: 支持從sqlserver數(shù)據(jù)庫(kù)獲取數(shù)據(jù)。oraclereader : 支持從oracle數(shù)據(jù)庫(kù)獲取數(shù)據(jù)。streamreader: 支持從stream流獲取數(shù)據(jù)（常用于測(cè)試）httpreader : 支持從http URL獲取數(shù)據(jù)。提供的Wr

6、iter插件如下：hdfswriter ：支持向hdbf寫(xiě)入數(shù)據(jù)。mysqlwriter ：支持向mysql寫(xiě)入數(shù)據(jù)。sqlserverwriter：支持向sqlserver寫(xiě)入數(shù)據(jù)。oraclewriter ：支持向oracle寫(xiě)入數(shù)據(jù)。streamwriter ：支持向stream流寫(xiě)入數(shù)據(jù)。（常用于測(cè)試）用戶(hù)可以根據(jù)需要開(kāi)發(fā)自己的Reader & Writer插件?，F(xiàn)在以HttpReader 和StreamWriter插件為例，使用eclipse分別說(shuō)明Reader和Writer插件開(kāi)發(fā)過(guò)程。Reader插件開(kāi)發(fā)（以httpreader為例）確定插件所需配置的參數(shù)確定插件參數(shù)，并在com

7、mon.plugin.ParamsKey.java中，創(chuàng)建靜態(tài)類(lèi)HttpReader，尤其注意對(duì)參數(shù)的注釋盡量參照源碼規(guī)范，DataX運(yùn)行時(shí)，會(huì)根據(jù)此處聲明的參數(shù)和注釋生成對(duì)應(yīng)的模板Job_xml.此處參數(shù)設(shè)置非常重要，如圖：圖 1構(gòu)建相應(yīng)包和類(lèi)結(jié)構(gòu)在源碼文件的plugins.reader包下構(gòu)建httpreader包，再在httpreader包下創(chuàng)建類(lèi)HttpReader，并讓之繼承common.plugin.Reader.圖 2圖 3實(shí)現(xiàn)重載方法獲得圖3所示效果：現(xiàn)在開(kāi)始分別實(shí)現(xiàn)init( ), connectToDb( ), startRead(LineSender sender), f

8、inish( ) 四個(gè)方法。Init( )：通過(guò)從Reader間接繼承自DefaultPlugin的PluginParam類(lèi)型的參數(shù)param獲取配置httpreader插件的參數(shù)(此處可以對(duì)參數(shù)進(jìn)行檢查和格式處理等操作)，如圖：圖 4connectToDb( ): 本插件不需要此操作，函數(shù)為空實(shí)現(xiàn)。（在數(shù)據(jù)庫(kù)相關(guān)插件中，主要操作是通過(guò)DbSource.getConnection(keyId)獲取connection），如在mysqlreader中，該函數(shù)為：圖 5startRead(LineSender sender)：根據(jù)init( )初始化的參數(shù)，連接相應(yīng)的http URL, 獲取其中數(shù)

9、據(jù)并用BufferedReader封裝，循環(huán)處理每行數(shù)據(jù)，調(diào)用sender.createLine() 產(chǎn)生line，并通過(guò)line.addField(fieldStr)把每行數(shù)據(jù)切分成字段后組裝成Line中的field,調(diào)用sender.sendToWriter(line)將此行數(shù)據(jù)寫(xiě)入Storage。如圖：圖 6finish()：本插件不需要此操作，函數(shù)實(shí)現(xiàn)為空。（在數(shù)據(jù)庫(kù)相關(guān)插件中，主要完成關(guān)閉connection操作。）如在mysqlreader中，該函數(shù)為：圖 7自定義split方法截止到目前，已經(jīng)實(shí)現(xiàn)了一個(gè)簡(jiǎn)單的httpreader（從一個(gè)http URL讀取數(shù)據(jù)），目前還未實(shí)現(xiàn)讀取

10、多個(gè)http URL的功能，為此，可以在common.plugin.ParamsKey.java中的靜態(tài)類(lèi)HttpReader中，添加如下兩項(xiàng)：圖 8然后讓httpreader插件覆蓋默認(rèn)的split(PluginParam param)方法。操作如下：split(PluginParam param)方法對(duì)param按照規(guī)則切分，生成List，DataX框架會(huì)根據(jù)此List產(chǎn)生插件實(shí)例執(zhí)行任務(wù)。圖 9為了清晰結(jié)構(gòu)或者切分規(guī)則復(fù)雜，建議再構(gòu)建類(lèi)HttpURLSplitter輔助split方法實(shí)現(xiàn)切分，該類(lèi)需要繼承Splitter：圖 10圖 11實(shí)現(xiàn)init( )方法和split( )方法即可：

11、圖 12Writer插件開(kāi)發(fā)（以streamwriter為例）確定插件參數(shù)、構(gòu)建相應(yīng)包和類(lèi)結(jié)構(gòu)與Reader插件類(lèi)似，略過(guò)，截圖如下：圖 1圖 2圖 3實(shí)現(xiàn)重載方法init( ) 、connectToDb()、finish( )三個(gè)方法與Reader中意義相似，此處略過(guò)。commint( ) ：本插件不需要此操作，函數(shù)實(shí)現(xiàn)為空。（在數(shù)據(jù)庫(kù)相關(guān)插件中，主要完成提交任務(wù)操作。）startWrite(LineReceiver linereceiver) ：根據(jù)init( )初始化的參數(shù)，按照streamwriter插件配置信息，把linereceiver從Storage獲取的數(shù)據(jù)通過(guò)xxx方法寫(xiě)入line中，循環(huán)處理每行數(shù)據(jù)（此處處理方式為直接輸出），即可。如圖：圖 4插件運(yùn)行配置（以httpreader為例）注冊(cè)插件在conf/plugins.xml中，添加如下配置項(xiàng)，以注冊(cè)插件：圖 1修改build.xml文件，打包圖 21. 若不給自己設(shè)限，則 HYPERLINK /jingdian/ganwu/ 人生中就沒(méi)有限制你發(fā)揮的藩籬。2. 若不是心寬似海，哪有人生風(fēng)平浪靜。在紛雜的塵世里，為自己留下一片純靜的心靈空間，不管是潮起潮落，也不管是陰晴圓缺，你都可以免去浮躁，義無(wú)反顧，勇往直前，輕松自如地走好人生路上的每一步3. 花一些時(shí)間，總會(huì)看清一些事。用一些事情，總會(huì)看