Kettle使用文档.docx

1、Kettle使用文档1.Kettle简介Kettle是一款国外开源的ETL工具，纯java编写，可以在Window、Linux、Unix上运行，数据抽取高效稳定。Kettle 中文名称叫水壶，该项目的主程序员MATT 希望把各种数据放到一个壶里，然后以一种指定的格式流出。Kettle主要由资源库、数据库、作业（job）、转换、步骤五部分组成，资源库是用来存储用户所编写的作业和转换（即kjb文件和ktr文件）一般是在数据库中存储，方便用户的查找和使用。数据库就是处理数据是需要的数据库。作业是用来确定一个工程中需要使用的转换和转换的执行顺序。转换是数据在ktr文件中的具体转换过程，类似于Java的

2、一个方法，而作业就类似于java的一个类，它可以调用各种不同的方法（转换）。2.Kettle框架搭建（1）下载安装可以从http:/kettle.pentaho.org下载最新版的 Kettle软件 ，同时，Kettle 是绿色软件，下载后，解压到任意目录即可。（2）配置环境变量使用Kettle前提是配置好Java的环境变量,因为Kettle是java编写，需要本地的JVM的运行环境。配置Java环境变量可参考:配置Kettle环境变量步骤：一、在系统的环境变量中添加KETTLE_HOME变量，目录指向 kettle的安装目录：D:kettledata-integration（具体以安装路径

3、为准）二、新建系统变量：KETTLE_HOME变量值： D:kettledata-integration（具体以安装路径为准， Kettle的解压路径，直到Kettle.exe所在目录）三、选择PATH添加环境变量：变量名：PATH变量值：%KETTLE_HOME%;（3）Kettle工具的运行在Windows系统下运行，只需要解压kettle文件后，双击data-integration文件夹中的Spoon.bat文件在Linux下运行则双击data-integration文件夹中的Spoon.sh文件3.Kettle的基本概念（1）作业（job）负责将【转换】组织在一起进而完成某一块工作，通

4、常我们需要把一个大的任务分解成几个逻辑上隔离的作业，当这几个作业都完成了，也就说明这项任务完成了。1.Job Entry：一个Job Entry 是一个任务的一部分，它执行某些内容。2.Hop：一个Hop 代表两个步骤之间的一个或者多个数据流。一个Hop 总是代表着两个Job Entry 之间的连接，并且能够被原始的Job Entry 设置，无条件的执行下一个Job Entry，直到执行成功或者失败。3.Note：一个Note 是一个任务附加的文本注释信息。1）创建作业2）START控件核心对象-通用-START中选取START是一个job的任务入口，只有无条件的任务可以从此入 口开始任务，也

5、可以在此设置任务定时运行和重复运行。3）成功控件核心对象-通用-成功 中选取代表job的结束，没有任何数据的操作4）转换核心对象-通用-转换 中选取是job中具体的数据转换操作，只要把之前写好的ktr文件路径设置上 去，就可以运行对应的ktr文件（2）转换（Transformation）定义对数据操作的容器，数据操作就是数据从输入到输出的一个过程，可以理解为比作业粒度更小一级的容器，我们将任务分解成作业，然后需要将作业分解成一个或多个转换，每个转换只完成一部分工作。1.Value：Value 是行的一部分，并且是包含以下类型的的数据：Strings、floating point Numbers

6、、unlimited precision BigNumbers、Integers、Dates、或者Boolean。2.Row：一行包含0 个或者多个Values。3.Output Stream：一个Output Stream 是离开一个步骤时的行的堆栈。4.Input Stream：一个Input Stream 是进入一个步骤时的行的堆栈。5.Step：转换的一个步骤，可以是一个Stream或是其他元素。6.Hop：一个Hop 代表两个步骤之间的一个或者多个数据流。一个Hop 总是代表着一个步骤的输出流和一个步骤的输入流。7.Note：一个Note 是一个转换附加的文本注释信息。1）创建转换

