kettle系列-4.kettle定制化開發工具類

      要說的話這個工具類還是比較簡單的，每個方法體都比較小，但用起來還是可以的，把開發中一些常用的步驟封裝了下，不用去kettle源碼中找相關操作的具體實現了。

      算了廢話不多了，直接上重點，代碼如下：

import java.util.List;

import org.apache.log4j.Logger;
import org.pentaho.di.core.KettleEnvironment;
import org.pentaho.di.core.database.DatabaseMeta;
import org.pentaho.di.core.exception.KettleException;
import org.pentaho.di.core.exception.KettleSecurityException;
import org.pentaho.di.job.JobMeta;
import org.pentaho.di.job.entries.job.JobEntryJob;
import org.pentaho.di.job.entries.trans.JobEntryTrans;
import org.pentaho.di.job.entry.JobEntryBase;
import org.pentaho.di.job.entry.JobEntryCopy;
import org.pentaho.di.repository.AbstractRepository;
import org.pentaho.di.repository.LongObjectId;
import org.pentaho.di.repository.RepositoryDirectoryInterface;
import org.pentaho.di.repository.StringObjectId;
import org.pentaho.di.repository.filerep.KettleFileRepository;
import org.pentaho.di.repository.filerep.KettleFileRepositoryMeta;
import org.pentaho.di.repository.kdr.KettleDatabaseRepository;
import org.pentaho.di.repository.kdr.KettleDatabaseRepositoryMeta;
import org.pentaho.di.trans.TransMeta;
import org.pentaho.di.trans.TransPreviewFactory;
import org.pentaho.di.trans.step.BaseStepMeta;
import org.pentaho.di.trans.step.StepMeta;
import org.pentaho.di.trans.step.StepMetaInterface;
import org.pentaho.di.trans.steps.jobexecutor.JobExecutorMeta;
import org.pentaho.di.trans.steps.transexecutor.TransExecutorMeta;


 /**
 * ClassName: KettleUtils <br/>
 * Function: kettle定制化開發工具集. <br/>
 * date: 2015年4月29日 上午8:56:24 <br/>
 * @author jingma
 * @version 0.0.1
 * @since JDK 1.6
 */
public class KettleUtils {
    /**
     * LOG:日志
     */
    public static Logger log = Logger.getLogger(KettleUtils.class);
    /**
     * repository:kettle資源庫
     */
    private static AbstractRepository repository;
    /**
    * 轉換模板
    */
    private static TransMeta transMetaTemplate;
    /**
    * 作業模板
    */
    private static JobMeta jobMetaTemplate;
    
    /**
     * getInstance:獲取的單例資源庫. <br/>
     * @author jingma
     * @return 已經初始化的資源庫
     * @throws KettleException 若沒有初始化則拋出異常
     * @since JDK 1.6
     */
    public static AbstractRepository getInstanceRep() throws KettleException{
        if(repository!=null){
            return repository;
//        }else if(1==1){
//            //TODO jingma:這里以后添加讀取配置初始化連接資源庫
        }else{
            throw new KettleException("沒有初始化資源庫");
        }
    }
    
    /**
     * createFileRep:創建文件資源庫. <br/>
     * @author jingma
     * @param id 資源庫id
     * @param name 資源庫名稱
     * @param description 資源庫描述
     * @param baseDirectory 資源庫目錄
     * @return 已經初始化的資源庫
     * @throws KettleException 
     * @since JDK 1.6
     */
    public static AbstractRepository createFileRep(String id, String name, String description, String baseDirectory) throws KettleException{
        destroy();
        //初始化kettle環境
        if(!KettleEnvironment.isInitialized()){
            KettleEnvironment.init();
        }
        repository = new KettleFileRepository();
        KettleFileRepositoryMeta fileRepMeta = new KettleFileRepositoryMeta( id, name, description, baseDirectory);
        repository.init(fileRepMeta);
        log.info(repository.getName()+"資源庫初始化成功");
        return repository;
    }
    
