java整合kettle部署任务到远程服务器
由于kettle服务器时不时断电重启，部署的那些任务需要手动一个一个点比较麻烦，所以希望能通过代码一把部署多个任务或转换到服务器上。导入了相关jar包，但有些和项目原有的冲突，然后一点点试过来导入了一部分。
在本地运行任务和转换的能找到很多参考代码，然后部署到远程服务器上的相较少一些，找了许久，单独部署转换到服务器上的时候测试成功，但是任务中包含转换的就不行，会一直报找不到转换，无论我把转换的路径写成绝对路径还是${Internal.Job.Filename.Directory}这种都不行，我想着在部署job之前把trans先部署上去但是这样也不行... 好多相关文档都是下载要币的
然后我就下载了源码，首先在kettle中遇到这种问题，是因为运行环境的send resources to this server没有打钩，我就照着这种思路去找源码中的这个参数。
我觉得这个参数肯定是个boolean，首先搜索send resource等找到了疑似，然后想是不是有什么RunConfiguration的类需要在部署到服务器上之前进行设置，但找了半天没找到
后来又去搜DefaultRunConfiguration这个类，最终在DefaultRunConfigurationExecutor.java中找到了一行设置这个类的sendresource变量的。。。而且这个JobExecutionConfiguration刚好是部署到服务器上市用到的配置类，这才有了下面代码的180行。成功了~啊~解决了真的好开心哦~
1 package org.ssh.hip.job.util; 2 3 import org.pentaho.di.cluster.SlaveServer; 4 import org.pentaho.di.core.KettleEnvironment; 5 import org.pentaho.di.core.database.DatabaseMeta; 6 import org.pentaho.di.core.exception.KettleException; 7 import org.pentaho.di.job.Job; 8 import org.pentaho.di.job.JobExecutionConfiguration; 9 import org.pentaho.di.job.JobMeta; 10 import org.pentaho.di.repository.ObjectId; 11 import org.pentaho.di.repository.RepositoryDirectoryInterface; 12 import org.pentaho.di.repository.kdr.KettleDatabaseRepository; 13 import org.pentaho.di.repository.kdr.KettleDatabaseRepositoryMeta; 14 import org.pentaho.di.trans.Trans; 15 import org.pentaho.di.trans.TransExecutionConfiguration; 16 import org.pentaho.di.trans.TransMeta; 17 import org.slf4j.Logger; 18 import org.slf4j.LoggerFactory; 19 20 public class KettleEnvironments2 { 21 22 static final Logger logger = LoggerFactory.getLogger(KettleEnvironments2.class); 23 24 25 public static KettleDatabaseRepository repository; 26 public static DatabaseMeta databaseMeta; 27 public static RepositoryDirectoryInterface directory; 28 public static SlaveServer remoteServer; 29 30 /** 31 * 配置资源库环境 并接连接的资源库 32 * @return 33 * 34 * @return 35 * @throws KettleException 36 */ 37 public static void repositoryCon(String repositoryName,String repositoryType,String repositoryConnectType,String repositoryIp,String repositoryDBName,String repositoryDBPort,String 38 repositoryDBUsername,String repositoryDBPassword, String repositoryUsername, String repositoryPassword) throws KettleException { 39 // 初始化 40 KettleEnvironment.init(); 41 // 数据库连接元对象 42 // （kettle数据库连接名称(KETTLE工具右上角显示)，资源库类型，连接方式，IP，数据库名，端口，用户名，密码） //cgmRepositoryConn 43 // databaseMeta = new DatabaseMeta("ETL", "Oracle", "Native", "192.168.1.28","orcl", "1521", 44 // "etl", "etl");//资源库数据库地址，我这里采用oracle数据库 45 databaseMeta = new DatabaseMeta(repositoryName, repositoryType, repositoryConnectType, repositoryIp, repositoryDBName, repositoryDBPort, 46 repositoryDBUsername, repositoryDBPassword);//资源库数据库地址，我这里采用oracle数据库 47 // 数据库形式的资源库元对象 48 KettleDatabaseRepositoryMeta kettleDatabaseMeta = new KettleDatabaseRepositoryMeta(); 49 kettleDatabaseMeta.setConnection(databaseMeta); 50 // 数据库形式的资源库对象 51 repository = new KettleDatabaseRepository(); 52 // 用资源库元对象初始化资源库对象 53 repository.init(kettleDatabaseMeta); 54 // 连接到资源库 55 repository.connect(repositoryUsername, repositoryPassword);// 默认的连接资源库的用户名和密码 56 directory = repository.loadRepositoryDirectoryTree(); 57 if (repository.isConnected()) { 58 System.out.println("连接成功"); 59 } else { 60 System.out.println("连接失败"); 61 } 62 } 63 64 /** 65 * 加载子服务器 66 */ 67 public static void initRemoteServer(String serverHostname, String serverPort, 68 String serverName, String serverUsername, String serverPassword, boolean serverMaster) { 69 remoteServer = new SlaveServer(); 70 remoteServer.setHostname(serverHostname); 71 remoteServer.setPort(serverPort); 72 remoteServer.setName(serverName); 73 remoteServer.setUsername(serverUsername); 74 remoteServer.setPassword(serverPassword); 75 remoteServer.setMaster(serverMaster); // 是否是主服务器 76 } 77 78 /** 79 * 调用Transformation示例 80 * 81 * @param rep 82 * @param transName 83 */ 84 public static void runTrans(String path, String transName) { 85 86 try { 87 // 根据指定的字符串路径 找到目录，在 r_transformation中id_directory=0的代表根目录 88 RepositoryDirectoryInterface dir = repository.findDirectory(path); 89 // 根据转换名称获取转换id 90 // 这个值就是从SELECT ID_TRANSFORMATION FROM `r_transformation` where name='fromto';来的 91 ObjectId id = repository.getTransformationID(transName, dir); 92 System.out.println("------ObjectId=" + id); 93 // 根据转换id查询与其关联的step和hop 94 // select name from r_step where ID_STEP=(SELECT ID_STEP_FROM FROM `r_trans_hop` where 95 // ID_TRANSFORMATION='1'); 96 // select name from r_step where ID_STEP=(SELECT ID_STEP_TO FROM `r_trans_hop` where ID_TRANSFORMATION='1'); 97 TransMeta transMeta = repository.loadTransformation(id, null); 98 // 99 Trans trans = new Trans(transMeta); 100 trans.execute(null);// 执行转换 101 trans.waitUntilFinished(); // 等待转换执行结束 102 if (trans.getErrors() != 0) { 103 System.out.println("trans executed failed"); 104 } else { 105 System.out.println("trans executed OK"); 106 } 107 } catch (Exception e) { 108 e.printStackTrace(); 109 } 110 } 111 112 113 /** 114 * 本地运行job 115 * @param path 116 * @param jobName 117 * @throws KettleException 118 */ 119 public static void runJob(String path, String jobName) throws KettleException { 120 // jobname 是Job脚本的路径及名称 121 // new JobMeta(jobName, null); // new出来的路径变成了D盘的 122 // new JobMeta(jobName, repository); // new出来的路径变成了D:\workspace\ekhip_git\sshapp-ekhips\sshapp-ekhips-auth\sshapp-hip-app\sshapp-hip-job\ 123 RepositoryDirectoryInterface dir = repository.findDirectory(path); 124 125 // 方法1 126 JobMeta jobMeta = repository.loadJob(jobName, dir, null, null); 127 Job job = new Job(repository, jobMeta); 128 129 // 方法2 130 // ObjectId id = repository.getJobId(jobName, dir); 131 // JobMeta jobMeta = repository.loadJob(id, null); 132 // Job job = new Job(repository, jobMeta); 133 134 // 向Job 脚本传递参数，脚本中获取参数值：${参数名} 135 // job.setVariable(paraname, paravalue); 136 137 // try { 138 // SlaveServerConfig config = new SlaveServerConfig("192.168.1.208", 9000, true);//ip与端口与通过Carte.bat启动时指定的参数一样 139 // Carte.runCarte(config); 140 // } catch (Exception e) { // 通过该静态方法启动carte 141 // e.printStackTrace(); 142 // } 143 144 job.start(); 145 // job.waitUntilFinished(); // 这个是等待job执行完，但是有的任务是定时的重复的 146 if (job.getErrors() > 0) { 147 System.out.println("job executed failed"); 148 } else { 149 System.out.println("job executed OK"); 150 } 151 } 152 153 /** 154 * 远程服务器上执行job 155 * @param dir 156 * @param jobName 157 * @throws KettleException 158 */ 159 public static void executeJobRemote(String path, String jobName) throws KettleException { 160 161 RepositoryDirectoryInterface directory = repository.loadRepositoryDirectoryTree(); // Default 162 163 RepositoryDirectoryInterface dir = repository.findDirectory(path); 164 165 // JobMeta jobMeta = repository.loadJob(jobName, jobdir, null, null); // reads 166 167 // 方法2 168 ObjectId id = repository.getJobId(jobName, dir); 169 JobMeta jobMeta = repository.loadJob(id, null); 170 171 System.out.println("------ObjectId=" + id); 172 173 // last 174 // version 175 JobExecutionConfiguration jobExecutionConfiguration = new JobExecutionConfiguration(); 176 // RunConfiguration run = new RunConfiguration(); 177 jobExecutionConfiguration.setRemoteServer(remoteServer); 178 // jobExecutionConfiguration.setRepository(repository); // 可能是下面指定了repository，这边不写也没差别 179 // jobExecutionConfiguration.setRunConfiguration("28"); // 没有差别 180 jobExecutionConfiguration.setPassingExport(true); // send resources to this server 181 String lastCarteObjectId = Job.sendToSlaveServer(jobMeta, jobExecutionConfiguration, repository, null); 182 System.out.println(lastCarteObjectId); 183 184 // SlaveServerJobStatus jobStatus = null; 185 // do { 186 // Thread.sleep(5000); 187 // jobStatus = remoteServer.getJobStatus(jobMeta.getName(), lastCarteObjectId, 0); 188 // } while (jobStatus != null && jobStatus.isRunning()); 189 // Result oneResult = new Result(); 190 // System.out.println(jobStatus); 191 // if (jobStatus.getResult() != null) { 192 // // 流程完成，得到结果 193 // oneResult = jobStatus.getResult(); 194 // System.out.println("Result:" + oneResult); 195 // } else { 196 // System.out.println("取到空了"); 197 // } 198 199 // SlaveServerJobStatus jobStatus = null; 200 // Result oneResult = new Result(); 201 // while (true) { 202 // try { 203 // jobStatus = remoteServer.getJobStatus(jobMeta.getName(), lastCarteObjectId, 0); 204 // if (jobStatus.getResult() != null) { 205 // // The job is finished, get the result... 206 // oneResult = jobStatus.getResult(); 207 // break; 208 // } 209 // } catch (Exception e1) { 210 // oneResult.setNrErrors(1L); 211 // break; // Stop looking too, chances are too low the server 212 // // will 213 // // come back on-line 214 // } 215 // } 216 } 217 218 /** 219 * 远程服务器上执行trans 220 * @param dir 221 * @param jobName 222 * @throws KettleException 223 */ 224 public static void executeTransRemote(String path, String transName) throws KettleException { 225 226 RepositoryDirectoryInterface dir = repository.findDirectory(path); 227 // 根据转换名称获取转换id 228 // 这个值就是从SELECT ID_TRANSFORMATION FROM `r_transformation` where name='fromto';来的 229 ObjectId id = repository.getTransformationID(transName, dir); 230 System.out.println("------ObjectId=" + id); 231 // 根据转换id查询与其关联的step和hop 232 // select name from r_step where ID_STEP=(SELECT ID_STEP_FROM FROM `r_trans_hop` where 233 // ID_TRANSFORMATION='1'); 234 // select name from r_step where ID_STEP=(SELECT ID_STEP_TO FROM `r_trans_hop` where ID_TRANSFORMATION='1'); 235 TransMeta transMeta = repository.loadTransformation(id, null); 236 237 // last 238 // version 239 TransExecutionConfiguration transExecutionConfiguration = new TransExecutionConfiguration(); 240 241 transExecutionConfiguration.setRemoteServer(remoteServer); 242 243 transExecutionConfiguration.setRepository(repository); 244 245 String lastCarteObjectId = Trans.sendToSlaveServer(transMeta, transExecutionConfiguration, repository, null); 246 System.out.println(lastCarteObjectId); 247 248 } 249 250 }