7、创建简单的转换，首先配置数据库连接，一个转换中可以创建多个数据库连接，而且是不同类型的数据库也都可以，方便从一个数据库中抽取数据到另一个数据库中2）表输入（数据来源）从核心组件树菜单中拖拽一个表输入组件，用于编写sql语句3）表输出（数据接收）从核心组件树菜单中拖拽一个表输出组件，用于将上一步骤传递的数据保存到数据库中4）获取系统信息双击后弹出页面如下图：点击类型弹出下图框如下图所示：点击选择所需要的信息即可5）字段选择双击打开控件：选择和修改：制定需要流向输出流的字段的精度顺序和名称。获取选择的字段：默认获取输入流中的字段。列映射：指的是字段名称=字段该名成之后的映射，如果选择了移除或者 元

8、数据，则 不能生成映射移除：指定从输出流删除的字段。获取移除的字段：默认获取所有的输入流字段元数据：修改元数据字段的名称、类型、长度、精度等。6）合并记录功能描述：合并两个数据流，并根据某个关键字排序。这两个数据被比较，以标识相等的、变更的、删除的和新建的记录。这个步骤允许你比较两个行流，如果你想在两个不同的时间比较数据，这个是比较有用的。两个行流被合并，一个旧数据源，一个是新数据源双击【合并记录】控件，根据表示字段flagfield的值判断相等、变更、 删除的和新建的记录。1. “identical” 旧数据和新数据一样2. “changed” 数据发生了变化;3. “new” 新数据中有而

9、旧数据中没有的记录4. “deleted” 旧数据中有而新数据中没有的记录关键字段：用于定位两个数据源中的同一条记录。比较字段：对于两个数据源中的同一条记录中，指定需要比较的字段。合并后的数据将包括旧数据来源和新数据来源里的所有数据，对于变化的数据，使用新数据代替旧数据，同时在结果里用一个标示字段，来指定新旧数据的比较结果。注意：旧数据和新数据需要事先按照关键字段排序。旧数据和新数据要有相同的字段名称。例子：旧数据： 新数据 合并后的数据field1, field2 field1, field2 field1; field2; flag1, 1 1, 1 1; 1; identical2, 2

10、 2, 9 2; 9; changed3, 3 5, 5 3; 3; deleted4, 4 4; 4; deleted5; 5; new4.Kettle使用打开kettle工具后会出现如下界面：（1）新建作业路径：文件新建作业。（2）新建2个连接 （mysql、oracle）路径：主对象树-作业-作业1-DB连接。连接名称:数据库连接名称（随意填写，不能为空）连接:连接的数据库类型连接方式:数据库的连接方式（本文档只选择了Native的连接方式）主机名称:数据库具体的Ip地址数据库名称：指定连接的数据库名称端口号：数据库监听的tcp/ip端口号用户名:指定连接数据库登录时的用户名密码：指定连

11、接数据库登录时的密码填写完数据库连接的基本信息后，点解界面中的测试按钮进行数据库连接测试，如成功连接会弹出如下信息，反之则会弹出出错信息，出错后根据相关的出错信息进行修改。Oracle数据库同理：注：需要把数据库连接需要的jar包添加到lib文件夹下。（1）Kettle实现数据库整库迁移一、采用软件自带方式 首先在目标数据库中先创建一个抽取记录表TIME_TABLE表内2个字段为： etl_time(抽取时间)table_name(抽取表名)1.新建一个job创建两个DB连接：mysql、ORCL（源数据库和目标数据库连接），在菜单中找到【复制多表向导】，点击进行相关操作：2.选择源数据库和目