    /**
     * createDBRep:創建數據庫資源庫. <br/>
     * @author jingma
     * @param name 數據庫連接名稱
     * @param type 數據庫類型
     * @param access 訪問類型
     * @param host ip地址
     * @param db 數據庫名稱
     * @param port 端口
     * @param user 數據庫用戶名
     * @param pass 數據庫密碼
     * @return 初始化的資源庫
     * @throws KettleException 
     * @since JDK 1.6
     */
    public static AbstractRepository createDBRep(String name, String type, String access, String host, 
            String db, String port, String user, String pass) throws KettleException{
        return createDBRep( name, type, access, host, 
             db, port, user, pass, "DBRep", "DBRep", "數據庫資源庫");
    }
    
    /**
     * createDBRep:創建數據庫資源庫. <br/>
     * @author jingma
     * @param name 數據庫連接名稱
     * @param type 數據庫類型
     * @param access 訪問類型
     * @param host ip地址
     * @param db 數據庫名稱
     * @param port 端口
     * @param user 數據庫用戶名
     * @param pass 數據庫密碼
     * @param id 資源庫id
     * @param repName 資源庫名稱
     * @param description 資源庫描述
     * @return 已經初始化的資源庫
     * @throws KettleException 
     * @since JDK 1.6
     */
    public static AbstractRepository createDBRep(String name, String type, String access, String host, 
            String db, String port, String user, String pass,String id, String repName, String description) throws KettleException{
        destroy();
        //初始化kettle環境
        if(!KettleEnvironment.isInitialized()){
            KettleEnvironment.init();
        }
        //創建資源庫對象
        repository = new KettleDatabaseRepository();
        //創建資源庫數據庫對象，類似我們在spoon里面創建資源庫
        DatabaseMeta dataMeta = new DatabaseMeta(name, type, access, host, db, port, user, pass);
        //資源庫元對象
        KettleDatabaseRepositoryMeta kettleDatabaseMeta = 
                new KettleDatabaseRepositoryMeta(id, repName, description, dataMeta);
        //給資源庫賦值
        repository.init(kettleDatabaseMeta);
        log.info(repository.getName()+"資源庫初始化成功");
        return repository;
    }
    
    /**
     * connect:連接資源庫. <br/>
     * @author jingma
     * @return 連接后的資源庫
     * @throws KettleSecurityException
     * @throws KettleException
     * @since JDK 1.6
     */
    public static AbstractRepository connect() throws KettleSecurityException, KettleException{
        return connect(null,null);
    }
    
    /**
     * connect:連接資源庫. <br/>
     * @author jingma
     * @param username 資源庫用戶名
     * @param password 資源庫密碼
     * @return 連接后的資源庫
     * @throws KettleSecurityException
     * @throws KettleException
     * @since JDK 1.6
     */
    public static AbstractRepository connect(String username,String password) throws KettleSecurityException, KettleException{
        repository.connect(username, password);
        log.info(repository.getName()+"資源庫連接成功");
        return repository;
    }
    
    /**
     * setRepository:設置資源庫. <br/>
     * @author jingma
     * @param repository 外部注入資源庫
     * @since JDK 1.6
     */
    public static void setRepository(AbstractRepository repository){
        KettleUtils.repository = repository;
    }
    
    /**
     * destroy:釋放資源庫. <br/>
     * @author jingma
     * @since JDK 1.6
     */
    public static void destroy(){
        if(repository!=null){
            repository.disconnect();
            log.info(repository.getName()+"資源庫釋放成功");
        }
    }

    /**
     * loadJob:通過id加載job. <br/>
     * @author jingma
     * @param jobId 數字型job的id，數據庫資源庫時用此方法
     * @return job元數據
     * @throws KettleException
     * @since JDK 1.6
     */
    public static JobMeta loadJob(long jobId) throws KettleException {
        return repository.loadJob(new LongObjectId(jobId), null);
    }

    /**
     * loadJob:通過id加載job. <br/>
     * @author jingma
     * @param jobId 字符串job的id，文件資源庫時用此方法
     * @return job元數據
     * @throws KettleException
     * @since JDK 1.6
     */
    public static JobMeta loadJob(String jobId) throws KettleException {
        return repository.loadJob(new StringObjectId(jobId), null);
    }

