KETTLE使用经验总结

1、kettle使用经验总结成都树之联科技有限公司 kettle使用经验总结成都树之联科技有限公司1目录 TOC o 1-3 h z u HYPERLINK l _Toc475613074 1. 软件安装 PAGEREF _Toc475613074 h 1 HYPERLINK l _Toc475613075 1.1. 安装说明 PAGEREF _Toc475613075 h 1 HYPERLINK l _Toc475613076 1.2. 安装jdk PAGEREF _Toc475613076 h 1 HYPERLINK l _Toc475613077 1.3. 安装PDI PAGEREF _To

2、c475613077 h 1 HYPERLINK l _Toc475613078 1.4. 创建资源库 PAGEREF _Toc475613078 h 2 HYPERLINK l _Toc475613079 1.5. 修改配置文件 PAGEREF _Toc475613079 h 3 HYPERLINK l _Toc475613080 1.6. 启动服务器 PAGEREF _Toc475613080 h 6 HYPERLINK l _Toc475613081 2. 操作说明 PAGEREF _Toc475613081 h 7 HYPERLINK l _Toc475613082 2.1. 运行转换

3、/作业 PAGEREF _Toc475613082 h 7 HYPERLINK l _Toc475613083 2.2. 转换的并行 PAGEREF _Toc475613083 h 7 HYPERLINK l _Toc475613084 2.3. 作业的并行 PAGEREF _Toc475613084 h 9 HYPERLINK l _Toc475613085 2.4. 集群的使用 PAGEREF _Toc475613085 h 9 HYPERLINK l _Toc475613086 2.5. 记录日志 PAGEREF _Toc475613086 h 11 HYPERLINK l _Toc47

4、5613087 2.6. 连接hbase PAGEREF _Toc475613087 h 16 HYPERLINK l _Toc475613088 2.7. 读取XML文件 PAGEREF _Toc475613088 h 26 HYPERLINK l _Toc475613089 2.8. 连接hive2 PAGEREF _Toc475613089 h 31 HYPERLINK l _Toc475613090 2.9. 大字段处理 PAGEREF _Toc475613090 h 33 HYPERLINK l _Toc475613091 3. 性能优化 PAGEREF _Toc475613091

5、h 34 HYPERLINK l _Toc475613092 3.1. 利用好数据库性能 PAGEREF _Toc475613092 h 34 HYPERLINK l _Toc475613093 3.2. 用并行或者集群解决好数据插入瓶颈 PAGEREF _Toc475613093 h 35 HYPERLINK l _Toc475613094 3.3. 增大提交的记录数及大字段数据处理 PAGEREF _Toc475613094 h 35 HYPERLINK l _Toc475613095 3.4. 全量抽取先抽取后建索引 PAGEREF _Toc475613095 h 36 HYPERLIN

6、K l _Toc475613096 3.5. 增量抽取注意去重数据量 PAGEREF _Toc475613096 h 36 HYPERLINK l _Toc475613097 3.6. 利用中间表分段处理数据 PAGEREF _Toc475613097 h 37 HYPERLINK l _Toc475613098 3.7. 聚合优先 PAGEREF _Toc475613098 h 37 HYPERLINK l _Toc475613099 4. 常见问题解决 PAGEREF _Toc475613099 h 37 HYPERLINK l _Toc475613100 4.1. 大量数据抽取导致内存溢

7、出 PAGEREF _Toc475613100 h 37 HYPERLINK l _Toc475613101 4.2. 字段值丢失 PAGEREF _Toc475613101 h 37 HYPERLINK l _Toc475613102 4.3. 输出记录数大于输入记录数 PAGEREF _Toc475613102 h 38kettle使用经验总结kettle使用经验总结成都树之联科技有限公司1软件安装安装说明本文档对应的产品及版本是pdi-ce-130。由于该产品使用java开发，所以需要在服务器上配置java环境。如果是linux系统，为了操作方便，可以在linux上部署服务器，Windo

8、ws启动客户端。安装jdk下载jdk1.7或以上安装包，安装成功后，配置java环境变量。JAVA_HOME: java安装目录CLASSPATH：%JAVA_HOME%libdt.jar;%JAVA_HOME%libtools.jarPATH：添加%JAVA_HOME%bin;%JAVA_HOME%jrebin;LINUX下配置环境变量：进入/etc/profile(系统)或.bash_profile(用户)export JAVA_HOME= /usr/local/java/jdk1.8.0_60export PATH=$JAVA_HOME/bin:$PATHexport CLASSPATH

