1.KettleUtil工具类，可以执行本地的转换或作业，可以连接kettle资源库并执行资源库中的转换或作业
1 package com.zxyp.kettle; 2 3 import org.pentaho.di.cluster.ClusterSchema; 4 import org.pentaho.di.cluster.SlaveServer; 5 import org.pentaho.di.core.KettleEnvironment; 6 import org.pentaho.di.core.database.DatabaseMeta; 7 import org.pentaho.di.core.exception.KettleException; 8 import org.pentaho.di.core.logging.LogLevel; 9 import org.pentaho.di.core.util.EnvUtil; 10 import org.pentaho.di.job.Job; 11 import org.pentaho.di.job.JobExecutionConfiguration; 12 import org.pentaho.di.job.JobMeta; 13 import org.pentaho.di.repository.RepositoryDirectoryInterface; 14 import org.pentaho.di.repository.kdr.KettleDatabaseRepository; 15 import org.pentaho.di.repository.kdr.KettleDatabaseRepositoryMeta; 16 import org.pentaho.di.trans.Trans; 17 import org.pentaho.di.trans.TransExecutionConfiguration; 18 import org.pentaho.di.trans.TransMeta; 19 20 public class KettleUtil { 21 22 private String connetionName = "carte"; 23 private String databaseType = "MYSQL"; 24 private String connectionType = "Native(JDBC)"; 25 private String hostAddress = "192.168.10.147"; 26 private String databaseName = "kettle"; 27 private String databasePort = "3306"; 28 private String userName = "root"; 29 private String password = "root"; 30 private String repoName = "repo"; 31 private String repoUserName = "admin"; 32 private String repoPassword = "admin"; 33 private String repoJobDir = "/"; 34 private String repoTransDir = "/"; 35 private String slaveName = "master"; 36 private String slaveHostname = "192.168.10.147"; 37 private String slavePort = "8080"; 38 private String slaveUsername = "cluster"; 39 private String slavePassword = "cluster"; 40 41 public String getConnetionName() { 42 return connetionName; 43 } 44 45 public void setConnetionName(String connetionName) { 46 this.connetionName = connetionName; 47 } 48 49 public String getDatabaseType() { 50 return databaseType; 51 } 52 53 public void setDatabaseType(String databaseType) { 54 this.databaseType = databaseType; 55 } 56 57 public String getConnectionType() { 58 return connectionType; 59 } 60 61 public void setConnectionType(String connectionType) { 62 this.connectionType = connectionType; 63 } 64 65 public String getHostAddress() { 66 return hostAddress; 67 } 68 69 public void setHostAddress(String hostAddress) { 70 this.hostAddress = hostAddress; 71 } 72 73 public String getDatabaseName() { 74 return databaseName; 75 } 76 77 public void setDatabaseName(String databaseName) { 78 this.databaseName = databaseName; 79 } 80 81 public String getDatabasePort() { 82 return databasePort; 83 } 84 85 public void setDatabasePort(String databasePort) { 86 this.databasePort = databasePort; 87 } 88 89 public String getUserName() { 90 return userName; 91 } 92 93 public void setUserName(String userName) { 94 this.userName = userName; 95 } 96 97 public String getPassword() { 98 return password; 99 } 100 101 public void setPassword(String password) { 102 this.password = password; 103 } 104 105 public String getRepoName() { 106 return repoName; 107 } 108 109 public void setRepoName(String repoName) { 110 this.repoName = repoName; 111 } 112 113 public String getRepoUserName() { 114 return repoUserName; 115 } 116 117 public void setRepoUserName(String repoUserName) { 118 this.repoUserName = repoUserName; 119 } 120 121 public String getRepoPassword() { 122 return repoPassword; 123 } 124 125 public void setRepoPassword(String repoPassword) { 126 this.repoPassword = repoPassword; 127 } 128 129 public String getRepoJobDir() { 130 return repoJobDir; 131 } 132 133 public void setRepoJobDir(String repoJobDir) { 134 this.repoJobDir = repoJobDir; 135 } 136 137 public String getRepoTransDir() { 138 return repoTransDir; 139 } 140 141 public void setRepoTransDir(String repoTransDir) { 142 this.repoTransDir = repoTransDir; 143 } 144 145 public String getSlaveName() { 146 return slaveName; 147 } 148 149 public void setSlaveName(String slaveName) { 150 this.slaveName = slaveName; 151 } 152 153 public String getSlaveHostname() { 154 return slaveHostname; 155 } 156 157 public void setSlaveHostname(String slaveHostname) { 158 this.slaveHostname = slaveHostname; 159 } 160 161 public String getSlavePort() { 162 return slavePort; 163 } 164 165 public void setSlavePort(String slavePort) { 166 this.slavePort = slavePort; 167 } 168 169 public String getSlaveUsername() { 170 return slaveUsername; 171 } 172 173 public void setSlaveUsername(String slaveUsername) { 174 this.slaveUsername = slaveUsername; 175 } 176 177 public String getSlavePassword() { 178 return slavePassword; 179 } 180 181 public void setSlavePassword(String slavePassword) { 182 this.slavePassword = slavePassword; 183 } 184 185 /** 186 * 调用trans文件 187 * @param transFileName 188 * @throws Exception 189 */ 190 public static void callNativeTrans(String transFileName) throws Exception{ 191 callNativeTransWithParams(null, transFileName); 192 } 193 194 /** 195 * 调用trans文件 带参数的 196 * @param params 197 * @param transFileName 198 * @throws Exception 199 */ 200 public static void callNativeTransWithParams(String[] params ,String transFileName) throws Exception{ 201 // 初始化 202 KettleEnvironment.init(); 203 EnvUtil.environmentInit(); 204 TransMeta transMeta = new TransMeta(transFileName); 205 //转换 206 Trans trans = new Trans(transMeta); 207 //执行 208 trans.execute(params); 209 //等待结束 210 trans.waitUntilFinished(); 211 //抛出异常 212 if(trans.getErrors() > 0){ 213 throw new Exception("There are errors during transformation exception!(传输过程中发生异常)"); 214 } 215 } 216 217 /** 218 * 调用job文件 219 * @param jobName 220 * @throws Exception 221 */ 222 public static void callNativeJob(String jobName) throws Exception{ 223 // 初始化 224 KettleEnvironment.init(); 225 226 JobMeta jobMeta = new JobMeta(jobName,null); 227 Job job = new Job(null, jobMeta); 228 //向Job 脚本传递参数，脚本中获取参数值：${参数名} 229 //job.setVariable(paraname, paravalue); 230 job.start(); 231 job.waitUntilFinished(); 232 if (job.getErrors() > 0) { 233 throw new Exception("There are errors during job exception!(执行job发生异常)"); 234 } 235 } 236 /** 237 * 资源库连接 238 * @return 连接到的资源库 239 * @throws KettleException 240 */ 241 public static Object RepositoryCon() throws KettleException { 242 // 初始化 243 // EnvUtil.environmentInit(); 244 KettleEnvironment.init(); 245 // 数据库连接元对象 246 DatabaseMeta dataMeta = new DatabaseMeta(new KettleUtil().getConnetionName(), new KettleUtil().getDatabaseType(), new KettleUtil().getConnetionName(),new KettleUtil().getHostAddress(), new KettleUtil().getDatabaseName(), new KettleUtil().getDatabasePort(), 247 new KettleUtil().getUserName(), new KettleUtil().getPassword()); 248 // 数据库形式的资源库元对象 249 KettleDatabaseRepositoryMeta repInfo = new KettleDatabaseRepositoryMeta(); 250 // 251 repInfo.setConnection(dataMeta); 252 repInfo.setName(new KettleUtil().getRepoName()); 253 // 数据库形式的资源库对象 254 KettleDatabaseRepository rep = new KettleDatabaseRepository(); 255 // 用资源库元对象初始化资源库对象 256 rep.init(repInfo); 257 // 连接到资源库 258 rep.connect(new KettleUtil().getRepoUserName(), new KettleUtil().getRepoPassword());// 默认的连接资源库的用户名和密码 259 if (rep.isConnected()) { 260 System.out.println("连接成功"); 261 return rep; 262 } else { 263 System.out.println("连接失败"); 264 return null; 265 } 266 } 267 /** 268 * 以子服务方式执行资源库中的job 269 * @param rep 270 * @param jobName 271 */ 272 public static void runJob(KettleDatabaseRepository rep, String jobName) { 273 try { 274 RepositoryDirectoryInterface dir = rep.findDirectory(new KettleUtil().getRepoJobDir());// 根据指定的字符串路径 找到目录 275 // 加载指定的job 276 JobMeta jobMeta = rep.loadJob(rep.getJobId(jobName, dir), null); 277 Job job = new Job(rep, jobMeta); 278 279 // 设置参数 280 //jobMeta.setParameterValue("method", "update"); 281 //jobMeta.setParameterValue("tsm5", "07bb40f7200448b3a544786dc5e28845"); 282 //jobMeta.setParameterValue("args"," {'fkid':'07bb40f7200448b3a544786dc5e28845','svctype':'Diffwkrlifehelp','content':'更新3','sysuuid':'01ee0e61f357476b8dbb4be49ddecc77','uid':'1033','role':'3999','posi':'2999'}"); 283 284 job.setLogLevel(LogLevel.BASIC); 285 //设置slaveserver信息 286 SlaveServer ssi = new SlaveServer(); 287 ssi.setHostname(new KettleUtil().getSlaveHostname()); 288 ssi.setPort(new KettleUtil().getSlavePort()); 289 ssi.setName(new KettleUtil().getSlaveName()); 290 ssi.setUsername(new KettleUtil().getSlaveUsername()); 291 ssi.setPassword(new KettleUtil().getSlavePassword()); 292 //为job设置slaveserve 293 job.setExecutingServer(new KettleUtil().getSlaveName()); 294 //ClusterSchema cluster = jobgetTransMeta().findFirstUsedClusterSchema(); 295 JobExecutionConfiguration jobExecutionConfiguration = new JobExecutionConfiguration(); 296 jobExecutionConfiguration.setExecutingLocally(false); 297 jobExecutionConfiguration.setExecutingRemotely(true); 298 jobExecutionConfiguration.setRemoteServer(ssi); 299 jobExecutionConfiguration.setRepository(rep); 300 301 String carteObjectId = Job.sendToSlaveServer(jobMeta, jobExecutionConfiguration, rep, null); 302 System.out.println(carteObjectId); 303 304 /*普通执行 305 job.run(); 306 job.waitUntilFinished();// 等待job执行完； 307 job.setFinished(true); 308 System.out.println(job.getResult()); 309 */ 310 } catch (Exception e) { 311 e.printStackTrace(); 312 } 313 } 314 315 public static void runTrans(KettleDatabaseRepository rep,String transName){ 316 try{ 317 RepositoryDirectoryInterface dir = rep.findDirectory(new KettleUtil().getRepoTransDir());//根据指定的字符串路径 找到目录 318 TransMeta tmeta = rep.loadTransformation(rep.getTransformationID(transName, dir), null); 319 //设置参数 320 //tmeta.setParameterValue("", ""); 321 Trans trans = new Trans(tmeta); 322 ClusterSchema cluster = trans.getTransMeta().findFirstUsedClusterSchema(); 323 if (cluster != null) { 324 TransExecutionConfiguration executionConfiguration = new TransExecutionConfiguration(); 325 executionConfiguration.setExecutingLocally(false); 326 executionConfiguration.setExecutingRemotely(false); 327 executionConfiguration.setExecutingClustered(true); //如果有就设置以集群方式运行 328 executionConfiguration.setClusterPosting(true); 329 executionConfiguration.setClusterPreparing(true); 330 executionConfiguration.setClusterStarting(true); 331 executionConfiguration.setClusterShowingTransformation(false); 332 executionConfiguration.setSafeModeEnabled(false); 333 executionConfiguration.setRepository(rep); 334 executionConfiguration.setLogLevel(LogLevel.BASIC); 335 executionConfiguration.setVariables(trans.getTransMeta()); 336 TransMeta transMeta = trans.getTransMeta(); 337 try { 338 Trans.executeClustered(transMeta, executionConfiguration); 339 System.out.println("执行完毕"); 340 } catch (Exception e) { 341 e.printStackTrace(); 342 } 343 } else {/*扩展元数据注入的转换可以集群运行--结束代码 */ 344 trans.startThreads(); 345 while (!trans.isFinished() && !trans.isStopped()) { 346 347 } 348 if(trans.getErrors()>0){ 349 System.out.println("有异常"); 350 } 351 System.exit(0); 352 } 353 354 /*普通执行 355 trans.execute(null);//执行trans 356 trans.waitUntilFinished(); 357 if(trans.getErrors()>0){ 358 System.out.println("有异常"); 359 } 360 */ 361 }catch(Exception e){ 362 e.printStackTrace(); 363 } 364 } 365 366 367 }
2.kettle资源库中添加数据库
Object repo = KettleUtil.RepositoryCon(); KettleDatabaseRepository kettleRepo = (KettleDatabaseRepository) repo; String conn_name="ask"; String type = "MYSQL"; String access = "Native"; String host_name="127.0.0.1"; String db_name="ask"; String db_port="3306"; String user_name="root"; String pass_word="root"; DatabaseMeta database = new DatabaseMeta(conn_name, type, access, host_name, db_name, db_port, user_name, pass_word); kettleRepo.save(database, null);
3.kettle资源库中删除数据库
Object repo = KettleUtil.RepositoryCon();
KettleDatabaseRepository kettleRepo = (KettleDatabaseRepository) repo;
kettleRepo.deleteDatabaseMeta("ask");
4.kettle资源库中查询及更新数据库信息
Object repo = KettleUtil.RepositoryCon(); KettleDatabaseRepository kettleRepo = (KettleDatabaseRepository) repo; List<DatabaseMeta> databaseMetas = kettleRepo.readDatabases(); for (DatabaseMeta databaseMeta : databaseMetas) { System.out.println(databaseMeta.getDatabaseName()); if("ask".equals(databaseMeta.getDatabaseName())) { databaseMeta.setName("newName"); kettleRepo.save(databaseMeta, null); } }