    /**
     * loadTrans:加載作業. <br/>
     * @author jingma
     * @param jobname 作業名稱
     * @param directory 作業路徑
     * @return 作業元數據
     * @since JDK 1.6
     */
    public static JobMeta loadJob(String jobname, String directory) {
        return loadJob(jobname, directory, repository);
    }
    /**
     * loadTrans:加載作業. <br/>
     * @author jingma
     * @param jobname 作業名稱
     * @param directory 作業路徑
     * @param repository 資源庫
     * @return 作業元數據
     * @since JDK 1.6
     */
    public static JobMeta loadJob(String jobname, String directory,AbstractRepository repository) {
        try {
            RepositoryDirectoryInterface dir = repository.findDirectory(directory);
            return repository.loadJob(jobname,dir,null, null);
        } catch (KettleException e) {
            log.error("獲取作業失敗,jobname:"+jobname+",directory:"+directory, e);
        }
        return null;
    }
    
    /**
     * loadTrans:加載轉換. <br/>
     * @author jingma
     * @param transname 轉換名稱
     * @param directory 轉換路徑
     * @return 轉換元數據
     * @since JDK 1.6
     */
    public static TransMeta loadTrans(String transname, String directory) {
        return loadTrans(transname, directory, repository);
    }
    
    /**
     * loadTrans:加載轉換. <br/>
     * @author jingma
     * @param transname 轉換名稱
     * @param directory 轉換路徑
     * @param repository 資源庫
     * @return 轉換元數據
     * @since JDK 1.6
     */
    public static TransMeta loadTrans(String transname, String directory,AbstractRepository repository) {
        try {
            RepositoryDirectoryInterface dir = repository.findDirectory(directory);
            return repository.loadTransformation( transname, dir, null, true, null);
        } catch (KettleException e) {
            log.error("獲取轉換失敗,transname:"+transname+",directory:"+directory, e);
        }
        return null;
    }

    /**
     * loadTrans:根據job元數據獲取指定轉換元數據. <br/>
     * @author jingma
     * @param jobMeta job元數據
     * @param teansName 轉換名稱
     * @return 轉換元數據
     * @since JDK 1.6
     */
    public static TransMeta loadTrans(JobMeta jobMeta, String teansName) {
        JobEntryTrans trans = (JobEntryTrans)(jobMeta.findJobEntry(teansName).getEntry());
        TransMeta transMeta = KettleUtils.loadTrans(trans.getTransname(), trans.getDirectory());
        return transMeta;
    }

    /**
     * 根據轉換元數據和步驟名稱獲取具體的步驟元數據的復制. <br/>
     * 一般是不需要這用這個方法的，該方法獲取的實體不屬於該轉換，相當於一個復制 ，修改了直接保存transMeta是沒有保存到修改的。<br/>
     * 若需要修改轉換，可以使用：(T)transMeta.findStep(stepName).getStepMetaInterface()，
     * 這個方法獲取的步驟是屬於該轉換的，修改后，直接保存transMeta就能實現轉換修改<br/>
     * @author jingma
     * @param transMeta 轉換元數據
     * @param stepName 步驟名稱
     * @param stepMeta 具體的步驟元數據對象
     * @return 從資源庫獲取具體數據的步驟元數據
     * @since JDK 1.6
     */
    public static <T extends BaseStepMeta> T loadStep(TransMeta transMeta, String stepName,
            T stepMeta) {
        StepMeta step = transMeta.findStep(stepName);
        try {
            stepMeta.readRep(KettleUtils.getInstanceRep(), null, step.getObjectId(), KettleUtils.getInstanceRep().readDatabases());
        } catch (KettleException e) {
            log.error("獲取步驟失敗", e);
        }
        return stepMeta;
    }
    /**
    * 根據作業元數據和作業實體名稱獲取具體的作業實體元數據的復制。<br/>
    * 一般是不需要這用這個方法的，該方法獲取的實體不屬於該job了，相當於一個復制 ，修改了直接保存jobMeta是沒有保存到修改的。<br/>
    * 若需要修改job，可以使用：(T)jobMeta.findJobEntry(jobEntryName).getEntry()，
    * 這個方法獲取的實體是屬於job，修改后，直接保存jobMeta就能實現job修改<br/>
    * @author jingma
    * @param jobMeta 作業元數據
    * @param jobEntryName 作業實體名稱
    * @param jobEntryMeta 要獲取的作業實體對象
    * @return 加載了數據的作業實體對象
    */
    public static <T extends JobEntryBase> T loadJobEntry(JobMeta jobMeta, String jobEntryName,
            T jobEntryMeta) {
        try {
            jobEntryMeta.loadRep(KettleUtils.getInstanceRep(), null, 
                    jobMeta.findJobEntry(jobEntryName).getEntry().getObjectId(), 
                    KettleUtils.getInstanceRep().readDatabases(),null);
        } catch (KettleException e) {
            log.error("獲取作業控件失敗", e);
        }
        return jobEntryMeta;
    }