9、=$JAVA_HOME/lib/dt.jar:$JAVA_HOME/lib/tools.jar配置完成后检查java -version 和$JAVA_HOME是否启用安装PDI地址/projects/data-integration/下载PDI二进制文件pdi-ce-130.zip到各服务器中并解压，data-integration目录里面包含了PDI所有内容。根据需要连接的数据库，将驱动包放入lib文件夹下。创建资源库DI可使用文件或资源库的形式来存储作业、转换和其它服务器配置。使用资源库来存储任务，可以方便地存储和分发作业到不同的服务器执行，实现数据的并发处理。打开SPOON时，会提示新建

10、资源库；或者进入主界面后，通过工具资源库连接资源库，打开资源库连接对话框：点击左上角 “+” 按钮，选择Kettle database repository项目，在弹出的资源库信息框中，点新建按钮新建一个数据库连接，选择JDBC连接数据库，并填写连接到数据库的相应信息，该数据库就是存储DI资源的数据库。然后点击“创建或更新”按钮，在弹出的框中输入管理员密码。出现一个SQL语句的对话框，点击执行按钮。这样资源库就创建好了，输入用户名admin和密码登录。打开用户主目录下的.kettle目录，找到repositories.xml文件，该文件存储了服务器到资源库的数据库连接，将它复制到各服务器用户目

11、录下的.kettle目录下。如果linux 的root目录了下没有.kettle目录，将整个目录拷贝到root下。修改配置文件在data-integration/pwd目录中有几个示例配置文件，它们是DI服务器启动的配置文件。在主服务器上，修改master配置文件，在段中，修改主机名，端口，用户名和密码等信息，为安全起见，密码需使用Encr.bat或encr.sh加密：（encr -kettle password）（注意：地址请使用IP，不然其他服务器无法读取） master ip 8080cluster1 Encrypted 2be98afc86aa7f2e4a815bb63ca97bd8b

12、 Y 添加段，设置资源库的连接方式： KettleDatabaseRepository kettle_res kettle reposity admin Encrypted 2be98afc86aa7f2e4cb79ce71da9fa6d4 在从服务器上修改配置文件，需要配置，段，同样地，需要添加段。 master ip 8080 cluster1 Encrypted 2be98afc86aa7f2e4a815bb63ca97bd8b Y Y slave-8081 ip 8081 cluster2 Encrypted 2be98afc86aa7f2e4a815bb63ca97bd8b N Ke

13、ttleDatabaseRepository kettle_res kettle reposity admin Encrypted 2be98afc86aa7f2e4cb79ce71da9fa6d4 启动服务器在data-integration路径下使用master配置文件来启动主服务器，如：nohup ./carte.sh pwd/carte-config-master-8080.xml &使用slave文件启动从服务器nohup ./carte.sh pwd/carte-config-slave-8080.xml &至此：kettle集群配置完成。可以使用http:/localhost:8

14、080/kettle/getSlaves查看主服务器下子服务器连接情况操作说明运行转换/作业进入主服务器data-integration路径下，使用pan命令提示转换可以使用的参数然后以pan /trans:转换名的格式运行相应转换，如果使用了资源库，一定要加上/rep、/user、/pass参数。同理,用kitchen命令来运行作业。./kitchen.sh -rep RE -user admin -pass admin -dir /GP测试 -job 作业测试./pan.sh -rep RE -user admin -pass admin -dir /GP测试 -trans -直接输出测试

15、转换的并行kettle运行采用的是流水线的机制(各个步骤并行执行，只要前一个步骤产生了新的内容，后一个步骤就会执行)，如果某一个步骤执行比其他步骤缓慢，就会拖累整个转换的执行，可以采用步骤复制来解决这个问题。首先要保证上一步骤到该步骤数据发送方式为Round-robin(复制模式会导致数据重复)然后右键该步骤，选择“改变开始复制的数量”，然后输入一个复制数X，代表启用X个线程共同执行这个步骤。注意：多线程同时执行会影响数据的排序，如果后面还需要排序(去重、记录集连接等)，建议不要复制，因为排序操作会影响整个转换流水线的执行。作业的并行如果作业项之间没有严格的先后关系，也可以并行作业。在某一作业

16、项上选择“launch next entries in parallel”,该项后面的作业项会在该项完成后同时启动执行。集群的使用打开spoon.bat，新建一个转换。选择主对象树子服务器，右键新键子服务器：填写内容与master/slave保持一致，其中主服务器要勾选如上图选项。所有服务器均建立对应的子服务器后，可以建立集群了。在主对象树中选择kettle集群schemas，填写schema名称，端口、sokects缓存大小、sockets刷新时间间隔、socketts数据是否压缩。在右边点击选择子服务器，选择需要的子服务器。Schema名称即集群的名称。端口即集群对外服务的端口.socke

