关于通过java调用datax,返回任务执行的方法

public static void entry(final String[] args) throws Throwable { Options options = new Options(); options.addOption("job", true, "Job config."); options.addOption("jobid", true, "Job unique id."); options.addOption("mode", true, "Job runtime mode."); BasicParser parser = new BasicParser(); CommandLine cl = parser.parse(options, args); String jobPath = cl.getOptionValue("job"); // 如果用户没有明确指定jobid, 则 datax.py 会指定 jobid 默认值为-1 String jobIdString = cl.getOptionValue("jobid"); RUNTIME_MODE = cl.getOptionValue("mode"); Configuration configuration = ConfigParser.parse(jobPath); long jobId; if (!"-1".equalsIgnoreCase(jobIdString)) {  jobId = Long.parseLong(jobIdString); } else {  // only for dsc & ds & datax 3 update  String dscJobUrlPatternString = "/instance/(//d{1,})/config.xml";  String dsJobUrlPatternString = "/inner/job/(//d{1,})/config";  String dsTaskGroupUrlPatternString = "/inner/job/(//d{1,})/taskGroup/";  List<String> patternStringList = Arrays.asList(dscJobUrlPatternString,   dsJobUrlPatternString, dsTaskGroupUrlPatternString);  jobId = parseJobIdFromUrl(patternStringList, jobPath); } boolean isStandAloneMode = "standalone".equalsIgnoreCase(RUNTIME_MODE); if (!isStandAloneMode && jobId == -1) {  // 如果不是 standalone 模式，那么 jobId 一定不能为-1  throw DataXException.asDataXException(FrameworkErrorCode.CONFIG_ERROR, "非 standalone 模式必须在 URL 中提供有效的 jobId."); } configuration.set(CoreConstant.DATAX_CORE_CONTAINER_JOB_ID, jobId); //打印vmInfo VMInfo vmInfo = VMInfo.getVmInfo(); if (vmInfo != null) {  LOG.info(vmInfo.toString()); } LOG.info("/n" + Engine.filterJobConfiguration(configuration) + "/n"); LOG.debug(configuration.toJSON()); ConfigurationValidate.doValidate(configuration); Engine engine = new Engine(); engine.start(configuration); }

@Override public void start() { LOG.info("DataX jobContainer starts job."); boolean hasException = false; boolean isDryRun = false; try {  this.startTimeStamp = System.currentTimeMillis();  isDryRun = configuration.getBool(CoreConstant.DATAX_JOB_SETTING_DRYRUN, false);  if (isDryRun) {  LOG.info("jobContainer starts to do preCheck ...");  this.preCheck();  } else {  userConf = configuration.clone();  LOG.debug("jobContainer starts to do preHandle ...");  this.preHandle();  LOG.debug("jobContainer starts to do init ...");  this.init();  LOG.info("jobContainer starts to do prepare ...");  this.prepare();  LOG.info("jobContainer starts to do split ...");  this.totalStage = this.split();  LOG.info("jobContainer starts to do schedule ...");  this.schedule();  LOG.debug("jobContainer starts to do post ...");  this.post();  LOG.debug("jobContainer starts to do postHandle ...");  this.postHandle();  LOG.info("DataX jobId [{}] completed successfully.", this.jobId);  this.invokeHooks();  } } catch (Throwable e) {  LOG.error("Exception when job run", e);  hasException = true;  if (e instanceof OutOfMemoryError) {  this.destroy();  System.gc();  }  if (super.getContainerCommunicator() == null) {  // 由于 containerCollector 是在 scheduler() 中初始化的，所以当在 scheduler() 之前出现异常时，需要在此处对 containerCollector 进行初始化  AbstractContainerCommunicator tempContainerCollector;  // standalone  tempContainerCollector = new StandAloneJobContainerCommunicator(configuration);  super.setContainerCommunicator(tempContainerCollector);  }  Communication communication = super.getContainerCommunicator().collect();  // 汇报前的状态，不需要手动进行设置  // communication.setState(State.FAILED);  communication.setThrowable(e);  communication.setTimestamp(this.endTimeStamp);  Communication tempComm = new Communication();  tempComm.setTimestamp(this.startTransferTimeStamp);  Communication reportCommunication = CommunicationTool.getReportCommunication(communication, tempComm, this.totalStage);  super.getContainerCommunicator().report(reportCommunication);  throw DataXException.asDataXException(   FrameworkErrorCode.RUNTIME_ERROR, e); } finally {  if (!isDryRun) {  this.destroy();  this.endTimeStamp = System.currentTimeMillis();  if (!hasException) {   //最后打印cpu的平均消耗，GC的统计   VMInfo vmInfo = VMInfo.getVmInfo();   if (vmInfo != null) {   vmInfo.getDelta(false);   LOG.info(vmInfo.totalString());   }   LOG.info(PerfTrace.getInstance().summarizeNoException());   this.logStatistics();  }  } } }

private void logStatistics() { long totalCosts = (this.endTimeStamp - this.startTimeStamp) / 1000; long transferCosts = (this.endTransferTimeStamp - this.startTransferTimeStamp) / 1000; if (0L == transferCosts) {  transferCosts = 1L; } if (super.getContainerCommunicator() == null) {  return; } Communication communication = super.getContainerCommunicator().collect(); communication.setTimestamp(this.endTimeStamp); Communication tempComm = new Communication(); tempComm.setTimestamp(this.startTransferTimeStamp); Communication reportCommunication = CommunicationTool.getReportCommunication(communication, tempComm, this.totalStage); // 字节速率 long byteSpeedPerSecond = communication.getLongCounter(CommunicationTool.READ_SUCCEED_BYTES)  / transferCosts; long recordSpeedPerSecond = communication.getLongCounter(CommunicationTool.READ_SUCCEED_RECORDS)  / transferCosts; reportCommunication.setLongCounter(CommunicationTool.BYTE_SPEED, byteSpeedPerSecond); reportCommunication.setLongCounter(CommunicationTool.RECORD_SPEED, recordSpeedPerSecond); super.getContainerCommunicator().report(reportCommunication); LOG.info(String.format(  "/n" + "%-26s: %-18s/n" + "%-26s: %-18s/n" + "%-26s: %19s/n"   + "%-26s: %19s/n" + "%-26s: %19s/n" + "%-26s: %19s/n"   + "%-26s: %19s/n",  "任务启动时刻",  dateFormat.format(startTimeStamp),  "任务结束时刻",  dateFormat.format(endTimeStamp),  "任务总计耗时",  String.valueOf(totalCosts) + "s",  "任务平均流量",  StrUtil.stringify(byteSpeedPerSecond)   + "/s",  "记录写入速度",  String.valueOf(recordSpeedPerSecond)   + "rec/s", "读出记录总数",  String.valueOf(CommunicationTool.getTotalReadRecords(communication)),  "读写失败总数",  String.valueOf(CommunicationTool.getTotalErrorRecords(communication)) )); LOG.info("task-total-info:" + dateFormat.format(startTimeStamp) + "|" +  dateFormat.format(endTimeStamp) + "|" +  String.valueOf(totalCosts) + "|" +  StrUtil.stringify(byteSpeedPerSecond) + "|" +  String.valueOf(recordSpeedPerSecond) + "|" +  String.valueOf(CommunicationTool.getTotalReadRecords(communication)) + "|" +  String.valueOf(CommunicationTool.getTotalErrorRecords(communication)) ); if (communication.getLongCounter(CommunicationTool.TRANSFORMER_SUCCEED_RECORDS) > 0  || communication.getLongCounter(CommunicationTool.TRANSFORMER_FAILED_RECORDS) > 0  || communication.getLongCounter(CommunicationTool.TRANSFORMER_FILTER_RECORDS) > 0) {  LOG.info(String.format(   "/n" + "%-26s: %19s/n" + "%-26s: %19s/n" + "%-26s: %19s/n",   "Transformer成功记录总数",   communication.getLongCounter(CommunicationTool.TRANSFORMER_SUCCEED_RECORDS),   "Transformer失败记录总数",   communication.getLongCounter(CommunicationTool.TRANSFORMER_FAILED_RECORDS),   "Transformer过滤记录总数",   communication.getLongCounter(CommunicationTool.TRANSFORMER_FILTER_RECORDS)  )); } }

public class DataxResult { //任务启动时刻 private long startTimeStamp; //任务结束时刻 private long endTimeStamp; //任务总时耗 private long totalCosts; //任务平均流量 private long byteSpeedPerSecond; //记录写入速度 private long recordSpeedPerSecond; //读出记录总数 private long totalReadRecords; //读写失败总数 private long totalErrorRecords; //成功记录总数 private long transformerSucceedRecords; // 失败记录总数 private long transformerFailedRecords; // 过滤记录总数 private long transformerFilterRecords; //字节数 private long readSucceedBytes; //转换开始时间 private long endTransferTimeStamp; //转换结束时间 private long startTransferTimeStamp; //转换总耗时 private long transferCosts;

private DataxResult logStatistics(DataxResult resultMsg) { long totalCosts = (this.endTimeStamp - this.startTimeStamp) / 1000; long transferCosts = (this.endTransferTimeStamp - this.startTransferTimeStamp) / 1000; if (0L == transferCosts) {  transferCosts = 1L; } if (super.getContainerCommunicator() == null) {  return resultMsg; } Communication communication = super.getContainerCommunicator().collect(); long byteSpeedPerSecond = communication.getLongCounter(CommunicationTool.READ_SUCCEED_BYTES)  / transferCosts; long recordSpeedPerSecond = communication.getLongCounter(CommunicationTool.READ_SUCCEED_RECORDS)  / transferCosts;   return resultMsg.getResultMsg(startTimeStamp,  endTimeStamp,  totalCosts,  byteSpeedPerSecond,  recordSpeedPerSecond,  communication.getLongCounter(CommunicationTool.TRANSFORMER_SUCCEED_RECORDS),  communication.getLongCounter(CommunicationTool.TRANSFORMER_FAILED_RECORDS),  communication.getLongCounter(CommunicationTool.TRANSFORMER_FILTER_RECORDS),  communication.getLongCounter(CommunicationTool.TRANSFORMER_FAILED_RECORDS),  communication.getLongCounter(CommunicationTool.TRANSFORMER_FILTER_RECORDS),  communication.getLongCounter(CommunicationTool.READ_SUCCEED_BYTES),  this.endTransferTimeStamp,  this.startTransferTimeStamp,  transferCosts ); }

@Override public DataxResult start(DataxResult dataxResult) { ... DataxResult result = new DataxResult(); result = logStatistics(dataxResult); ... return result; }

 public DataxResult mockstart(Configuration allConf) { ... DataxResult dataxResult = new DataxResult(); return container.start(dataxResult); }

public class test { public static void main(String[] args) { String[] datxArgs = {"-job", CoreConstant.DATAX_HOME + "//job//job2.json", "-mode", "standalone", "-jobid", "-1"}; try {  DataxResult dataxResult= Engine.mockentry(datxArgs); } catch (Throwable e) {  e.printStackTrace(); } }}