    /**
     * saveTrans:保存轉換. <br/>
     * @author jingma
     * @param transMeta 轉換元數據
     * @throws KettleException
     * @since JDK 1.6
     */
    public static void saveTrans(TransMeta transMeta) throws KettleException {
//        repository.save(transMeta, null, new RepositoryImporter(repository), true );
        repository.save(transMeta, null, null, true );
    }

    /**
     * saveJob:保存job. <br/>
     * @author jingma
     * @param jobMeta job元數據
     * @throws KettleException
     * @since JDK 1.6
     */
    public static void saveJob(JobMeta jobMeta) throws KettleException {
//        repository.save(jobMeta, null, new RepositoryImporter(repository), true );
        repository.save(jobMeta, null, null, true );
    }

    /**
     * isDirectoryExist:判斷指定的job目錄是否存在. <br/>
     * @author jingma
     * @param directoryName
     * @return
     * @since JDK 1.6
     */
    public static boolean isDirectoryExist(String directoryName) {
        try {
            RepositoryDirectoryInterface dir = repository.findDirectory(directoryName);
            if(dir==null){
                return false;
            }else{
                return true;
            }
        } catch (KettleException e) {
            log.error("判斷job目錄是否存在失敗！",e);
        }
        return false;
    }

    /**
    * 將步驟smi設置到轉換trans中<br/>
    * @author jingma
    * @param teans 轉換元數據
    * @param stepName 步驟名稱
    * @param smi 步驟
    */
    public static void setStepToTrans(TransMeta teans, String stepName, StepMetaInterface smi) {
        try {
            StepMeta step = teans.findStep(stepName);
            step.setStepMetaInterface(smi);
        } catch (Exception e) {
            log.error("將步驟smi設置到轉換trans中-失敗",e);
        }
    }

    /**
    * 將步驟smi設置到轉換trans中並保存到資源庫 <br/>
    * @author jingma
    * @param teans 轉換元數據
    * @param stepName 步驟名稱
    * @param smi 步驟
    */
    public static void setStepToTransAndSave(TransMeta teans, String stepName, StepMetaInterface smi) {
        setStepToTrans( teans, stepName, smi);
        try {
            KettleUtils.saveTrans(teans);
        } catch (KettleException e) {
            log.error("將步驟smi設置到轉換trans中並保存到資源庫-失敗",e);
        }
    }

    /**
    * 步驟數據預覽 <br/>
    * @author jingma
    * @param teans 轉換
    * @param testStep 步驟名稱
    * @param smi 步驟實體
    * @param previewSize 預覽的條數
    * @return 預覽結果
    */
    public static List<List<Object>> stepPreview(TransMeta teans,
            String testStep, StepMetaInterface smi, int previewSize) {
        TransMeta previewMeta = TransPreviewFactory.generatePreviewTransformation(
                teans,
                smi,
                testStep);
        TransPreviewUtil tpu = new TransPreviewUtil(
                previewMeta,
                new String[] { testStep },
                new int[] { previewSize } );
        tpu.doPreview();
        return TransPreviewUtil.getData(tpu.getPreviewRowsMeta(testStep),tpu.getPreviewRows(testStep));
    }

