Kettle实现数据库迁移

Kettle实现数据库迁移

需求：

作数据仓库时,须要将业务系统CRM抽取到数据仓库的缓冲层，业务系统使用的是SqlServer数据库，数据仓库的缓冲层使用的是mysql数据库，为实现数据库的迁移，即将SqlServer数据库中的全部表与数据迁移到Mysql。 

解决办法: kettle设计一整套流程实现，读取数据库中表->建立表->表数据抽取

实现过程：

整套流程分为：2个job，4个trans。使用到的Trans插件：表输入、字段选择、复制记录到结果、从结果获取记录、设置变量、java脚本、表输出。

一、表数据抽取做业:

做用:首先获取数据库中全部的表名称 而后调用子Job进行表的建立、数据抽取

2.表名称获取流程  

要迁移的源库表名称获取，并设置到结果集，为下面的job使用。 其中的表输入使用的是show tables，复制数据库中全部的表，也能够从表中或者excel中输入，实现更加小粒度的控制。


show tables 结果为Tables_in_数据库名称,和具体数据库有关，故须要更名

三、子做业: 实现单个表格的建立及抽取

四、表名称变量设置

 上一步的子转换 

五、入库表结构建立  

执行的Java代码以下

public boolean processRow(StepMetaInterface smi, StepDataInterface sdi) throws KettleException
{
    // First, get a row from the default input hop

	Object[] r = getRow();

	org.pentaho.di.core.database.DatabaseMeta dbmeta = null;
	System.out.println(123);
    System.out.println( getTrans().getRepository());
    System.out.println(456);
	java.util.List list = getTrans().getRepository().readDatabases();
	
	if(list != null && !list.isEmpty())
	{
		for(int i=0;i<list.size();i++)
		{
			dbmeta = (org.pentaho.di.core.database.DatabaseMeta)list.get(i);
    //test1 为数据库名称
			if("test1".equalsIgnoreCase(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)
			{
				String sql = db.getDDLCreationTable(tablename, data.inputRowMeta);//${TABLENAME}
							
				db.execStatement(sql.replace(";", ""));

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

六、表数据抽取流程

引用原文:

一、源表若存在有blob的表，会有问题，多是因为表输出没有指定字段的缘由

二、以上的操做使用的是仓库，kettle repo会报错
三、将原文中String sql = db.getDDL(tablename, data.inputRowMeta);函数名 getDDL 改成 getDDLCreationTable

四、去除了原文中建立表以前表输入一个操做，原文当有空表须要复制时候，会报错

参考:

原文地址:  用Kettle的一套流程完成对整个数据库迁移

data-integration\samples\jobs\process all tables   实现整个数据库的迁移，

代码下载

http://pan.baidu.com/s/1nt7LOj3