17、ts缓存大小:sockets缓存Sockets刷新时间间隔：达到多少行记录时刷新到子服务器。Sockets数据是否压缩：如果网络状况差，则建议选择。网络良好时不用选择。相比固定子服务器的集群，使用动态集群可以在部分子服务器出现故障时仍然保证转换正常执行，动态集群只需要在建立时勾选“Dynamic cluster”，并且只选取主服务器即可。配置完成后，在对应的转换步骤上右键，选择“集群”，再选择已经配置好的集群。选择然后会发现该步骤多出”CX2”，表示有2个子服务器来执行。记录日志双击空白转换/作业空白处或者右击，选择转换设置，可以打开转换设置窗口，选择日志项进行日志设置。每一个日志都手动设置比

18、较麻烦，而且不方便统一，可以采用触发器方式自动设置日志，方法如下：先建立日志库(KETTLELOG)和相关日志表(直接在KETTLE上执行SQL创建，保证表结构和KATTLE日志结构的一致性)，然后使用以下Job创建日志触发器：转换日志触发器：DELIMITER $DROP TRIGGER IF EXISTS r_trans_log$CREATE DEFINER=rootlocalhost TRIGGER r_trans_log BEFORE INSERT ON R_TRANS_ATTRIBUTEFOR EACH ROWbegin- Kettle转换操作-日志配置触发器 DECLARE v_t

19、rans_log varchar(30) default TB_TRANS_LOG; DECLARE v_step_log varchar(30) default TB_TRANS_STEP_LOG; DECLARE v_performance_log varchar(30) default TB_TRANS_PERFORMANCE_LOG; DECLARE v_channel_log varchar(30) default TB_TRANS_CHANNEL_LOG; - 数据库连接名(日志) DECLARE v_database_name varchar(30) default KETTLE

20、LOG;- 更新日志表if new.code = TRANS_LOG_TABLE_TABLE_NAME thenset new.value_str = v_trans_log;elseif new.code = STEP_LOG_TABLE_TABLE_NAME thenset new.value_str = v_step_log;elseif new.code = PERFORMANCE_LOG_TABLE_TABLE_NAME thenset new.value_str = v_performance_log;elseif new.code = CHANNEL_LOG_TABLE_TABL

21、E_NAME thenset new.value_str = v_channel_log;elseif new.code = TRANS_LOG_TABLE_CONNECTION_NAME thenset new.value_str = v_database_name;elseif new.code = STEP_LOG_TABLE_CONNECTION_NAME thenset new.value_str = v_database_name;elseif new.code = PERFORMANCE_LOG_TABLE_CONNECTION_NAME thenset new.value_st

22、r = v_database_name;elseif new.code = CHANNEL_LOG_TABLE_CONNECTION_NAME thenset new.value_str = v_database_name;end if;end $DELIMITER ;作业日志触发器：DELIMITER $DROP TRIGGER IF EXISTS r_job_log$CREATE DEFINER=rootlocalhost TRIGGER r_job_log BEFORE INSERT ON R_JOB_ATTRIBUTEFOR EACH ROWbegin- Kettle转换操作-日志配置

23、触发器 DECLARE v_job_log varchar(30) default TB_JOB_LOG; DECLARE v_job_entry_log varchar(30) default TB_JOB_ENTRY_LOG; DECLARE v_job_channel_log varchar(30) default TB_JOB_CHANNEL_LOG; - 数据库连接名(日志) DECLARE v_database_name varchar(30) default KETTLELOG;- 更新日志表if new.code = JOB_LOG_TABLE_TABLE_NAME thens

24、et new.value_str = v_job_log;elseif new.code = JOB_ENTRY_LOG_TABLE_TABLE_NAME thenset new.value_str = v_job_entry_log;elseif new.code = CHANNEL_LOG_TABLE_TABLE_NAME thenset new.value_str = v_job_channel_log;elseif new.code = JOB_LOG_TABLE_CONNECTION_NAME thenset new.value_str = v_database_name;elsei

25、f new.code = JOB_ENTRY_LOG_TABLE_CONNECTION_NAME thenset new.value_str = v_database_name;elseif new.code = CHANNEL_LOG_TABLE_CONNECTION_NAME thenset new.value_str = v_database_name;end if;end $DELIMITER ;触发器建好后，转换和作业执行时都会有完整的日志记录，可以修改触发器取舍某些日志项。连接hbase2.6.1统一hadoop版本进入kettle安装目录，路径data-integrationpl