    /**
    * 將指定job復制到KettleUtils中的資源庫 <br/>
    * @author jingma
    * @param jobName job名稱
    * @param jobPath job路徑
    * @param repository 來源資源庫
    * @throws KettleException
    */
    public static void jobCopy(String jobName,String jobPath,AbstractRepository repository) throws KettleException {
        JobMeta jobMeta = KettleUtils.loadJob(jobName,jobPath,repository);
        for(JobEntryCopy jec:jobMeta.getJobCopies()){
            if(jec.isTransformation()){
                JobEntryTrans jet = (JobEntryTrans)jec.getEntry();
                transCopy(jet.getObjectName(), jet.getDirectory(),repository);
            }else if(jec.isJob()){
                JobEntryJob jej = (JobEntryJob)jec.getEntry();
                jobCopy(jej.getObjectName(),jej.getDirectory(),repository);
            }
        }
        jobMeta.setRepository(KettleUtils.getInstanceRep());
        jobMeta.setMetaStore(KettleUtils.getInstanceRep().getMetaStore());
        if(!isDirectoryExist(jobPath)){
            //所在目錄不存在則創建
            KettleUtils.repository.createRepositoryDirectory(KettleUtils.repository.findDirectory("/"), jobPath);
        }
        KettleUtils.saveJob(jobMeta);
    }

    /**
    * 將指定轉換復制到KettleUtils中的資源庫 <br/>
    * @author jingma
    * @param jobName 轉換名稱
    * @param jobPath 轉換路徑
    * @param repository 來源資源庫
    * @throws KettleException
    */
    public static void transCopy(String transName,String transPath,AbstractRepository repository) throws KettleException {
        TransMeta tm = KettleUtils.loadTrans(transName, transPath, repository);
        for(StepMeta sm:tm.getSteps()){
            if(sm.isJobExecutor()){
                JobExecutorMeta jem = (JobExecutorMeta)sm.getStepMetaInterface();
                jobCopy(jem.getJobName(),jem.getDirectoryPath(),repository);
            }
            else if(sm.getStepMetaInterface() instanceof TransExecutorMeta){
                TransExecutorMeta te = (TransExecutorMeta)sm.getStepMetaInterface();
                transCopy(te.getTransName(), te.getDirectoryPath(),repository);
            }
        }
        if(!isDirectoryExist(transPath)){
            //所在目錄不存在則創建
            KettleUtils.repository.createRepositoryDirectory(KettleUtils.repository.findDirectory("/"), transPath);
        }
        tm.setRepository(KettleUtils.getInstanceRep());
        tm.setMetaStore(KettleUtils.getInstanceRep().getMetaStore());
        KettleUtils.saveTrans(tm);
    }

    /**
     * @return transMetaTemplate 
     */
    public static TransMeta getTransMetaTemplate() {
//        if(transMetaTemplate==null){
//            setTransMetaTemplate(KettleUtils.loadTrans(SysCode.TRANS_TEMPLATE_NAME, SysCode.TEMPLATE_DIR));
//        }
        return transMetaTemplate;
    }

    /**
     * @param transMetaTemplate the transMetaTemplate to set
     */
    public static void setTransMetaTemplate(TransMeta transMetaTemplate) {
        KettleUtils.transMetaTemplate = transMetaTemplate;
    }

    /**
     * @return jobMetaTemplate 
     */
    public static JobMeta getJobMetaTemplate() {
//        if(jobMetaTemplate==null){
//            setJobMetaTemplate(KettleUtils.loadJob(SysCode.JOB_TEMPLATE_NAME, SysCode.TEMPLATE_DIR));
//        }
        return jobMetaTemplate;
    }

    /**
     * @param jobMetaTemplate the jobMetaTemplate to set
     */
    public static void setJobMetaTemplate(JobMeta jobMetaTemplate) {
        KettleUtils.jobMetaTemplate = jobMetaTemplate;
    }
    
}

      上面就是我定制化開發中編寫的工具類，基本是一個獨立可用的類，里面的模版jobMetaTemplate、transMetaTemplate可用刪除，也可以根據需要使用。其中的步驟預覽功能需要依賴另外一個類，這個類是我從kettleUI層提取出來並做了修改的，這里貼出代碼吧：

