kettle培训手册

1、Kettle简介： Kettle是使用Java语言开发的一个开源ETL工具(Extract-Transform-Loading)，完成数据的抽取、转换、加载工作。Kettle的意思是“水壶”，把各种各样的数据倒进水壶中，经过转换、处理，以我们所需的格式输出对我们有价值的信息。问题：为什么要用kettle？需求：不同的数据库(同一种类型但实例不同的数据库，或者不同类型的数据库)之间、不同的数据存储形式(数据库、文本文件)之间需要协同工作，计算、处理、转换成我们需要的数据。实际应用场景：对帐系统需要读取ATM的交易流水文件到Oracle数据库中并进行统计。不使用kettle的情况：专门编写程序(J

2、ava或C)导入到数据库中，在Oracle中编写存储过程进行批量统计。使用kettle的情况：通过“文本文件输入”step，指定分隔符，即可完成导入，无需编写程序；使用kettle中的各种对象可以完成统计功能，不需要编写存储过程。两种情况对比：情况一：需要编写专门的导入程序，不具备通用性。存储过程在不同类型的数据库中语法不同，也不具备通用性。但不需要掌握kettle。情况二：不需要编写专门的导入程序，不同的文件格式只需要修改“文本文件输入”step中的对应参数即可。不需要针对不同的数据库编写不同版本的存储过程，只需要使用标准SQL就可以具备通用性。需要掌握kettle。好处：跨数据库、跨平台，

3、减少我们的工作量。Kettle中有两种类型的对象，分别对应两种类型的文件。一种是Transformation(以后简称Trans)，即转换；一种是Job，即工作。Trans完成具体的任务，Job用于控制转换之间的执行顺序。第一部分：与数据库打交道1创建数据库连接<1>新建一个Transformation，在“DB连接”下创建一个“数据库连接”，如下图所示：<2>如果创建kettle不支持的数据库的连接，ConnectionType选择Generic database。2共享连接可以在不同的Transformation中共享，可以减少数据库配置工作。右键单击需要共享的连接

4、，选择“Share”选项，如下图所示：3用变量保存参数，在不同的环境下切换时会省心省力。设置环境变量(编辑->设置环境变量,Ctrl+Alt+J)，如下图所示：在需要使用变量的地方通过$变量名读取，参考“创建数据库连接”。4从数据库中读取数据。(加工数据需要先读取数据)<1>不带参数读取数据。表输入(table Input)，获取SQL查询语句，针对数据表的各种Actions。<2>带参数读取数据。两种参数方式：第一种：占位符(?)规则：关键字(例如：table、index)，标识符(表名、索引名)不能通过这种占位符(?)参数传递。可以这样传递参数，但所有的参数类

5、型必须一样。这样传递参数时，参数类型可以不同。例子为：注意：带参数查询在“表输入”中有时不能预览，需要运行转换去验证是否正确。执行多次SQL语句，但条件取值不同，例如如下条件：第二种：用变量名，例如：$ORDER_COLUMN。关键字(例如：table、index)，标识符，整条SQL 语句都可以用变量替换，但使用时要小心。在运行时才确定SQL语句：设置环境变量(Ctrl+Alt+J)，输入ORDER_COLUMN及其值，如下：定义一个“表输入”，在“SQL”中输入带参数的SQL语句如下：$ORDER_COLUMN就是刚才设定的环变量。选中“替换SQL语句里的变量”复选框。相当于执行SELEC

6、T I_USERID, C_USERLOGINNAME, C_USERLOGINPWD, C_USERNAME, I_POSITIONID, D_CREATETIME, D_LASTLOGINTIME, I_LOGINCOUNT, I_ENABLED, C_MEMO FROM SYS_USER ORDER BY i_userid注意：这里只替换环境变量，不替换占位符变量(?)。定义输出到文件，如下：在同一条SQL语句中，同一个变量可以多次使用，如下：5.Insert/update(1)文本文件输入：选中“文件”选项卡,选择指定的文本文件，点击“增加”按钮，把选中的文本文件添加到“选中的文件”列

