（最详细图文教程）利用Kettle进行数据的批量迁移

（最详细图文教程）利用Kettle进行数据的批量迁移，以及迁移遇到的问题MySql向Oracle导入

ShiSanQ

11010人浏览 · 2021-12-21 17:22:36

ShiSanQ · 2021-12-21 17:22:36 发布

（最详细图文教程）利用Kettle进行数据的批量迁移

一、引言

一、引言

第一次用Kettle工具进行数据迁移时就遇到了很多坑，因为网上很多教程又不太详细，所以在填坑的路上花费了太多时间，决定写一篇关于数据抽取的文章。希望可以帮到初次接触Kettle的你们。（一定要看解答部分）

二、功能

利用Kettle工具对数据库的数据进行一个批量迁移，下面我用MySql数据库向Oracle数据库导入进行演示

三、事前准备

将MySql对应的版本驱动包放入Kettle的lib目录下（因为不放的话Kettle无法连接到数据库。不知道驱动包是哪个版本的话可以根据MySql的版本来进行一个搜索）。
将Oracle对应的版本驱动包放入Kettle的lib目录下（Oracle安装目录有驱动包，在目录下找即可,找不到就乖乖去下载。以我为例：）
E:\Oracle\product\11.2.0\dbhome_1\jdbc\lib
全放进去就行（事实上只放一个就能连接，方便省事直接全拉进去）
创建Kettle资源库（下面解释原因）

四、操作步骤-建立转换（五个）

4.1、建立我们第一个转换（get tables_1)

友情提示
以下功能块名称直接复制粘贴到Kettle左上角搜索框更快喔

创建表输入

这里要注意不同数据库的查询语句是不一样的,查完最好预览一下。
MySql：

show tables  //mysql 查询数据库中有哪些数据表

Oracle:

select * from user_tab_comments  //oracle 查询本用户的表,视图等

Alt

创建Excel输出
这里注意要是文件名带后缀扩展名就不用再填了，没带的话还是要保留扩展名的。

到这里为止我们第一个转换就已经完成了，具体看图：

4.2、建立我们第二个转换（get excel_2)

创建Excel输入
这里注意的点很多
点击导航栏【文件】先点浏览选择好文件后，再点增加

点击导航栏【工作表】点击获取工作表名称

点击导航栏【字段】点击获取来自头部数据的字段
复制一下名称和类型下面要用
创建复制记录到结果
这里创建好就行啥也不用管

到这里我们第二个转换也就完成了
在这里插入图片描述

4.3、建立我们第三个转换（set path_3)

创建从结果获取记录
字段名称和类型我们上面已经复制好了依次填入即可

在这里插入图片描述

创建设置变量
点击获取字段获取即可，变量名默认或者自己设置一个都可以，但是要记住因为下面要用，变量活动类型照着我设置就行，有空也可以百度看看其他类型的作用。
温馨提示：不跟上一个对象建立连接无法获取字段的喔。

这样我们第三个转换就做好了。

4.3、建立我们第四个转换（run java_4）

创建表输入
注意
数据库的连接别选错，我们是mysql向oracle导入所以我们这里连接mysql，记得要选中替换SQL语句里的变量，SQL输入框的格式为:

select * from ${变量名}  //你上面设置的变量名

在这里插入图片描述

创建Java 代码
注意：代码里面有两处需要修改的地方，仔细看下面注释，先看图

这是要插入的代码

public boolean processRow(StepMetaInterface smi, StepDataInterface sdi) throws KettleException
{
   	Object[] r = getRow();
 
	//本地连接获取数据库元数据
	//org.pentaho.di.core.database.DatabaseMeta dbmeta = getTransMeta().findDatabase("target");
 
	//资源库连接获取数据库元数据
	org.pentaho.di.core.database.DatabaseMeta dbmeta = null;
	/**
	解释为什么要创建资源库
	这行代码是从资源库来连接数据库
	要是你不创建资源库那么它将无法读取到数据库内容
	从而报空指针异常，导致java代码无法使用
	*/
	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);
			/**
			以我为例
			我是从mysql向oracle导入数据库
			所以我这边填写oracle数据库连接名称
			*/
			if("odb".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();
 			//这里“vtable”是我设置的变量名，你用的时候要换成你的
			String tablename = getVariable("vtable");
 
			//在日志中显示创建表名
			logBasic("开始创建表：" + tablename);
			
			if(tablename!=null && tablename.trim().length()>0)
			{
								
				//获取表名称和输入行数据
				String sql =  db.getDDLCreationTable(tablename ,getInputRowMeta());
 
				db.execStatement(sql.replace(";", ""));
 
				//在日志中显示sql语句
				logBasic(sql);
			}
		}
		catch(Exception e)
		{			
			logError("创建表出现异常",e);
			
		}finally{
			db.disconnect();
		}
	}
	return false;
}

到这里我们第四个转换也已经完成
在这里插入图片描述

4.3、建立我们第五个转换（set tablesdata_5）

创建表输入
注意：我是从mysql向oracle导入所以这里的表输入连接的数据库是mysql，表输入的具体步骤跟4.3一样，我就不在讲了。
创建表输出
注意：我是从mysql向oracle导入所以这里的表输入连接的数据库是oracle，目标模式填你要导入的那个模式，目标表就是那个变量名，其他不用管。

${变量名}  //括号是你设置的变量名

在这里插入图片描述
到此我们五个转换就已经完成了。

五、操作步骤-建立作业（两个）

5.1、建立我们第一个作业（job1）

很简单，创建一个Start和三个转换，以及一个成功，双击转换，分别引入上面我们创建的第三个，第四个，第五个，转换。具体看图：
在这里插入图片描述

5.2、建立我们第二个作业（job2）

这个更简单，一个Start,两个转换，一个作业，一个成功。转换选择，第一个和第二个。作业选择我们刚刚创建的job1,注意引入的时候记得勾选执行每一个输入行，不勾选会报错。
在这里插入图片描述
接下来看我们整个的流程图，到这里我们的一个批量迁移操作就完成了，大家按我的步骤来，要细心，程序可以运行成功的。