使用Kettle进行数据库迁移操作
使用Kettle进行数据库迁移操作
影非弦 发表于6个月前
使用Kettle进行数据库迁移操作
  • 发表于 6个月前
  • 阅读 11
  • 收藏 0
  • 点赞 0
  • 评论 0

标题:腾讯云 新注册用户域名抢购1元起>>>   

思路:先从源库读取需要迁移的表名,将表明作为变量,并在目标库创建这些表,最后表到表抽取数据

整个迁移流程包括2个job,4个transform。

具体如下:

1.总的Job:

2.获取表名流程的转换:

3.子job表数据抽取作业

3.1转换 表名变量设置

 

3.2转换 创建表结构

表输入如下,注意:这里的查询必须要查询出数据才行,不然后面会无法获取表结构来创建表

(有些人说这里加个条件where 1=2,不需要数据只需要表结构,我自己测试,发现这样无法获取到数据,后面的java脚本中可以打印这些信息,加了条件会报错,打印信息为null)

java脚本:


    public boolean processRow(StepMetaInterface smi, StepDataInterface sdi) throws KettleException
    {
    	// First, get a row from the default input hop
    	//
    	Object[] r = getRow();
		//logBasic("r.size="+r.length);
   		org.pentaho.di.core.database.DatabaseMeta dbmeta = null;
    	java.util.List list = getTrans().getRepository().readDatabases();//3.x中获取资源库的所有数据库连接信息用getDatabases();
    	if(list != null && !list.isEmpty())
    	{
    		for(int i=0;i<list.size();i++)
    		{
    			dbmeta = (org.pentaho.di.core.database.DatabaseMeta)list.get(i);
    			//下面是目标库的数据库连接,大家可根据需要修改
    			if("testb".equalsIgnoreCase(dbmeta.getName()))
    			{
					logBasic("数据库连接名为:"+dbmeta.getName());
    				break;
    			}
    		}
    	}
    	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)
    			{
					logBasic("data.inputRowMeta="+data.inputRowMeta);
    				String sql = db.getDDL(tablename, data.inputRowMeta);//${TABLENAME}
					logBasic("sql="+sql);
    				db.execStatement(sql.replace(";", ""));
    				logBasic(sql);
    			}
    		}catch(Exception e){
    			logError("创建表出现异常",e);
    		}finally{
    			db.disconnect();
    		}
    	}
    	return false;
	} 

3.3数据抽取

运行结果:

