用途和背景 @H_404_6@
ETL (extract,transform,load)工具是用于数据库数据迁移清洗处理等操作的工具。
我们有时会面临一种情况:实时运行的系统由于需求的变化而去更新DB的结构,这样便需要收集系统实时运行的数据,然后将其传入到新的DB结构中去,以确保实时运行的数据不会丢失。当然我们可以通过sql(在没有bytea字段的情况下)或JDBC去完成整个流程,但是这样做远远不如ETL高效。
@H_404_6@
在Kettle中,在处理Postgresql数据时候,我们目前发现2个会对工作产生影响的Bug。其一是对bytea数据的处理,其二是对boolean数据的处理。我们可以简单对它的源码进行修改,使它可以运行起来。
@H_404_6@
修改了org.pentaho.di.core.database.Database.java中关于Blob的处理
@H_404_6@
更改前为:
@H_404_6@
case ValueMetaInterface.TYPE_BINARY :
@H_404_6@
{
@H_404_6@
if (databaseMeta.supportsGetBlob()){
@H_404_6@
Blob blob = rs.getBlob(i+1);
@H_404_6@
//这里连接数据库为postgresql时,会抛出异常 @H_404_6@
if (blob!=null){
@H_404_6@
data[i] = blob.getBytes(
1L
,(int)blob.length());
@H_404_6@
}else{
@H_404_6@
data[i] = null;
@H_404_6@
}
@H_404_6@
}
@H_404_6@
else{
@H_404_6@
data[i] = rs.getBytes(i+1);
@H_404_6@
}
@H_404_6@
}
@H_404_6@
break;
@H_404_6@
更改后为:
@H_404_6@
case ValueMetaInterface.TYPE_BINARY :
@H_404_6@
{
@H_404_6@
if (databaseMeta.supportsGetBlob()){
@H_404_6@
try{
@H_404_6@
Blob blob = rs.getBlob(i+1);
@H_404_6@
if (blob!=null){
@H_404_6@
data[i] = blob.getBytes(
1L
,(int)blob.length());
@H_404_6@
}else{
@H_404_6@
data[i] = null;
@H_404_6@
}
@H_404_6@
}catch(Exception ex){
@H_404_6@
try{
@H_404_6@
data[i] = rs.getBytes(i+1);
@H_404_6@
}catch(Exception e){
@H_404_6@
data[i] = null;
@H_404_6@
}
@H_404_6@
}
@H_404_6@
}else{
@H_404_6@
data[i] = rs.getBytes(i+1);
@H_404_6@
}
@H_404_6@
}
@H_404_6@
break;
@H_404_6@
@H_404_6@
修改了org.pentaho.di.core.database. PostgresqlDatabaseMeta.java中关于Boolean类型的处理,在默认情况下即PostgresqlDatabaseMeta 的基类BaseDatabaseMeta中是默认不支持Boolean数据类型的, supportsBooleanDataType返回值是false,我们在PostgresqlDatabaseMeta中overwrite这个函数以实现Kettle对PostGresql中关于Boolean数据类型的支持。
public boolean supportsBooleanDataType() {
return true;
}
@H_404_6@
修改这两处后,利用ant进行重新build,就可以生成全新的Kettle。 但是在修改二中这样做又引入了新的Bug,即不能将Kettle的repository创建到Postgresql中,这样也就意味着我们不能以Postgresql数据库形式共享项目文件。但是这样做的好处是极大的方便我们日常的工作,可以方便的进行数据迁移清洗等操作。当我们需要进行项目共享的时候,我们可以通过CVS等方式实现repository共享。
@H_404_6@
基本概念 @H_404_6@
我们常使用Spoon(是Kettle中的一个子项目)进行数据transfer,这里只是自己对下面一些基本概念的理解,具体请自行参见Spoon的文档。
@H_404_6@
Transform:定义了数据transfer的基本操作流程,通常包括Input, Output和Hop
Job:定义了一组数据transfer的操作流程,可以包括多个Job或transform。
Input:定义了数据流的起点
Output:定义了数据流的终点
Step:是transform的基本构成点,Input和Output都是一种step
DB Connection:定义了数据库连接的名字,通常和Input以及Output配合使用
Start:定义了Job中transfer的起点
Hop:表示Step间数据的流向
sql脚本:JOB中多个Transform或子JOB之间表示需要执行的sql脚本
以上仅仅是罗列了最基本最常用的Step, 在Spoon中还有多种Step可供使用以方便Job的设计和使用,具体的可以参见Spoon文档。
@H_404_6@
A.新建一个Transform @H_404_6@
一、打开Spoon,选择没有资源库 @H_404_6@
二、新建transform @H_404_6@
三、新建 DB Connection @H_404_6@
四、新建 input和output @H_404_6@
五、建立input和output之间的数据流Hop @H_404_6@
六、保存Transform @H_404_6@
@H_404_6@
一个建立好的transform如图所示
@H_404_6@
B.新建一个子JOB @H_404_6@
一、打开Spoon,选择没有资源库 @H_404_6@
二、新建JOB @H_404_6@
三、新建DB Connection @H_404_6@
四、新建Start节点 @H_404_6@
五、新建sql脚本,设置要执行的脚本,选择必要的DB Connection @H_404_6@
六、新建Transform节点,设置已经存在的Transform @H_404_6@
七、建立多个Step之间的Hop @H_404_6@
八、保存Job @H_404_6@
一个建立好的子JOB如图所示
@H_404_6@
C、新建JOB @H_404_6@
一、打开Spoon,选择没有资源库 @H_404_6@
二、新建JOB @H_404_6@
三、新建Start节点 @H_404_6@
四、新建JOB节点,设置已经存在的JOB @H_404_6@
五、建立多个Step之间的Hop @H_404_6@
六、保存Job