12、标数据库3.选择所需迁移的表4.编辑生成的job文件名：qy.kjb，和文件目录，编辑好后+【Finish】点击Finish之后界面如下图所示，5.把当前抽取时间和表名记录到创建的TIME_TABLE表中，通过SQL脚本，核心对象-脚本-SQL，左键拖拽进主页面通过Shift将其连接起来在这个SQL脚本中填写insert语句，把当前时间和表名添加到数据库抽取记录表中6.点击运行按钮二、自定义数据库迁移（oracle mysql）1.创建一个主job：数据库迁移.kjb2.创建转换：获取表名称.ktr1) 创建一个DB连接：oraclesource（源数据库连接）2) 表输入注：oracle中获

13、取表名SQL:select * from user_tables，mysql中为：show tables3) 字段选择点击【获取选择的字段】，添加新的字段，选中TABLE_NAME并改名成“tablename”，将其余字段删除4) 添加一个【复制记录到结果字符串】3.新建一个job：抽取表.kjb1) 新建一个DB连接：mysqltarget4.新建一个转换：表名称变量设置.ktr1) 添加一个【从结果获取记录】2）添加一个【设置变量】5.添加一个【检查表是否存在】6.新建一个转换：创建目标库表结构.ktr1) 新建两个DB连接：oraclesource，mysqltarget创建过程同上2）

14、添加一个【表输入】3）添加一个【Java代码】，代码如下：1.public boolean processRow(StepMetaInterface smi, StepDataInterface sdi) throws KettleException2.3.4.5.Object r = getRow();6.7.8./本地连接获取数据库元数据9.10./org.pentaho.di.core.database.DatabaseMeta dbmeta = getTransMeta().findDatabase(target);11.12.13./资源库连接获取数据库元数据14.15.org.pe

15、ntaho.di.core.database.DatabaseMeta dbmeta = null;16.17.18.java.util.List list = getTrans().getRepository().readDatabases();19.20.21.if(list != null & !list.isEmpty()22.23.24.25.for(int i=0;i0)67.68.69.70.71./获取表名称和输入行数据72.73.String sql = db.getDDLCreationTable(tablename ,getInputRowMeta();74.75.76.

16、db.execStatement(sql.replace(;, );77.78.79./在日志中显示sql语句80.81.logBasic(sql);82.83.84.85.86.87.catch(Exception e)88.89. 90.91.logError(创建表出现异常,e);92.93.94.finally95.96.db.disconnect();97.98.99.100.101.102.return false;103.104.105.7.新建一个转换：表抽取.ktr1）新建两个DB连接：oraclesource，mysqltarget创建过程同上2） 添加一个【表输入】3）

17、添加一个表输出8.运行数据库迁移.kjb（2）Kettle增量抽取方案新建一个转换在核心对象中找到【表输入】【表输入2】【表输出】【阻塞数据】【执行SQL脚本】组件，并通过Shift连接在一起【表输入】：查询该表的最新的一次抽取时间【表输入2】：把上一步查询出来的最新抽取时间当做查询条件，查出要抽取的数据【表输出】把查询出来要抽取的数据，抽取到oracle数据库中【执行SQL语句】：修改成目标数据库抽取过来的时间最大的一条记录，把数据抽取记录表的时间修改以便下一次抽取表输入：旧数据来源的步骤（来源于mysql数据库）表输入2：新数据来源的步骤（来源于oracle数据库）合并记录：将输入源与目的

18、源的每个字段数据根据唯一字段比较后 到值映射值映射：使用字段名 为起的后面用到的一个变量名（可任意起），源值列为系统默认 1. “identical” 旧数据和新数据一样2. “changed” 数据发生了变化;3. “new” 新数据中有而旧数据中没有的记录4. “deleted” 旧数据中有而新数据中没有的记录过滤记录步骤：通过flagfield的值来执行后面的分支，也可以通过Switch/Case组件执行后面的分支例如：5.Kettle定时任务新建一个【作业】，【核心对象】【通用】里面把【START】【转换】【成功】组件拖拽出来并通过Shift连接【START】组件：设置重复操作，时间间隔2秒【转换】组件：浏览，选择已保存的ktr文件【成功】：提示操作成功。判断数据数据是否变化：