1.不同資料庫中的表，保證這些表的結構相同

2.表中資料和生成文字資料

3.自定義常量設定和java程式碼

import java.sql.*;
import org.pentaho.di.core.database.*;




public boolean processRow(StepMetaInterface smi, StepDataInterface sdi) throws KettleException
{
    Object[] r = getRow();
    if (r == null) {
        setOutputDone();
        return false;
    }

//獲取資料庫名和表名
    String dbName = getInputRowMeta().getString(r, "conname", null );
    String tablename = getInputRowMeta().getString(r, "tablename", null );
    if (dbName==null||tablename==null) {
      throw new KettleException("Unable to find field with name "+tablename+" in the input row.");
    }
    logBasic("table---"+tablename);

//資料庫連線
    Database database=null;
    DatabaseMeta databaseMeta=null;
    try  {
          databaseMeta = getTransMeta().findDatabase(dbName);
         if (databaseMeta==null) {
         logError("A connection with name "+dbName+" could not be found!");
         setErrors(1);
         return false;
    }
      database = new Database(getTrans(), databaseMeta);
      database.connect();
       logBasic("success!");
    } catch(Exception e) {
      logError("Connecting to database "+dbName+" failed.", e);
      setErrors(1);
      return false;
    }

//查詢表資料
   String sql="select id,name from "+tablename;
   ResultSet resultSet;
     try {
      resultSet = database.openQuery(sql);
      Object[] idxRow = database.getRow(resultSet);
      RowMetaInterface idxRowMeta =null;
      if(idxRow!=null){
           idxRowMeta=database.getReturnRowMeta();
      }
      int i=0;
      while(idxRow!=null){
          r = createOutputRow(r, data.outputRowMeta.size());
          int index = getInputRowMeta().size();
        // Add the index name
        //
        r[index++] = idxRowMeta.getString(idxRow, "id", null);

        // Add the column name
        
        r[index++] = idxRowMeta.getString(idxRow, "name", null);
        putRow(data.outputRowMeta, r);
          idxRow = database.getRow(resultSet);
          i++;
     }
     logBasic("idxRow--length"+i);
    
     }
   catch(Exception e) {
     throw new KettleException(e);
   }

//釋放連線
   if (database!=null) {
      database.disconnect();
      database.closeQuery(resultSet);
    }

    return true;
}