7、表框中：点击“显示文件内容”，如下：选中“内容”选项卡，把分隔符改为“,”(分号)，如下所示：选中“字段”选项卡，并点击“获取字段”，如下所示：把所有字段的数据类型改为String。(2) 创建一个"Insert/Update"，双击，指定数据库连接、目标模式、数据表、用来查询的关键字(先点击“获取字段”)、更新字段(先点击“获取和更新字段”)，如下所示：按“F9”运行。"插入/更新"操作实现逻辑：先根据关键字做判断，如果不存在，则执行插入操作；如果已经存在，再判断是否存在差异，存在则更新，不存在则不更新。If(关键字取值是否存在 = true)If(需

8、要更新的字段值是否存在差异 = ture) 执行update操作 /否则不执行update操作 else /关键字取值不存在执行insert操作技巧：(1)如果只是插入数据则用“表输出”，“表输出”比"插入/更新"快，因为不需要判断。(2)如果只是更新数据则用“更新”，“更新”比"插入/更新"快，因为不需要判断。6.用“表输出”、“更新step”替换“插入/更新step”，速度比“插入/更新step”要快。问题：为什么会快？答：不用判断要更新的记录每个字段是否有变化，而是直接更新。注意事项：(1) 文本文件输入注意分隔符、数据类型。(2) 表输出选中“S

9、pecify database fields”，指定输出字段。(3)“表输出”指定查询关键字、更新字段；选中“使用批量插入”，要注要Commit size。如果只插入2条记录，但Commit size是10则执行时会出错；问题：是否有方法强制最后提交？“更新”选中“Use batch updates”，要注要Commit size。如果只更新2条记录，但Commit size是10，则不会提交。问题：是否有方法强制最后提交？(4) 数据表必须有唯一主键。作为异常判断条件。7.Combination lookup/update创建一个“文本文件输入”，双击进入编辑页面，在“文件”选项卡中指定文本

10、文件并添加到“选中的文件”列表中，如下图所示：选中“内容”选项卡，指定“分隔符”为“;”(分号)，如下图所示：选中“字段”选项卡，点击“获取字段”按钮获取字段。注意：如果分隔符选择错误，则这里获取不到正确的结果。创建“联合更新/查询”(英文是Combination lookup/update，在“数据仓库”分类中)。指定数据库连接为pdiTest，目标表为“offices”，点击“获字字段”，这里是指要插入到目标的字段。其中“维度字段”是指目标表的字段，“在流里面的字段”是指文本文件中的字段。在“代理关键字”中填写主键字段officeCode，代理键的创建选择“使用表里的最大值 + 1”。如下

11、图所示：也可以使用sequence自动生成主键值(先创建offices_seq)，如下图所示：create sequence offices_seqstart with 100increment by 1;可以只插入部分字段，在“关键字段”列表中选中不需要插入的字段并删除即可，如下图所示：8.通过已存在的主键值确定新记录的主键值。例如:主键的取值形式为A00001、A00002、A00003、A00004。(1)创建表输出step，并在SQL脚本处填写以下脚本，读取最大的author_id值，如下图所示：(2) 创建文本文件输出step，添加authors.txt文本文件，指定分隔符为“|”，

12、如下图所示：(3)创建“记录关联 (笛卡尔输出)”(在连接分类中，英文叫Join)，把前2个step指向此step，如下图所示：(4)创建“增加序列”step(在转换分类中，英文叫"Add sequence")，指定步骤名称、值的名称、起始值、增加根据，如下图所示：(5)创建“转换->计算器”，进如下图配置：(6)创建“表输出”step，如下图所示:整个转换如下图所示:演示。9.Delete data from a table。目标：带条件删除记录，部分条件在excel文件中。(1)从Excel文件读取要删除的产品分类名称。创建“Excel输入”step并指定包括要删

13、除产品分类名称的Excel文件，获取Excel文件的字段。如下图所示：(2)创建"数据库查询"step(在查询分类中)，选择产品分类表categories，指定查询所需的关键字：category = category，查询表返回的值id_category。如下图所示：这里相当于执行select id_category from categories where category = category(从Excel文件中读取)。(3)读取变量到流中(因为第4步的delete操作只能从流中读取数据)。创建“Get Variables”step(在作业分类中)。如下图所示：(4)

