kettle迁移创建表和表数据

public boolean processRow(StepMetaInterface smi, StepDataInterface sdi) throws KettleException
 {
     // First, get a row from the default input hop
  //
  Object[] r = getRow();

  String databasesXML =
         "<?xml version=\"1.0\" encoding=\"UTF-8\"?>" +
         "<connection>" +
           "<name>kettle</name>" +
           "<server>127.0.0.1</server>" +
           "<type>Oracle</type>" +
           "<access>Native</access>" +
           "<database>orcl</database>" +
           "<port>1521</port>" +
           "<username>kscenter20151207</username>" +
           "<password>kscenter</password>" +
         "</connection>";
  org.pentaho.di.core.database.DatabaseMeta dbmeta = new org.pentaho.di.core.database.DatabaseMeta(databasesXML);;

  if(dbmeta!=null)
  {
   org.pentaho.di.core.database.Database db=new org.pentaho.di.core.database.Database(dbmeta);
   
   try
   {
    db.connect();

    String tablename = getVariable("TABLENAME");

    logBasic("开始创建表：" + tablename);
    
    if(tablename!=null && tablename.trim().length()>0)
    {
     org.pentaho.di.core.row.RowMetaInterface outputRowMeta = getInputRowMeta().clone();
     logBasic("row:"+outputRowMeta);
     String sql = db.getDDL(tablename, data.inputRowMeta);//${TABLENAME}
     //String sql = db.getDDL(tablename, outputRowMeta);//${TABLENAME}
     db.execStatement(sql.replace(";", ""));

     logBasic(sql);
    }
   }
   catch(Exception e)
   {   
    logError("创建表出现异常",e);
    
   }finally{
    db.disconnect();
   }
  }
  return false;
 }