KETTLE表的批量处理

批量生成表结构
- 一使用kettle 自带功能
- 二自制批量建表job 研究中
">
批量抽取数据
- 获取表名步骤
- 数据抽取步骤
  - 批量抽取表.rar">批量抽取表.rar

批量生成表结构

一使用kettle 自带功能

在工具向导里面，找到复制多表向导

选择对应输入输出的连接

选择需要抽取的表

输入job名称及连接地址

kettle可以自动检测在目标表中没有的表，并创建好对应转换

二自制批量建表job 研究中

public boolean processRow(StepMetaInterface smi, StepDataInterface sdi) throws KettleException {
  if (first) {
    first = false;
    /* TODO: Your code here. (Using info fields)
    FieldHelper infoField = get(Fields.Info, "info_field_name");
    RowSet infoStream = findInfoRowSet("info_stream_tag");
    Object[] infoRow = null;
    int infoRowCount = 0;
    // Read all rows from info step before calling getRow() method, which returns first row from any
    // input rowset. As rowMeta for info and input steps varies getRow() can lead to errors.
    while((infoRow = getRowFrom(infoStream)) != null){
      // do something with info data
      infoRowCount++;
    }
    */
  }
Object[] r = getRow();
//本地连接获取目标数据库连接
org.pentaho.di.core.database.DatabaseMeta dbmeta = getTransMeta().findDatabase("B_Mysql_1");
if(dbmeta!=null)
{
    org.pentaho.di.core.database.Database db=new org.pentaho.di.core.database.Database(dbmeta);
    try
    {
        db.connect();
        //String tablename = getVariable("table_name");
        String tablename = 'bpc';
        //在日志中显示创建表名
        logBasic("开始创建表：" + tablename);
        if(tablename!=null && tablename.trim().length()>0)
        {
            logBasic("123");
            String sql = db.getDDLCreationTable(tablename,data.inputRowMeta );//${TABLENAME}
            // 目前错误在这里，找不到表结构
            logBasic("124");
            //sql = sql.replace("TABLE", "TABLE public.");
            db.execStatement(sql.replace(";", ""));
            logBasic(sql);
        }
    }
    catch(Exception e)
    {
        logError("创建表出现异常",e);
    }finally{
        db.disconnect();
    }
}
return false;
}