26、uginspentaho-big-data-pluginhadoop-configurations下每一个文件夹对应不同的hadoop版本。kettle以插件的形式对hadoop不同版本进行支持，每个版本的kettle都会有默认支持版本（如kettle5.4支持hadoop2.1，kettle6.1支持hadoop2.3），如果默认版本与hadoop环境的版本不一致，需要对相应jar包进行替换，下面以kettle5.4支持hadoop2.3进行举例说明。进入hadoop-configurations目录，将hdp22复制一份，改名为hdp23；清空hdp23libclient目录下的jar包；

27、将Hadoop安装目录下share目录下的所有jar包复制到client目录下(这里为了方便，我直接从kettle6.1对应文件夹下复制jar包)；将hadoop安装目录下的（linux下已经配好的）配置文件复制到hdp23目录下,复制的文件为图中红框部分。这样一来，kettle就可以支持hadoop的其他版本了。2.6.2修改配置文件1、在hdp23目录下perties文件中添加hadoop版本号并添加如下一行vider=NO_AUTH2、在pentaho-big-data-plugin目录下perties文件中修改hdp选项(要支持哪个hadoop版本就用相应的文件夹名)3、在C:Wind

28、owsSystem32driversetc目录下修改域名映射文件hosts，将hadoop集群的域名与Ip映射关系添加进去2.6.3修改kettle客户端用户权限为了避免访问hbase出现权限不足，需要将kettle安装文件的主属修改为和hdfs文件系统主属一致。在计算机管理里添加用户组，并将当前用户修改为hdfs文件系统用户名，并添加到刚刚建立的用户组中完成该过程后需要注销并重新登录。2.6.4连接hdfs 新建一个转换，在“主对象树”中选择“Hadoop cluster”，点击右键选择“New Cluster”，填写相关信息，如下图所示因为版本5.4还没有提供连接测试，所以只能通过后面操作

29、是否报错来发现问题。但是kettle6.1以后提供了测试功能，可以直接测试，结果如下图表明kettle与hadoop连接成功。2.6.5 hbase input首先，在Configure query页签选择已创建的Hadoop Cluster集群,然后切换到第二个页签Create mappings,点击Get table names (这里可以看到hbase里的表，说明前面连接成功),选择需要往外读的表， 然后给要建的映射关系命名(如下图的distest)，在下方创建映射关系。说明：1需要输出的每个列(column name)都设置别名(Alias) ;2必须要有一个字段来存储行健(distr

30、ict_id);3Type全部选用String点击sve mapping将该映射关系保存到hbase中(pentaho_mappings)。最切换至标签一，点击get mapped table names获取hbase表名，点击get mappings for the specified table选择需要使用的映射关系，再点击get key/fields info获取映射关系，保存设置。 2.6.6 hbase outputHbase output的设置和Hbase input十分类似，仅有以下两个区别：1配置映射关系时可以直接获取前面步骤输入的字段(get incoming fields)

31、；2配置连结时只需选择映射关系不用再列出具体匹配（区别如下图）2.6.7注意事项1，因为hdfs文件系统的权限控制，kettle运行环境（windows客户端或者linux服务器）的权限要与其一致；2，如果遇到hbase连结通过(Get table names能够看到Hbase的表),但是执行时报Unable to obtain a connection to HBase: null错误，需要将已保存转换关闭并重新打开，看看该组件是否能正常打开，如果不能，说明是kettle自身组件问题。（经测试，kettle6.1的hbase output组件存在该问题，设置正常，保存后重新打开转换组件不能打

32、开，报错如下）读取XML文件2.7.1注意事项使用输入下的 XML 文件输入 (StAX解析) 组件来读取XML文件，结构如下XML文件结构如下2.7.2拆表我们读取的XML里面包含多个表的数据，我们需要将他进行拆分，分别取值。选取流程中的Switch / Case组件，通过对全路径的解析来达到拆分各表数据的目的，如图，我们把收案和立案信息表和与案实体名录表的信息分别进行输出和处理。2.7.3筛选与行列转换获取到各个表的数据之后，我们还需要将它转换为我们我们常见的横表结构，而非现在的字段名一个列，值一个列的竖表结构。这就需要应用到转换中的列转行组件如图，这个组件中，关键字段就是我们需要它每行的