14、进行删除操作，删除的条件是price <= 50 and category in (excel中的category)。创建“删除”step(在输出分类中)，指定删除条件，如下图所示：执行结果：删除了category为tents和tools的所有price<=50产品。通过比较如下SQL语句在执行上面转换前后的结果可以知道其差异。select c.category, count(*) from products p, categories c where p.id_category = c.id_category and p.price <=50 group by c.cate

15、gory order by category转换执行之前：转换执行之后：整个转换如下图所示：10.Creating or altering a database table from PDI(design time)。通过XML文件创建/修改表结构，并插入数据。11.Create or altering a database table from PDI(runtime)。12.inserting, deleting or updating a table based on a field.整体流程图：(1)创建“文本文件输入”step读取book_news.txt数据。具体设置如下：分隔符用

16、“|”，如下图：(2) 创建“转换->拆分字段”step，拆分更新记录中的author字段为lastname和firstname字段。具体设置如下：1分隔符为","。2去除空格类型为“去掉左右两端空格”。(3) 创建“查询->数据库查询”step，目的是根据lastname和firstname读取authors表中符合条件的作者的id_author。具体设置如下所示：(4)创建“输出->Synchronize after merge”step，目的是：在books表中找出符合条件的记录，comment为NEW的记录进行插入操作，comment为In Sto

17、ck的记录更新其price字段值。具体设置如下： 更新时需要id_title字段，插入不需要用id_title字段。一共10条记录：<1>更新了price的记录：select * From books where id_title in('123-400', '123-401','123-402')3条记录。<2>新插入的记录：select * from books where id_title in ('223-655','223-701','323-609','5

18、23-110')<3>没更新的记录：select * From books where id_title in('123-602', '523-111', '523-112')13.changing_the_connection_database_at_run-time应用场景：统计表结构相同但不同物理数据库的数据并汇总，例如：同一套系统，但部署在不同的分支机构中，需要对其进行一个总的统计汇总。主体流程图：主流程(Job)转换一(Transformation)转换二(Transformation)(1)创建第一个转换“读取数据

19、库连接信息”。(2)创建一个“文本文件输入”step，对其进行如下设置：选中保存数据库连接信息的文件。保存数据库连接信息文件：(3) 创建“作业->复制记录到结果”，把上一个step指到这个step，并保存。(4) 创建第二个转换“汇总销售量”。(5) 创建“表输入”step，在SQL脚本中输入统计销售量的SQL语句，如下图所示：创建数据库连接，如下图所示：这里有几个取值都来自变量。Host Name:$HOST_NAMEDatabase Name:$DATABASE_NAMEUser Name:$USER_NAMEPassword:$PASS_WORD(6) 创建“作业->Get

20、 Variables”step，读取第一个转换传递过来的%BRANCH变量，其配置如下图所示：(7) 创建“输出->文本文件输出”step，把各个数据库统计出来的销售量汇总输出，其配置如下图所示：指定输出文件的路径和名称。选中“追加方式”复选框，指定“分隔符”。指定输出的数据，包括branch和sales(第1个step中的统计结果)关键点：定义转换内部变量，接收另第一个转换传递过来的变量。Ctrl+T(编辑->设置.)->选择“Parameters”选项卡，如下图所示：(8)定义一个工作(Job)“主流程”，在其中创建一个“通用->START”step和两个“通用-&

21、gt;Transformation”，并指定其分别对应第一、第二个转换。如下图所示：(9)运行主流程Job，观察生成的文本文件的数据是否正确。13. creating_parent_child_tables(导入包含父子关系的记录到数据表)需求：员工数据保存在文本文件中，员工数据有上下级关系(通过rep_to建立)，需要把所有员工数据导入employees表中，并建立上下级关系。思路：<1>创建一个Transformation，读取总经理记录(最顶层员工)，并插入到employees表。<2>创建一个Transformation，分别读取各级员工记录(这个例子中有3级，分别是副总、销售经理、销售代表)，并插入到employees表。<3>创建一个Job，把上面的Transformation串联起来，其中第一个Transformation只需要执行1次，第二个Transformation需要执行3次，但需要分别传入不同的参数值(level，区分不同的用户级别)。实现步骤：(1)创建一个Transformation(“插入总经理记录”)，创建一个“文本文件输入”step，分隔符用“|”，如下所示：(2)创建“过滤记录”step，具体设置如下：(3)创建“表输出”step，具体如下图所示：|(4)创建另一个Transforma