/*! ******************************************************************************
 *
 * Pentaho Data Integration
 *
 * Copyright (C) 2002-2013 by Pentaho : http://www.pentaho.com
 *
 *******************************************************************************
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with
 * the License. You may obtain a copy of the License at
 *
 *    http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 *
 ******************************************************************************/

package com.iflytek.kettle.utils;

import java.util.ArrayList;
import java.util.List;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.pentaho.di.core.Const;
import org.pentaho.di.core.exception.KettleException;
import org.pentaho.di.core.exception.KettleValueException;
import org.pentaho.di.core.logging.KettleLogStore;
import org.pentaho.di.core.row.RowMetaInterface;
import org.pentaho.di.core.row.ValueMetaInterface;
import org.pentaho.di.trans.Trans;
import org.pentaho.di.trans.TransMeta;
import org.pentaho.di.trans.debug.BreakPointListener;
import org.pentaho.di.trans.debug.StepDebugMeta;
import org.pentaho.di.trans.debug.TransDebugMeta;
import org.pentaho.di.trans.step.StepMeta;

/**
 * Takes care of displaying a dialog that will handle the wait while previewing a transformation...
 *
 * @author Matt
 * @since 13-jan-2006
 */
public class TransPreviewUtil {
  public static final int MAX_BINARY_STRING_PREVIEW_SIZE = 1000000;
  private static Log log = LogFactory.getLog(TransPreviewUtil.class);
  private TransMeta transMeta;
  private String[] previewStepNames;
  private int[] previewSize;
  private Trans trans;

  private boolean cancelled;
  private String loggingText;
  private TransDebugMeta transDebugMeta;

  /**
   * Creates a new dialog that will handle the wait while previewing a transformation...
   */
  public TransPreviewUtil( TransMeta transMeta, String[] previewStepNames, int[] previewSize ) {
    this.transMeta = transMeta;
    this.previewStepNames = previewStepNames;
    this.previewSize = previewSize;

    cancelled = false;
  }

  public void doPreview() {
    // This transformation is ready to run in preview!
    trans = new Trans( transMeta );

    // Prepare the execution...
    //
    try {
      trans.prepareExecution( null );
    } catch ( final KettleException e ) {
        log.error("", e);
      return;
    }

    // Add the preview / debugging information...
    //
    transDebugMeta = new TransDebugMeta( transMeta );
    for ( int i = 0; i < previewStepNames.length; i++ ) {
      StepMeta stepMeta = transMeta.findStep( previewStepNames[i] );
      StepDebugMeta stepDebugMeta = new StepDebugMeta( stepMeta );
      stepDebugMeta.setReadingFirstRows( true );
      stepDebugMeta.setRowCount( previewSize[i] );
      transDebugMeta.getStepDebugMetaMap().put( stepMeta, stepDebugMeta );
    }

    // set the appropriate listeners on the transformation...
    //
    transDebugMeta.addRowListenersToTransformation( trans );

    // Fire off the step threads... start running!
    //
    try {
      trans.startThreads();
    } catch ( final KettleException e ) {
        log.error("", e);
      // It makes no sense to continue, so just stop running...
      //
      return;
    }

    final List<String> previewComplete = new ArrayList<String>();

    while ( previewComplete.size() < previewStepNames.length
      && !trans.isFinished() ) {
      // We add a break-point that is called every time we have a step with a full preview row buffer
      // That makes it easy and fast to see if we have all the rows we need
      //
      transDebugMeta.addBreakPointListers( new BreakPointListener() {
        public void breakPointHit( TransDebugMeta transDebugMeta, StepDebugMeta stepDebugMeta,
          RowMetaInterface rowBufferMeta, List<Object[]> rowBuffer ) {
          String stepName = stepDebugMeta.getStepMeta().getName();
          previewComplete.add( stepName );
        }
      } );

      // Change the percentage...
      try {
        Thread.sleep( 500 );
      } catch ( InterruptedException e ) {
          log.error("", e);
        // Ignore errors
      }

    }

    trans.stopAll();

    // Capture preview activity to a String:
    loggingText =
      KettleLogStore.getAppender().getBuffer( trans.getLogChannel().getLogChannelId(), true ).toString();

  }

