java遠程調(diào)用kettle說明與代碼

頁Kettle遠程任務(wù)執(zhí)行

遠程服務(wù)器KETTLE提供了名為carte的webserver程序，也叫slaveserver,啟動該程序可以把主機作為kettle的運行服務(wù)器，可以接收其他kettle客戶端發(fā)送過來的ETL任務(wù)。此功能為ETL任務(wù)分布式執(zhí)行提供了方便?？梢栽诙鄠€服務(wù)器上運行cartewebserver,在調(diào)用不同的ETL任務(wù)時，指定不同的目標服務(wù)器執(zhí)行。kettle提供了carte.bat和carte.sh（linux）批處理腳本來啟動子服務(wù)器，這種啟動方式分為兩種使用主機號和端口號Carte8080Carte218081使用配置文件Carte/foo/bar/carte-config.xmlCarte/carte-config.xml如果clusterschema中定義了Dynamiccluster選項，則必須使用配置文件來進行啟動，當這個子服務(wù)器啟動時，它需要向配置文件中“masters”中列出的主服務(wù)器列表中匯報其運行狀態(tài)（通過調(diào)用主服務(wù)器的registerSlave服務(wù)），已達到動態(tài)地設(shè)置子服務(wù)器的目的。配置文件格式<slave_config>

<masters>

<slaveserver>

<name>master1</name>AllocateServerSocketServlet：分配一個新的socket端口號。這個端口號是基于你在定義clusterschema中設(shè)置的端口號，依次加1StartJobServlet：執(zhí)行服務(wù)器上某個指定的作業(yè)StopJobServlet:停止正在運行的作業(yè)GetJobStatusServlet：獲得某個指定作業(yè)的狀態(tài)AddJobServlet：向當前的子服務(wù)器中添加某個作業(yè)。RegisterSlaveServlet：注冊某個服務(wù)器的信息。服務(wù)器信息包括子服務(wù)器是否活動、最新活動的時間、最新不活動的時間。這個在dynamiccluster中需要用到，由從屬服務(wù)器向主服務(wù)器匯報當前狀態(tài)。GetSlavesServlet：獲得集群中子服務(wù)器的信息AddExportServlet：以zip文件的形式向caret服務(wù)器傳遞作業(yè)或者轉(zhuǎn)換信息，并將信息加入到服務(wù)器中。運行任務(wù)在kettle的集成設(shè)計環(huán)境spoon中，你可以選擇轉(zhuǎn)換中的“運行”菜單項，或者按F9快捷鍵，彈出以下的窗口這里有三個選項來決定轉(zhuǎn)換是以什么方式來執(zhí)行本地執(zhí)行:轉(zhuǎn)換或者作業(yè)將在你現(xiàn)在使用的JVM中運行。遠程執(zhí)行:允許你指定一個想運行轉(zhuǎn)換的遠程服務(wù)器。這需要你在遠程服務(wù)器上安裝PentahoDataIntegration（Kettle）并且運行Carte子服務(wù)器。集群方式執(zhí)行:允許你在集群環(huán)境下執(zhí)行作業(yè)或者轉(zhuǎn)換 jobMeta=repository.loadJob(jobName,directory,null,null); JobExecutionConfigurationjobExecutionConfiguration=newJobExecutionConfiguration(); jobExecutionConfiguration.setRemoteServer(remoteSlaveServer); jobExecutionConfiguration.setRepository(repository);代碼中需要定義slaveserver，然后在任務(wù)執(zhí)行配置中設(shè)置slaveserver，然后把任務(wù)發(fā)送給slaveserver程序?qū)崿F(xiàn)首先，需要在項目中引入執(zhí)行kettle所需要的jar包：實際應(yīng)該引用更多的jar包，包括lib/中的所有包Libext/commons中的所有包等。調(diào)用的源碼如下其中在8服務(wù)器上運行carte.bat88765命令來啟動slaveserver其中調(diào)用方法如下 publicstaticvoidmain(String[]args){//設(shè)置資源庫信息 OracleRepositoryInforposInfo=newOracleRepositoryInfo(); rposInfo.setDbHostname("07"); rposInfo.setDbName("ora1"); rposInfo.setDbPort("1521"); rposInfo.setDbType("Oracle"); rposInfo.setDbUsername("kettle"); rposInfo.setDbPassword("kettle"); rposInfo.setRepoId("admin"); rposInfo.setRepoName("admin"); rposInfo.setRepoPassword("admin"); rposInfo.setRepoUsername("admin");//新建job對象，初始化資源庫 CallJobctf=newCallJob(rposInfo);//設(shè)置slaveserver信息 SlaveServerInfossi=newSlaveServerInfo(); ssi.setServerHost("8"); ssi.setServerPort("8765"); ssi.setServerName("test"); ssi.setServerUsername("cluster"); ssi.setServerPassword("cluster");//為job設(shè)置slaveserve ctf.initSlaveServer(ssi); try{ //ctf.executeJobLocal("WF_DD_T_AP_DA_RTU_DAY_STATS_QUANTITY"); //調(diào)用遠程執(zhí)行job ctf.executeJobRem