33、值转换为新表字段的那一列，分组字段就是通过哪个字段确定哪些数据在新表的同一行。下方关键字段就是行转列的依据值。这个组件需要对分组字段进行排序，所以在之前要先有排序组件，再使用筛选组件排除开始结束标签和R标签的干扰。输出结果如下连接hive2因为kettle安装后自带hadoop插件，其中包含了Hive的驱动，所以只需要将pentaho-big-data-plugin的配置文件perties指向对应的hdp版本即可然后使用Hadoop Hive2建立数据库连接测试通过，如果测试没有通过或者报了驱动问题，可以参考2.6同步hadoop版本从hive数据库读出数据只需要使用表输入组件即可，写语句的时

34、候注意在表前面加上数据库名，否则会默认取default库取值， 往hive数据库写值用表输出组件即可，但是要注意在数据库字段匹配时hive表显示的字段是带表明的,如图中test.id，这样的话数据无法正常插入Hive表。解决方法是手动去掉test.字样，只使用hive里的字段名即可。错误处理: org.apache.hadoop.ipc.RemoteException:Permission denied: user=hive, access=WRITE, inode=/apps/hive/warehouse/zwz_dw.db/test/.hive-staging_hive_2017-02-2

35、8_17-16-14_234_2052200583886124766-2:hdfs:hdfs:drwxr-xr-x该错误表明,hive用户不具有写入数据权限,需要将Hive数据源的配置用户改为hdfs，同时将本机的用户改为用户组hdfs下的hdfs用户大字段处理Kettle没有大字段类型，所以大字段需要转换为base64编码，然后再在目标表中用text/longtext/blob来存储。转换base64需要使用js来进行，配置及JS代码如下：var bytes;var base64img;if(PHOTO = null)base64img = null;elsebytes = Pmons.co

36、dec.binary.Base64.encodeBase64(PHOTO);base64img = new Packages.java.lang.String(bytes);使用变量指定输出路径我们有很大一部分的任务需要在服务器上执行，为了兼顾本地和服务器使用同样的调度，有文件输出的转换需要我们使用变量来指定路径。首先我们需要建立job，增加变量，然后再指定路径的时候将变量代替路径。需要注意的是，当输出文本文档的时候，需要强制封闭解析文本文件实际的执行中，上方的浏览只是用于找到路径，实际是在下方的文件上读取循环循环都是在job中完成，我们用到了检验字段值这个组件主要思路是，做一张记录表，含有判

37、断标签，当未扫描的记录，标签数大于1的时候，继续循环，执行一条记录，打一次标签，直到全部执行，退出循环使用Sqoopkettle的sqoop组件是在作业的big data模块下面，截止编写这个文档的时候，我仍未测通。所以，在这个地方，我们采用了仍在作业下面，脚本模块的执行SHELL脚本来使用sqoop组件。关于脚本执行有几个注意事项1.为避免出现错误，需要在LINUX环境下执行2.kettle服务器与sqoop服务器不在一起，则需要重定向语句3.切换用户后的第一个exit是退出该用户到root例子:性能优化利用好数据库性能Kettle在对数据分组，排序，连接时，都需要先进行排序，而排序需要生成

38、中间文件，这样处理起来会很慢，如果可以利用数据库的索引，在数据库中直接完成相应操作，则会快很多。用并行或者集群解决好数据插入瓶颈一般来说，数据插入会是整个转换中最好费性能的地方，可以使用并行或者集群来提升插入效率。但是需要注意如果复制数过多，可能导致数据库连接数超出，需要增大数据库的连接数。另外,在更新的时候要特别注意避免资源争用导致的锁表。增大提交的记录数及大字段数据处理Kettle往数据库写值默认的提交记录数是100，如果没有大字段，将该值增大后可以减少提交次数，提升插入效率，建议20000-50000。如果遇到大字段，提交数建议50-100，同时在转换属性-杂项里将“集合里的记录数”和“每个反馈行的处理记录数”减小，并且“集合里的记录数”设置值为“表输出”复制倍数*提交数还要注意，在大字段数据传输时，要坚决避免在kettle中进行排序。全量抽取先抽取后建索引首次插入数据时，在抽取完成后再去建立相关字段的索引可以提高数据插入效率。增量抽取注意去重数据量业务系统上做增量抽取通常会受限于原有业务表是否有时间戳字段，如果在业务时间上做增量，抽取前必须先去重，这时候会使用delete，如果去重记录数太多，就必须对这部分记录进行分批提交，建议使用存储过程来完成。创建好带参数的存储过程，在需要去重时调用“执行SQL语句”步骤并传入参数。注意：将该存储过程