  /**
   * @param stepname
   *          the name of the step to get the preview rows for
   * @return A list of rows as the result of the preview run.
   */
  public List<Object[]> getPreviewRows( String stepname ) {
    if ( transDebugMeta == null ) {
      return null;
    }

    for ( StepMeta stepMeta : transDebugMeta.getStepDebugMetaMap().keySet() ) {
      if ( stepMeta.getName().equals( stepname ) ) {
        StepDebugMeta stepDebugMeta = transDebugMeta.getStepDebugMetaMap().get( stepMeta );
        return stepDebugMeta.getRowBuffer();
      }
    }
    return null;
  }

  /**
   * @param stepname
   *          the name of the step to get the preview rows for
   * @return A description of the row (metadata)
   */
  public RowMetaInterface getPreviewRowsMeta( String stepname ) {
    if ( transDebugMeta == null ) {
      return null;
    }

    for ( StepMeta stepMeta : transDebugMeta.getStepDebugMetaMap().keySet() ) {
      if ( stepMeta.getName().equals( stepname ) ) {
        StepDebugMeta stepDebugMeta = transDebugMeta.getStepDebugMetaMap().get( stepMeta );
        return stepDebugMeta.getRowBufferMeta();
      }
    }
    return null;
  }

  /**
   * @return true is the preview was canceled by the user
   */
  public boolean isCancelled() {
    return cancelled;
  }

  /**
   * @return The logging text from the latest preview run
   */
  public String getLoggingText() {
    return loggingText;
  }

  /**
   *
   * @return The transformation object that executed the preview TransMeta
   */
  public Trans getTrans() {
    return trans;
  }

  /**
   * @return the transDebugMeta
   */
  public TransDebugMeta getTransDebugMeta() {
    return transDebugMeta;
  }
  

  /**
   * Copy information from the meta-data input to the dialog fields.
 * @param rowMetaInterface 
   */
  public static List<List<Object>> getData(RowMetaInterface rowMeta, List<Object[]> buffer) {
    List<List<Object>> result = new ArrayList<List<Object>>();
    List<Object> row1 = new ArrayList<Object>();
    for ( int i = 0; i < buffer.size(); i++ ) {
      row1 = new ArrayList<Object>();
      Object[] row = buffer.get( i );
      getDataForRow( rowMeta, row1, row );
      result.add(row1);
    }
    return result;
  }

  public static int getDataForRow( RowMetaInterface rowMeta, List<Object> row1, Object[] row ) {
    int nrErrors = 0;

    // Display the correct line item...
    //
    for ( int c = 0; c < rowMeta.size(); c++ ) {
      ValueMetaInterface v = rowMeta.getValueMeta( c );
      String show;
      try {
        show = v.getString( row[c] );
        if ( v.isBinary() && show != null && show.length() > MAX_BINARY_STRING_PREVIEW_SIZE ) {
          // We want to limit the size of the strings during preview to keep all SWT widgets happy.
          //
          show = show.substring( 0, MAX_BINARY_STRING_PREVIEW_SIZE );
        }
      } catch ( KettleValueException e ) {
        nrErrors++;
        if ( nrErrors < 25 ) {
          log.error( Const.getStackTracker( e ) );
        }
        show = null;
      } catch ( ArrayIndexOutOfBoundsException e ) {
        nrErrors++;
        if ( nrErrors < 25 ) {
          log.error( Const.getStackTracker( e ) );
        }
        show = null;
      }

      if ( show != null ) {
          row1.add(show);
      } else {
        // Set null value
          row1.add("<null>");
      }
    }

    return nrErrors;

  }
}

      以上兩個類復制到自己項目中就可以用了，這個工具類是根據我自己需求編寫的，根據使用場景不同，你也可以添加其他方法，豐富該類的功能。

      關於使用示例，本來也想貼出來的，但相關代碼依賴較多，同時工具類本身注釋很詳細，方法體很小，自己看一遍就能了解個大概，所以這里就不提供使用示例代碼，希望對有需要的人有所幫助。