Kettle源码分析之Job执行.doc

Kettle源码分析之Job执行1. 源码结构如图所示 主要有五个source folder组成src 包含调度逻辑和具体的执行代码。最重要的两个包为org.pentaho.di.job和org.pentaho.di.trans。src-ui 主要是一些图形界面的代码，最主要的是org.pentaho.di.ui包。test未研究, 应该是一些Junit测试的代码，scr-core 主要是定义了系统用到的一些值类型和对应的接口、异常类、XML相关的接口及封装、log设置、组件加载、行数据、元信息等src-db 最主要的是org.pentaho.di.core.database包，包含各数据库对应的元数据类和不同数据库必须继承的基类和必须实现的接口。2 Job执行相关的类与接口2.1 JobGraph类维护整个Job编辑区的信息和相应的操作主要的成员变量和方法private JobMeta jobMeta; 由Job编辑面板动态维护private Repository rep;private Job job;private JobTracker jobTracker;用于跟踪日志记录private Date startDate, endDate, currentDate, logDate, depDate;private boolean active, stopped;状态位private List sourceRows; 返回结果的数据内容private Result result;每次执行完一个jobentry返回结果public synchronized void startJob(JobExecutionConfiguration executionConfiguration) throws KettleException2.2 JobMeta维护整个Job的元数据。主要成员变量：protected String name;protected String filename;public List jobcopies;保存不同、 JobEntry或者相同JobEntry的副本列表public List jobhops;保存jobentries之间的链接关系。public List databases;2.3 JobEntryInterface每个具体org.pentaho.di.job.entries包下的 entry类需要实现的接口。包含execute（）方法。2.4 Result每一个jobEntryInterface的实现类在完成相应功能时，返回结果的类型。主要成员变量：private boolean result;执行是否出现异常private int exitStatus; 执行结果状态private List rows;一个jobEntry完成处理后的数据（若存在）private Map resultFiles;2.5 JobEntryCopy维护每一个不同entry或者相同entry的不同副本的信息主要成员：private JobEntryInterface entry;具体entry,执行入口private int nr;副本数，一个编辑区里可以出现多个相同组件private boolean selected;private Point location;图标位置private boolean draw;private ObjectId id;3Kettle之Kitchen为入口执行Job3.1 包src 源码文件夹下的org.pentaho.di.kitchen3.2 作用Kitchen是以命令行的方式调用执行Job3.3 主要使用的类和流程l 初始化Kettle环境l 解析命令行参数获取Job对象l 执行Job3.3.1 初始化环境使用KettleEnvironment.init()方法，KettleEnvironment类待研究3.3.2 解析命令行参数-rep请求的资源库-user资源库用户名-pass资源库密码-job资源库中job的名称-dir资源库中的目录-file要执行的文件的文件名称包括路径-level日志级别-log指定日志文件-listdir列出资源库中所有目录-listjobs列出指定目录下的所有jobs-listrep列出可用的资源库-norep不记录日志到资源库3.3.3 执行JobJob继承Thread类，是一个线程，Job运行需使用job.start()方法public class Job extends Thread implements VariableSpace, NamedParams, HasLogChannelInterface, LoggingObjectInterface /./调用线程的start()方法，实际上是调用run()方法，job的run()方法会调用execute()方法，此方法称为job的execute1方法。 Job的execute1方法主要作用是从jobMeta中找到job的入口信息，然后开始条件调用job的execute2方法关键代码如下startpoint = jobMeta.findJobEntry(JobMeta.STRING_SPECIAL_START, 0, false); if (startpoint = null) throw new KettleJobException(BaseMessages.getString(PKG, Job.Log.CounldNotFindStartingPoint); JobEntrySpecial jes = (JobEntrySpecial) startpoint.getEntry(); Result res = null; boolean isFirst = true; while ( (jes.isRepeat() | isFirst) & !isStopped() isFirst = false; res = execute(0, null, startpoint, null, BaseMessages.getString(PKG, Job.Reason.Started); 主要功能是根据参数startpoint，提取对应的jobentry，执行对应的jobentry操作，再根据JobMeta的hop信息依次得到下一个jobentry，递归调用execute方法2调用，代码如下：/ What entry is next?JobEntryInterface jobEntryInterface = jobEntryCopy.getEntry();JobEntryInterface cloneJei = (JobEntryInterface) jobEntryInterface.clone();/以下是执行JobEntryInterface的实现类execute()方法final Result result = cloneJei.execute(prevResult, nr);/根据jobMeta的Hop信息，找到下面的个数/ Launch only those where the hop indicates true or false /int nrNext = jobMeta.findNrNextJobEntries(jobEntryCopy);for (int i=0;inrNext & !isStopped();i+)/ The next entry is.final JobEntryCopy nextEntry = jobMeta.findNextJobEntry(jobEntryCopy, i);/ See if we need to execute this.final JobHopMeta hi = jobMeta.findJobHop(jobEntryCopy, nextEntry);/./res = execute(nr+1, result, nextEntry, jobEntryCopy, nextComment);/. 在Kitchen.java中，或者result，进行相应的处理 result = job.getResult(); / Execute the selected job.4Kettle之Spoon为入口，执行Job4.1 Spoon执行Job代码分析Spoon是转换(transformation)或者作业(Job)的设计工具。允许用户通过图形界面来设计ETL转换过程，也可以在这个界面设计Job。Spoon对应于org.pentaho.di.ui.spoon包下面的Spoon类。4.1.1 Spoon类的main()方法public static void main(String a) throws KettleException try / The core plugin types dont know about UI classes. Add them in now / before the PluginRegistry inits. Display display = new Display(); Splash splash = new Splash(display); /启动画面、版权页 registerUIPluginObjectTypes(); KettleEnvironment.init();/ 初始化kettle环境 List args = new ArrayList(java.util.Arrays.asList(a); CommandLineOption commandLineOptions = getCommandLineArgs(args);/获取命令行参数/ ./ try staticSpoon.lifecycleSupport.onStart(staticSpoon); catch (LifecycleException e) /./ staticSpoon.setArguments(args.toArray(new Stringargs.size(); staticSpoon.start(splash, commandLineOptions); catch (Throwable t) /./ / Kill all remaining things in this VM! System.exit(0); 在Spoon界面中打开一个Job，界面如图所示 设计好Job后，点击启动按钮，触发监听器，调用Spoon类的runFile()方法4.1.2 Spoon类的runFile()方法代码如下public void runFile() executeFile(true, false, false, false, false, null, false); 调用Spoon类的executeFile()方法代码如下public void executeFile(boolean local, boolean remote, boolean cluster, boolean preview, boolean debug, Date replayDate, boolean safe) /. / JobMeta jobMeta = getActiveJob(); if (jobMeta != null) executeJob(jobMeta, local, remote, replayDate, safe); Kettle通过调用getActionJob()方法，获取当前的Job的jobMeta,然后调用Spoon类的executeJob()方法。4.1.3 Spoon类的executeJob()方法public void executeJob(JobMeta jobMeta, boolean local, boolean remote, Date replayDate, boolean safe) / .addJobLog(jobMeta); / JobLog jobLog = getActiveJobLog(); / jobLog.startJob(replayDate); try .executeJob(jobMeta, local, remote, replayDate, safe); catch (Exception e) new ErrorDialog(shell, Execute job, There was an error during job execution, e); delegates是SpoonDelegate的一个对象,SpoonDelegate是一个代理类，SpoonJobDelegate继承了SpoonDelegate，调用