2017/07/24 15:16:25 - Spoon - 正在开始任务...
2017/07/24 15:16:25 - 数据库迁移作业 - 开始执行任务
2017/07/24 15:16:25 - 数据库迁移作业 - 开始项[获取表名流程]
2017/07/24 15:16:25 - 源表名称获取 - 为了转换解除补丁开始  [源表名称获取]
2017/07/24 15:16:25 - 表输入.0 - Finished reading query, closing connection.
2017/07/24 15:16:25 - 表输入.0 - 完成处理 (I=2, O=0, R=0, W=2, U=0, E=0)
2017/07/24 15:16:25 - 字段选择.0 - 完成处理 (I=0, O=0, R=2, W=2, U=0, E=0)
2017/07/24 15:16:25 - 复制记录到结果.0 - 完成处理 (I=0, O=0, R=2, W=2, U=0, E=0)
2017/07/24 15:16:25 - 数据库迁移作业 - 开始项[表数据抽取作业]
2017/07/24 15:16:25 - 表数据抽取子作业 - 开始项[表明成变量设置]
2017/07/24 15:16:25 - 将记录中的表名设置为变量 - 为了转换解除补丁开始  [将记录中的表名设置为变量]
2017/07/24 15:16:25 - 从结果获取记录.0 - 完成处理 (I=0, O=0, R=1, W=1, U=0, E=0)
2017/07/24 15:16:25 - 设置变量.0 - Setting environment variables...
2017/07/24 15:16:25 - 设置变量.0 - Set variable TABLENAME to value [TESTM]
2017/07/24 15:16:25 - 设置变量.0 - Finished after 1 rows.
2017/07/24 15:16:25 - 设置变量.0 - 完成处理 (I=0, O=0, R=1, W=1, U=0, E=0)
2017/07/24 15:16:25 - 表数据抽取子作业 - 开始项[创建表结构]
2017/07/24 15:16:25 - 创建表结构 - 为了转换解除补丁开始  [创建表结构]
2017/07/24 15:16:26 - 表输入.0 - Finished reading query, closing connection.
2017/07/24 15:16:26 - 创建入库表结构.0 - 数据库连接名为:testb
2017/07/24 15:16:26 - 表输入.0 - 完成处理 (I=1, O=0, R=0, W=1, U=0, E=0)
2017/07/24 15:16:26 - 创建入库表结构.0 - 开始创建表:TESTM
2017/07/24 15:16:26 - 创建入库表结构.0 - data.inputRowMeta=[ID String(10)], [NAME String(200)], [AGE BigNumber], [ADDRESS String(200)]
2017/07/24 15:16:26 - 创建入库表结构.0 - sql=CREATE TABLE TESTM
2017/07/24 15:16:26 - 创建入库表结构.0 - (
2017/07/24 15:16:26 - 创建入库表结构.0 -   ID VARCHAR2(10)
2017/07/24 15:16:26 - 创建入库表结构.0 - , NAME VARCHAR2(200)
2017/07/24 15:16:26 - 创建入库表结构.0 - , AGE NUMBER
2017/07/24 15:16:26 - 创建入库表结构.0 - , ADDRESS VARCHAR2(200)
2017/07/24 15:16:26 - 创建入库表结构.0 - )
2017/07/24 15:16:26 - 创建入库表结构.0 - ;
2017/07/24 15:16:26 - 创建入库表结构.0 - CREATE TABLE TESTM
2017/07/24 15:16:26 - 创建入库表结构.0 - (
2017/07/24 15:16:26 - 创建入库表结构.0 -   ID VARCHAR2(10)
2017/07/24 15:16:26 - 创建入库表结构.0 - , NAME VARCHAR2(200)
2017/07/24 15:16:26 - 创建入库表结构.0 - , AGE NUMBER
2017/07/24 15:16:26 - 创建入库表结构.0 - , ADDRESS VARCHAR2(200)
2017/07/24 15:16:26 - 创建入库表结构.0 - )
2017/07/24 15:16:26 - 创建入库表结构.0 - ;
2017/07/24 15:16:26 - 创建入库表结构.0 - 完成处理 (I=0, O=0, R=1, W=0, U=0, E=0)
2017/07/24 15:16:26 - 表数据抽取子作业 - 开始项[数据抽取]
2017/07/24 15:16:26 - 数据迁移 - 为了转换解除补丁开始  [数据迁移]
2017/07/24 15:16:26 - 表输出.0 - Connected to database [testb] (commit=1000)
2017/07/24 15:16:26 - 表输入.0 - Finished reading query, closing connection.
2017/07/24 15:16:26 - 表输入.0 - 完成处理 (I=3, O=0, R=0, W=3, U=0, E=0)
2017/07/24 15:16:26 - 表输出.0 - 完成处理 (I=0, O=3, R=3, W=3, U=0, E=0)
2017/07/24 15:16:26 - 表数据抽取子作业 - 完成作业项[数据抽取] (结果=[true])
2017/07/24 15:16:26 - 表数据抽取子作业 - 完成作业项[创建表结构] (结果=[true])
2017/07/24 15:16:26 - 表数据抽取子作业 - 完成作业项[表明成变量设置] (结果=[true])
2017/07/24 15:16:26 - 表数据抽取子作业 - 开始项[表明成变量设置]
2017/07/24 15:16:26 - 将记录中的表名设置为变量 - 为了转换解除补丁开始  [将记录中的表名设置为变量]
2017/07/24 15:16:26 - 从结果获取记录.0 - 完成处理 (I=0, O=0, R=1, W=1, U=0, E=0)
2017/07/24 15:16:26 - 设置变量.0 - Setting environment variables...
2017/07/24 15:16:26 - 设置变量.0 - Set variable TABLENAME to value [TESTN]
2017/07/24 15:16:26 - 设置变量.0 - Finished after 1 rows.
2017/07/24 15:16:26 - 设置变量.0 - 完成处理 (I=0, O=0, R=1, W=1, U=0, E=0)
2017/07/24 15:16:26 - 表数据抽取子作业 - 开始项[创建表结构]
2017/07/24 15:16:26 - 创建表结构 - 为了转换解除补丁开始  [创建表结构]
2017/07/24 15:16:26 - 表输入.0 - Finished reading query, closing connection.
2017/07/24 15:16:26 - 表输入.0 - 完成处理 (I=1, O=0, R=0, W=1, U=0, E=0)
2017/07/24 15:16:26 - 创建入库表结构.0 - 数据库连接名为:testb
2017/07/24 15:16:26 - 创建入库表结构.0 - 开始创建表:TESTN
2017/07/24 15:16:26 - 创建入库表结构.0 - data.inputRowMeta=[ID String(10)], [NAME String(200)], [AGE BigNumber], [ADDRESS String(200)]
2017/07/24 15:16:26 - 创建入库表结构.0 - sql=CREATE TABLE TESTN
2017/07/24 15:16:26 - 创建入库表结构.0 - (
2017/07/24 15:16:26 - 创建入库表结构.0 -   ID VARCHAR2(10)
2017/07/24 15:16:26 - 创建入库表结构.0 - , NAME VARCHAR2(200)
2017/07/24 15:16:26 - 创建入库表结构.0 - , AGE NUMBER
2017/07/24 15:16:26 - 创建入库表结构.0 - , ADDRESS VARCHAR2(200)
2017/07/24 15:16:26 - 创建入库表结构.0 - )
2017/07/24 15:16:26 - 创建入库表结构.0 - ;
2017/07/24 15:16:26 - 创建入库表结构.0 - CREATE TABLE TESTN
2017/07/24 15:16:26 - 创建入库表结构.0 - (
2017/07/24 15:16:26 - 创建入库表结构.0 -   ID VARCHAR2(10)
2017/07/24 15:16:26 - 创建入库表结构.0 - , NAME VARCHAR2(200)
2017/07/24 15:16:26 - 创建入库表结构.0 - , AGE NUMBER
2017/07/24 15:16:26 - 创建入库表结构.0 - , ADDRESS VARCHAR2(200)
2017/07/24 15:16:26 - 创建入库表结构.0 - )
2017/07/24 15:16:26 - 创建入库表结构.0 - ;
2017/07/24 15:16:26 - 创建入库表结构.0 - 完成处理 (I=0, O=0, R=1, W=0, U=0, E=0)
2017/07/24 15:16:26 - 表数据抽取子作业 - 开始项[数据抽取]
2017/07/24 15:16:26 - 数据迁移 - 为了转换解除补丁开始  [数据迁移]
2017/07/24 15:16:26 - 表输出.0 - Connected to database [testb] (commit=1000)
2017/07/24 15:16:26 - 表输入.0 - Finished reading query, closing connection.
2017/07/24 15:16:26 - 表输入.0 - 完成处理 (I=3, O=0, R=0, W=3, U=0, E=0)
2017/07/24 15:16:27 - 表输出.0 - 完成处理 (I=0, O=3, R=3, W=3, U=0, E=0)
2017/07/24 15:16:27 - 表数据抽取子作业 - 完成作业项[数据抽取] (结果=[true])
2017/07/24 15:16:27 - 表数据抽取子作业 - 完成作业项[创建表结构] (结果=[true])
2017/07/24 15:16:27 - 表数据抽取子作业 - 完成作业项[表明成变量设置] (结果=[true])
2017/07/24 15:16:27 - 数据库迁移作业 - 开始项[成功]
2017/07/24 15:16:27 - 数据库迁移作业 - 完成作业项[成功] (结果=[true])
2017/07/24 15:16:27 - 数据库迁移作业 - 完成作业项[表数据抽取作业] (结果=[true])
2017/07/24 15:16:27 - 数据库迁移作业 - 完成作业项[获取表名流程] (结果=[true])
2017/07/24 15:16:27 - 数据库迁移作业 - 任务执行完毕
2017/07/24 15:16:27 - Spoon - 任务已经结束.

 

共有 人打赏支持
粉丝 4
博文 5
码字总数 3621
×
影非弦
如果觉得我的文章对您有用,请随意打赏。您的支持将鼓励我继续创作!
* 金额(元)
¥1 ¥5 ¥10 ¥20 其他金额
打赏人
留言
* 支付类型
微信扫码支付
打赏金额:
已支付成功
打赏金额: