ETL工具kettle学结.docx

精品文档概览Kettle也叫PDI(全称是Pentaho Data Integeration)，是一款开源的ETL工具，项目开始于2003年，2006年加入了开源的 BI 组织 Pentaho, 正式命名为PDI。官方网站：/术语1. Transformation转换步骤，可以理解为将一个或者多个不同的数据源组装成一条数据流水线。然后最终输出到某一个地方，文件或者数据库等。2. Job作业，可以调度设计好的转换，也可以执行一些文件处理（比较，删除等），还可以ftp上传，下载文件，发送邮件，执行shell命令等,3. Hop 连接转换步骤或者连接Job（实际上就是执行顺序）的连线 Transformation hop：主要表示数据的流向。从输入，过滤等转换操作，到输出。 Job hop：可设置执行条件：1， 无条件执行2， 当上一个Job执行结果为true时执行3， 当上一个Job执行结果为false时执行Kettle,etl设计及运行1. Kettle整体结构图 Kettle整体结构图2. 转换设计样例图绿色线条为hop,流水线 转换设计样例3. 运行方式 使用 java web start 方式运行的配置方法 命令行方式1） Windows下执行kitchen.bat，多个参数之间以“/”分隔，Key和value以”:”分隔例如：kitchen.bat /file: F:samplesdemo-table2table.ktr /level:Basic /log:test123.log/file： 指定转换文件的路径/level：执行日志执行级别/log: 执行日志文件路径2） Linux下执行kitchen.sh，多个参数之间以“-”分隔，Key和value以”=”分隔kitchen.sh -file=/home/updateWarehouse.kjb -level=Minimal如果设计的转换，Job是保存在数据库中，则命令如下：Kitchen.bat /rep:资源库名称 /user:admin /pass:admin /job:job名4. Xml保存转换，job流程设计 用户定义的作业可以保存在（xml格式）中或某一个特定的数据库中转换的设计文件以.ktr结尾（xml文格式），保存所有配置好的数据库连接，文件相对路径，字段映射关系等信息。Job的设计文件以.kjb结尾，下面是一个调用已经设计好的转换的job文件的一部分： demo test Transformation TRANS $Internal.Job.Filename.Directorydemo-table2table.ktr load customer data job $Internal.Job.Filename.Directory N N N N N 5. 数据库保存转换，job流程列出几个重要的表1） r_job: 保存job的id,name,status,执行时间，创建时间，修改时间等信息建表语句：DROP TABLE IF EXISTS r_job;CREATE TABLE r_job ( ID_JOB bigint(20) NOT NULL, ID_DIRECTORY int(11) DEFAULT NULL, NAME varchar(255) DEFAULT NULL, DESCRIPTION mediumtext, EXTENDED_DESCRIPTION mediumtext, JOB_VERSION varchar(255) DEFAULT NULL, JOB_STATUS int(11) DEFAULT NULL, ID_DATABASE_LOG int(11) DEFAULT NULL, TABLE_NAME_LOG varchar(255) DEFAULT NULL, CREATED_USER varchar(255) DEFAULT NULL, CREATED_DATE datetime DEFAULT NULL, MODIFIED_USER varchar(255) DEFAULT NULL, MODIFIED_DATE datetime DEFAULT NULL, USE_BATCH_ID char(1) DEFAULT NULL, PASS_BATCH_ID char(1) DEFAULT NULL, USE_LOGFIELD char(1) DEFAULT NULL, SHARED_FILE varchar(255) DEFAULT NULL, PRIMARY KEY (ID_JOB)2） r_jobentry:设计界面上的一个实体对应一个entry,通过job的id关联DROP TABLE IF EXISTS r_jobentry;CREATE TABLE r_jobentry ( ID_JOBENTRY bigint(20) NOT NULL, ID_JOB int(11) DEFAULT NULL, ID_JOBENTRY_TYPE int(11) DEFAULT NULL, NAME varchar(255) DEFAULT NULL, DESCRIPTION mediumtext, PRIMARY KEY (ID_JOBENTRY)3） r_jobentry_attribute ：job的详细信息表，包括job执行规则，执行过程中的参数来源，日志记录等 DROP TABLE IF EXISTS r_jobentry_attribute;CREATE TABLE r_jobentry_attribute ( ID_JOBENTRY_ATTRIBUTE bigint(20) NOT NULL, ID_JOB int(11) DEFAULT NULL, ID_JOBENTRY int(11) DEFAULT NULL, NR int(11) DEFAULT NULL, CODE varchar(255) DEFAULT NULL, VALUE_NUM double DEFAULT NULL, VALUE_STR mediumtext, PRIMARY KEY (ID_JOBENTRY_ATTRIBUTE)4） r_step：保存转换的步骤id,名字等r_step建表语句：DROP TABLE IF EXISTS r_step;CREATE TABLE r_step ( ID_STEP bigint(20) NOT NULL, ID_TRANSFORMATION int(11) DEFAULT NULL, NAME varchar(255) DEFAULT NULL, DESCRIPTION mediumtext, ID_STEP_TYPE int(11) DEFAULT NULL, DISTRIBUTE char(1) DEFAULT NULL, COPIES int(11) DEFAULT NULL, GUI_LOCATION_X int(11) DEFAULT NULL, GUI_LOCATION_Y int(11) DEFAULT NULL, GUI_DRAW char(1) DEFAULT NULL, PRIMARY KEY (ID_STEP)5） r_step_attribute：转换步骤的详细信息，字段重命名，字段映射等。通过外键id_transformation与r_step的id_transformation关联DROP TABLE IF EXISTS r_step_attribute;CREATE TABLE r_step_attribute ( ID_STEP_ATTRIBUTE bigint(20) NOT NULL, ID_TRANSFORMATION int(11) DEFAULT NULL, ID_STEP int(11) DEFAULT NULL, NR int(11) DEFAULT NULL, CODE varchar(255) DEFAULT NULL, -操作编码譬如：truncate，也可以是字段映射等信息 VALUE_NUM bigint(20) DEFAULT NULL, -操作值 VALUE_STR mediumtext, -操作值，譬如truncate对应的是Y或者N PRIMARY KEY (ID_STEP_ATTRIBUTE), UNIQUE KEY IDX_STEP_ATTRIBUTE_LOOKUP (ID_STEP,CODE,NR)说明：如果有一个字段firtstname映射到name则在r_step_attribute中增加两条记录。6. Kettle组成部分Chef：是一个图形用户界面，使用SWT开发，用来设计一个作业，转换，SQL，FTP，邮件，检查表存在，检查文件存在，执行SHELL脚本Kitchen：作业执行引擎，用来进行转换，校验，FTP上传。可以执行xml格式定义的任务以及保存在数据库上的。kitchen.bat /file:D:JobsupdateWarehouse.kjb /level:Basickitchen.sh -file=/PRD/updateWarehouse.kjb -level=Minimalkitchen.bat /rep:Production Repository /job:Update dimensions /dir:/Dimensions /user:matt /pass:somepassword123 /level:Basickitchen.bat /file: F:javapdi-ce-3.2.0-stabledata-integrationsamplestransformationsfilesdemo-table2table.ktr /level:Basic /log:test123.logSpoon：Spoon 是Kettle的另一个图形用户界面，用来设计数据转换过程Pan：Pan是一个 数据转换引擎，负责从不同的数据源读写和转换数据。pan.sh -file=/PRD/Customer Dimension.ktr -level=MinimalTransformation步骤1. 输入类型：1.1 Csv file input 读取csv文件，设置csv文件路径，可以设置csv文件的相对路径或者绝对路径，字段分隔符，文件读取的缓存大小等1.2 Excel Input 读取excel文件，和csv文件读取类似，增加了表单，表头，出错（是否忽略错误，严格的类型判断等）的处理1.3 Property Input 读取属性.properties文件1.4 Table input 从数据库读数据，动态绑定参数的SQL语句，参数替换可以从上一个步骤从获取。例如SELECT * FROM customer WHERE birthdate$current_date这里的$current_date在执行过程中会作为动态参数被替换掉。这个值是前一个转换步骤设置的。注：但是测试过程中发现如果上一个步骤设置的变量，在table input里面获取不到，变量设置必须作为一个单独的转换先执行一次，然后才能获取到这个变量。1.5 Text file input 主要是txt文件内容等，和csv差不多。1.6 Generate Rows 生成一些固定字段的记录，主要用来模拟一些数据进行测试。 1.7 Get File Names 读取给定目录或者文件全路径的文件名1.8 Get System Info 包括命令行输入的参数，操作系统时间，ip地址，一些特殊属性，kettle版本等 1.9 De-serialize from file 从二进制kettle cube文件中读取记录1.10 Access input 读取access数据库1.11 ESRI Shapefile Reader1.12 Fixed file input 读取固定大小文件1.13 Generate random value1.14 Get Files Rows Count 获取文件内容的行数1.15 Get SubFolder names1.16 Get data from XML从xml文件解析出数据1.17 LDAP Input从ldap库读取数据。1.18 LDIF Input读取ldap的ldif文件1.19 Mondrian InputMDX语言从Mondrian服务器上读取数据1.20 RSS Input1.21 S3 CSV Input1.22 SalesForce Input1.23 XBase input 读取XBase系列文件，如Foxpro文件，主要是数据库语言2. 输出类型2.1 Table output 将数据写入到数据库，可以指定是否truncate表，编辑前一步转换字段与现在表结构的字段映射关系。以及每次commit的记录数大小等。2.2 Text file output 将数据写入到文本文件，通常是csv文件2.3 Insert / Update 根据关键字找对应的记录，如果找不到则执行insert，否则执行update2.4 Update 跟insert/update类似，只是没有insert操作2.5 Delete 跟update类似，只是执行的是delete操作2.6 Excel Output 输出到excel，格式可以采用excel模板2.7 Serialize to file 将记录写到二进制文件中（cube文件）2.8 Access Output2.9 Properties Output输出到properties文件2.10 RSS Output2.11 SQL File Output将输出的sql insert语句保存到文件2.12 Synchronize after merge2.13 XML Output 输出到xml文件3. Transform类型3.1 Select values用于选择列，重命名列，指定列长度或精度 3.2 Filter rows通过使用一个表达式从输入行中过滤数据，将结果是TURE或FALSE的行输出到不同的节点。表达式是“”“OPERATOR”“” 的形式，其中OPERATOR可以是 =，, , =, REGEXP,IS NULL,IS NOT NULL, IN LIST, CONTAINS, STARTS WITH, ENDS WITH。 用户可以增加多个表达式，并用AND或OR连接。 3.3 Sort rows对指定的列以升序或降序排序，当排序的行数超过5000时需要临时表。 3.4 Add sequence为数据流增加一个序列，可以使用ORACLE中某一序列的值或由用户指定值 3.5 Dummy不做任何处理，一般作为流程的终点。 3.6 Join Rows (catesian product)对所有输入流做笛卡儿乘积。 3.7 Aggregate Rows聚集行数据，提供SUM，AVERAGE，COUNT，MIN，MAX，FIRST，LAST聚集函数，该类型不提倡使用，将来会被GROUP BY 类型替代。 3.8 Java Script value使用mozilla的rhino作为脚本语言，并提供了很多函数，用户可以在脚本中使用这些函数。例如 var prev_row; if (prev_row = null) prev_row = row; . String previousName = prev_row.getString(“Name”, “-”); . prev_row = row; 可以获得字段Name的前一条记录的值。 3.9 Row Normaliser该步骤可以从透视表（PIVOT TABLE）中还原数据到事实表，如从表一转换成表二，需要使用该步骤。3.10 Unique rows去掉输入流中的重复行，在使用该节点前要先排序，否则只能删除连续的重复行。 3.11 Group by分组 3.12 Calculator提供了一组函数对列值进行运算，使用该方式比用户自定义JAVA SCRIPT脚本速度更快。 3.13 Add constants增加常量值。 3.14 Row denormaliser同正规化过程相反。 3.15 Row flattener表扁平化处理除了上述基本节点类型外还定义了扩展节点类型，如SPLIT FIELDS，EXECUTE SQL SCRIPT，CUBE INPUT，CUBE OUTPUT等。图一中创建了一个简单的数据流程示例，共包括5个节点,其中Table input节点使用了SQL SERVER数据库中的一张表（三条记录）,Filter rows中定义了过滤条件,将符合条件的发送到file2节点,不符和条件的记录发送到Select values节点。 Select values节点中选择列，并对选择的列进行了设置，将结果发送到file1节点。file1,file2节点分别是两个文本文件节点,最终用来保存数 据。该流程运行后，可以在Log View面板中查看运行结果，如图四所示 从table input结点输入3条记录，经过滤后输出到file2节点2条记录（OUTPUT列中的3是指2条记录加1行列名），输出到file1节点1条记录 （OUTPUT列中的2是指1条记录加1行列名）。4. Flow4.1 Abort 忽略上一步的输入流，通常用在错误处理中，譬如不处理X条记录后的所有记录4.2 Append streams 主要用来处理步骤之间有优先级的问题。从两个步骤从读取数据流，指定步骤的读取顺序。4.3 Blocking Step阻塞所有的输出直到最后一条记录到达4.4 Detect empty stream 当输入流为空的时候，输出一条空的记录4.5 Dummy (do nothing)空操作。是一个空操作的插件4.6 Filter rows通过设定过滤条件来过滤记录4.7 Identify last row in a stream4.8 Switch / Case类似Java的switch语法，通过比较某一确定的字段值来将数据转发到不同的转化步骤5. Joins5.1 Merge join合并两种不同输入流，连接方式有内连，左外连接等。需要注意的是记录需要先按关键字进行排序5.2 Merge Rows (diff)用于比较两组输入数据，一般用于更新后的数据重新导入到数据仓库中。两组数据中一组是引用流，一组是比较流，每次比较后只有最新版本的行数据被 输出到下一步。比较结果包括：idectical一致：两组流的主键一致，值一致changed有变化：两组流的主键一致，值有一个或多个不同new新行：引用流中有而比较流中没有某一主键deleted被删 除的行：比较流中有而引用流中没有某一主键 比较流里面的数据除了被标记为deleted都会进入下一个步骤里面5.3 Sorted Merge对记录按某个关键字进行排序5.4 XML Join将一个XML文作为节点添加到另一个XML里面 合并前的XML文需要合并的XML合并后的XML文6. Scripting6.1 Execute SQL script执行SQL脚本，应该避免使用这一步骤，尽可能的使用“table input（select）”,”table output(insert)”,”update”,”delete”等步骤来替代。譬如动态创建表（表名是可变的，table1,table2,table3）：SQL脚本是：CREATE TABLE?(ID INTEGER);6.2 Execute row SQL script对Execute sql script的补充，增加了可以自定义sql语句的字段名6.3 Formula在数据流中执行公式6.4 Modified Java Script ValueModified Java Script Value应该说是转换步骤里最强大的一步，可以获取前一步的输入流的所有字段，调用Java api对数据做转换等操作，改变所有输出的值。还能通过设置转换状态常量对现有转换流程做改变，（忽略转换，设置为错误，继续转换）。脚本是 Mozilla 的 Rhino，Rhino 是一个Java实现的Javascript解释器。现在已经加入到 JDK 1.6 的 javax.script 包中了。对数据流进行修改等操作提供了常量,函数,输入字段,输出字段的列表显示1) Transformation scripts 已经创建的脚本2) Transformation constants 已经预先定义好的静态常量,不可更改,例如SKIP_TRANSFORMATION , ERROR_TRANSFORMATION, CONTINUE_TRANSFORMATION3) Transformation functions 类型转换,操作函数: 字符串,数字,日期之间的转换,字符串截取等 逻辑判断函数： isDate(var) ，isNumber(var)等特殊函数：LoadScriptFile(var)，getProcessCount(var)，print(var)，writeToLog(var)，getVariable(var,var)文件操作函数：createFolder(var)，deleteFile(var)，getLastModifiedTime(var,var)，moveFile(var,var,var)Input fields 获取输入流中字段的值 field.getNumber()Output fields set输出流中字段的值 field.setValue(99)例子：1) 过滤Null字段 var a;if ( fieldname.isNull() ) a = 0;else a = fieldName.getString();2) 字符串截取将字符串“12345McDonalds”前面的数字部分分割出来var str = field.getString();var code = ;var name = ;for (i = 0; i str.length(); i+ ) c = str.charAt(i); if ( ! java.lang.Character.isDigit(c) ) code = str.substring(0, i); name = str.substring(i); Alert(code=+code+, name=+name); break; 3) 过滤记录行，控制转换流程trans_Status = CONTINUE_TRANSFORMATIONif (field.getString()=123) trans_Status = SKIP_TRANSFORMATION4) 使用java类库var mydate = 20090723;var parser = java.text.SimpleDateFormat(yyyyMMdd); /Must use fully qualified java classvar dateObj = parser.parse(mydate); /just like how you would do in javaAlert(dateObj);6.5 Regex Evaluation通过正则表达式验证输入字段6.6 User Defined Java Expression执行一些简单的java代码譬如表达式：firstname+ +name也可以用Java代码：new StringBuffer(firstname).append( ).append(name).toString()7. Lookup7.1 Call DB Procedure 执行存储过程并获得返回值，返回值只有一个，参数可以多个。7.2 Check if a column exists检查数据库表是否存在某列7.3 Database join改步骤允许查询等操作利用上一步的数据，譬如参数动态绑定的查询语句，可以被上一步某个字段的值替换掉7.4 Database lookup和database join功能类似，从数据库查询数值，作为新的字段添加到数据流中。可将前面的输出流的值作为查询比较参数7.5 Dynamic SQL row动态SQL查询记录行数。7.6 File exists检查文件是否存在，文件名由上一步传来7.7 HTTP Post处理POST请求，url可以从上一步数据里获取，也可以在该步骤指定，可以指定编码，请求参数等。7.8 HTTP client仅仅是一个带参数的URL请求，url可以从上一步数据里获取，也可以在该步骤指定，不支持soap例如 http:/?param1=value1¶m2=value27.9 Stream lookup从其他转换步骤产生的数据流中查询数据7.10 Table exists判断数据库中某张表是否存在，表名由前面步骤传来7.11 Web services lookup处理SOAP请求，数据类型转换是在步骤内部处理，如果有日期，数字等类型需要转换，建议全部返回字符串，然后使用“Select values”步骤做转换JOB类型1. 常用1.1 Start指定job执行规则，是否循环，循环规则等。1.2 Dummy空操作，主要用来多数据源汇总1.3 Abort job终止，忽略一个Job1.4 Display Msgbox Info使用图形化界面执行Job的时候显示消息框1.5 Dummy pluginJob里面空操作，可以用来将执行循环操作1.6 Job执行已经定义好的Job。job可以嵌套job。1.7 Ping a hostping1.8 Send SNMP trap发送SNMP trap 报文SNMP TRAP就是在SNMP设备发生状态变化的时候向管理器发出信号。不用管理器来检查。1.9 Set variablesSet变量1.10 Success当出错后可以强制将该job置成功1.11 Transformation执行定义好的转换1.12 Truncate tables1.13 Wait for SQL 当检测表的记录数是否达到一定条件1.14 Write To Logjob里面的日志记录，不同于程序自带的log4j等日志2. 邮件2.1 Get mails from POP从POP服务器获取邮件并存储到文件夹中2.2 Mail发送文本或者HTML格式邮件，可添加附件。2.3 Mail validator 通过发送SNMP TRAP到邮件服务器来验证EMAIL地址是否正确3. 文件管理3.1 Add filenames to result将文件夹或者多个文件加入到数据流中，以便在下一个job步骤中使用3.2 Compare folders比较两个文件夹下面的文件是否一致，可以选择只比较文件大小，也可以选择比较文件内容3.3 Copy Files3.4 Copy or Move result filenames根据上一步的执行结果得到的文件名，复制或者剪切文件3.5 Create a folder3.6 Create file3.7 Delete file3.8 Delete filenames from result根据上一步的执行结果得到的文件名，复制或者剪切文件3.9 Delete files3.10 Delete folders3.11 File Compare比较两个文件的内容3.12 HTTP 通过http协议从web服务器上获得文件3.13 Move Files移动文件3.14 Unzip fileUnzip解压缩文件3.15 Wait for file循环检测文件是否存在，否则直到超时失败3.16 Zip file采用zip压缩文件4. 文件传输4.1 FTP Delete删除FTP上的文件4.2 Get a file with FTP从FTP获取文件，可以设置编码方式，连接超时时间等。文件保存路径，文件名中是否包含日期，时间，时间日期是否需要特殊格式化，是否覆盖文件。4.3 Get a file with SFTP与ftp类似，只是采用Secure FTP protocol4.4 Put a file with FTP上传文件，本地文件路径，通过*等模糊匹配要上传的文件。并执行上传的远程目录。4.5 Put a file with SFTP4.6 SSH2 Get4.7 SSH2 Put5. Scripting5.1 JavaScript编辑javascript脚本并执行。5.2 SQL执行SQL脚本可以执行写好的sql脚本，指定sql路径即可。也可以插入编辑sql脚本。5.3 Shell执行SHELL脚本，可以执行已经写好的shell脚本 指定shell脚本路径即可，也可以自己插入，编辑shell脚本可以把前一个