pentaho(keetle)使用手册

pentaho使用

先展示一下用途和效果

1. 环境准备

1.1 pentaho是什么？

pentaho可读作“彭塔湖”，原名keetle 在keetle被pentaho公司收购后改名而来。

pentaho是一款开源ETL工具，纯java编写的C/S模式的工具，可绿色免安装，开箱即用。支持Windows、macOS、Linux平台。
pentaho有2个核心设计，即转换和作业。
转换是一个包含输入、逻辑处理、输出的完整过程，即ETL。
作业是一个提供定时执行转换的机制，即定时服务调度。
pentaho官网下载链接：Pentaho Community Edition Download | Hitachi Vantara

pentaho由主要四部分组成

• Spoon.bat/Spoon.sh ：勺子，是一个图形化界面，可图形化操作转换和作业
  
• Pan.bat/Pan.sh : 煎锅，可用命令行方式调用转换
  
• Kitchen.bat/Kitchen.sh : 厨房，可用命令行方式调用作业
  
• Carte.bat/Carte.sh ： 菜单，是一个轻量级web容器，可建立专用、远程的ETL Server
  

1.2 pentaho安装

Windows

由于是纯java编写，依赖jdk环境。所以需要先配置jdk环境，这里省略。
从官网下载pentaho安装包后，直接解压。

MacOS

1. tar -zxvf 安装包路径 -C 目标路径

复制代码

Linux

1. tar -zxvf 安装包路径 -C 目标路径

复制代码

目录结构

重点目录以及执行文件说明

• lib目录 ： 这是依赖库目录，例如各个数据库的jdbc驱动，都放在此目录下
  
• logs目录 ：这是转换和作业运行的默认日志输出目录
  
• simple-jndi目录 ：这是各个数据库的JNDI连接信息的全局配置
  
• Spoon.bat/Spoon.sh ：勺子，是一个图形化界面，可图形化操作转换和作业
  
• Pan.bat/Pan.sh : 煎锅，可用命令行方式调用转换
  
• Kitchen.bat/Kitchen.sh : 厨房，可用命令行方式调用作业
  
• Carte.bat/Carte.sh ： 菜单，是一个轻量级web容器，可建立专用、远程的ETL Server
  

在window上运行就用.bat格式脚本，MacOS 或者 Linux 平台上使用.sh格式脚本

2. 开始使用

pentaho内置了丰富的数据处理组件，本章节主要对pentaho界面上各个功能组件作用进行说明。
2.1 启动图形化界面

Windows

1. 运行 Spoon.bat

复制代码

MacOS

1. 运行 Spoon.sh

复制代码

Linux

1. 运行 Spoon.sh

复制代码

运行后会短暂没有任何反应，等待会议，就会出现界面

主对象树中有转换和作业

• 转换：所有的数据处理工作都在转换中完成
  
• 作业：这是一个任务
  

开始数据处理工作前，必需新建一个转换，因为只有新建了之后，才能使用数据处理组件，此时的核心对象树是空的。
2.2 转换

在 “核心对象树 –> 转换 –> 右键 –> 新建” 或 在 “文件 –> 新建 –> 转换” ,新建一个转换，核心对象树就会出现各类组件。依靠这些组件组合使用，完成数据处理工作。

2.2.1 主对象树

一个转换就是一个数据处理工作流程。这里主要是转换的配置，例如数据源连接，运行配置等。

2.2.2 核心对象树

包含各类数据处理组件。
2.3 数据处理组件

这里对一些常用的组件进行说明
输入

输入组件，即各类数据源，例如数据库，json，xml等
输出

输出组件，将处理后的数据进行输出保存
转换

这是数据转换的核心，在这里完成数据处理
应用

包含一些数据处理外的操作，例如发送邮件，写日志等
流程

用于控制数据处理流程，例如开始，结束，终止等
脚本

当内置转换组件完成不了数据处理的逻辑时，即可使用脚本组件，用自定义代码的方式来完成处理逻辑
查询

用于一些查询请求，例如http请求，数据库查询某个表是否存在等
连接

可用于多表，单表处理完后，进行记录合并
2.4 作业组件

在“文件–>新建–>作业”创建一个作业。
主对象树包含作业运行配置，DB连接配置等

核心对象树包含作业的各类组件

通用

作业流程组件，有开始、转换、成功、空处理等
邮件

发送邮件
文件管理

文件操作，创建、删除等
条件

条件处理，例如判断某个文件是否存在
脚本

使用shell，js、sql等脚本处理复杂作业逻辑
应用

作业处理，例如终止作业、写日志等
文件传输

定时作业来上传、下载文件
2.5 使用

上面介绍了各个组件用途，现在来完成一个完整的数据处理工作流程。
启动应用

略
新建转换

在 “核心对象树 –> 转换 –> 右键 –> 新建” 或 在 “文件 –> 新建 –> 转换” ,新建一个转换
配置DB连接

在主对象树中选择DB连接，右键新建

注意：连接数据库之前需要下载对应的jdbc驱动,例如连接pgsql则需要下载 postgresql-version.jar，r然后将驱动包放到安装目录下的\lib目录

这里以kingbase V8为例，因为这个踩了坑。经历如下：
内置的数据源里有KingbaseES，本以为可以直接用，结果发现连不上，报驱动错误。可能是因为内置的驱动版本跟数据库版本不一致，因为Kingbase V8的驱动不向前兼容。更新驱动后，依然不行。
然后发现，内置还有Generic database选项，这个是用来自定义连接内置数据源之外的数据库的。使用jdbc方式连接，需要一个连接串，驱动包名(前提是下载了对应的驱动包)，用户名，密码。然而，这种方式依然不行……
后来一想，干脆用pgsql的方式来连接kingbase，没想到连接成功！

选择输入

因为数据源是数据库，所以这里从输入组件中选择表输入，将其拖入到右侧面板中

配置输入

双击“表输入”组件 或 右键选择 “编辑步骤”

点击获取SQL查询语句，会弹出界面选择数据表

选择一个数据表后，提示

选择“是”

这里会自动填充获取数据的sql，也可以在这里加上各种where条件，获取需要的数据
点击“确定”

配置输出

如果是表结构一致，则可使用
因为目标数据源也是数据库，所以这里选择表输出。从输出组件选择表输出，拖入转换视图中

然后进行步骤连接。
方式一：按住shift键，鼠标左键点选“输入步骤”，会出现箭头，然后连接到“输出步骤”
方式二：鼠标左键框选输入和输出，然后右键，选择”新建节点连接“，选择”起始步骤“，”目标步骤“

点击“确定”
连接后如下：

双击“表输出”或右键选择“编辑步骤”

选择目标数据库中的数据表，然后点击”确定“

选择表输出，无法配置字段映射，所以前提是表结构一致才可使用。如果是异构表，需要字段映射的，则需要使用 插入/更新 组件

如果输入表和输出表结构不一致，即异构表，则需要使用插入/更新组件。从输出中选择插入/更新拖入转换视图中，然后进行步骤连接，进入输出配置

注意：一定要正确连接步骤，否则这步无法获取输入字段，输出字段

字段映射配置好后如下

点击“确定”

或

然后点击转换视图中的按钮，这个是运行

这个运行是运行一次，完成后就结束了。如果要定时运行，则需要作业。

点击“启动” 会弹出界面 保存 当前转换

输入保存的文件名称，然后点击“Save”即可

每个步骤都显示绿色的箭头，说明没有错误，正确的执行完了转换。也可以在日志输出查看.
日志：完成处理 (I=1, O=1, R=1, W=1, U=0, E=0)中的 I=1 表示 输入 1 行，O=1表示 输出 1 行，R=1 表示 读取 1 行，W=1 表示 写入 1 行
然后看一下数据输出结果
源表

目标表

定时作业

如果需要定时执行同步过程，那么就需要引入作业。在“文件–>新建–>作业” 创建一个作业。
在“通用”中选择Start拖入作业视图中

然后选择转换拖入视图，并进行步骤连接。

双击“转换”或右键选择“编辑作业入口”

点击“确定”
然后选成功组件拖入视图，并连接步骤

双击视图中的Start组件或右键”编辑作业入口“，进行作业调度配置

点击运行视图中的按钮。一个定时作业即完成

定时作业调度期间，程序不能退出！程序退出，作业即停止

至此一个完整的数据处理作业完成了。
3. 案例

3.1 简单同步

本部分对简单同步进行说明。简单同步是指不涉及复杂计算、转换等同步工作。
3.1.1 单表

即一对一同步，A表数据同步到B表，A与B的字段数量、类型、名称可能都不一样，因此需要一些字段类型转换，这都很容易。
处理过程详见2.5章节
3.1.2 多表

即2个及以上的表往一个表同步，同样也需要字段映射、类型转换等操作。
外键关联

这种通过某个字段(外键)关联的表，处理思路是在获取数据时，通过sql联表查询，获取到全部需要的数据。然后用单表同步方式进行处理。
多表合并

如果是异构表的话，获取到每个数据源后，使用Multiway merge join多路合并组件处理


合并后的记录可作为一个单表，然后进行单表同步的处理

合并是笛卡尔积，即A表n条记录，B表n条记录，结果就是n x n条记录，字段是A、B表全部字段，这种方式不建议采用，会消耗更多内存资源。建议拆分成单表同步

如果是同构表的话，可拆分为多个单表同步处理。
3.2 复杂同步

本部分对涉及到数据计算、转换的同步工作进行说明。有些复杂操作，无法直接使用组件进行处理，需要用到Script组件。这里主要对如何使用脚本组件完成数据处理进行说明。
这里先展示一个实际案例

这个过程是多表同步到一个表、涉及到字段类型转换、补充字段和值、数据计算、增补数据。由于计算和增补数据使用内置组件无法完成，因此这里使用了java脚本组件，自定义代码进行数据处理。
这里对字段类型转换、增加列、给某列设置值、java脚本进行说明。
字段类型转换

例如 数字类型 转为 字符串，字符串 转为 日期时间……
从转换中选择字段选择组件

双击“字段选择”或右键选择“编辑步骤”

选择“元数据”，在“字段名称”列选择字段，然后在“类型”列选择目标类型

增加列并设置随机数

在输入中找到生成随机数组件，拖入视图并连接步骤

双击生成随机数或右键选择“编辑步骤”

在“名称”列输入需要增加的字段名，类型选择生成随机数规则

点击“确定”后，运行转换，在preview data处可预览数据，可以看到增加了一列 uid 也有值

将列的值设置为常量

例如将上面随机数组件生成的值设置为常量1。在转换中选择将字段设置为常量组件，并连接步骤

双击“将字段设置为常量”或右键选择“编辑步骤”

在“字段”列选择需要设置的字段，这里选择上一步骤生成的“uid”字段，在“值替换”列输入值。
点击“确定”，运行转换，然后预览数据，可以看到uid的值被替换为1

java脚本

脚本有Java脚本、JavaScript脚本，SQL脚本等。这里使用Java脚本，脚本的目的是处理内置组件处理不了的逻辑。例如有10个地层，但是数据源中只记录了前9个地层，最后一个需要根据计算得到。
拖入Java脚本组件到转换视图中并连接步骤

双击“Java脚本”或右键选择“编辑步骤”

然后展开Code Snippits\Common use

选择Main拖入右侧编辑区，Main是整个脚本处理入口

其默认脚本结构如下

1. public boolean processRow(StepMetaInterface smi, StepDataInterface sdi) throws KettleException {
2.   if (first) {
3.     first = false;
4.         // 代码逻辑区域
5.       
6.     /* TODO: Your code here. (Using info fields)
8.     FieldHelper infoField = get(Fields.Info, "info_field_name");
10.     RowSet infoStream = findInfoRowSet("info_stream_tag");
12.     Object[] infoRow = null;
14.     int infoRowCount = 0;
16.     // Read all rows from info step before calling getRow() method, which returns first row from any
17.     // input rowset. As rowMeta for info and input steps varies getRow() can lead to errors.
18.     while((infoRow = getRowFrom(infoStream)) != null){
20.       // do something with info data
21.       infoRowCount++;
22.     }
23.     */
24.   }
26.   Object[] r = getRow();
28.   if (r == null) {
29.     setOutputDone();
30.     return false;
31.   }
33.   // It is always safest to call createOutputRow() to ensure that your output row's Object[] is large
34.   // enough to handle any new fields you are creating in this step.
35.   r = createOutputRow(r, data.outputRowMeta.size());
37.   /* TODO: Your code here. (See Sample)
39.   // Get the value from an input field
40.   String foobar = get(Fields.In, "a_fieldname").getString(r);
42.   foobar += "bar";
43.    
44.   // Set a value in a new output field
45.   get(Fields.Out, "output_fieldname").setValue(r, foobar);
47.   */
48.   // Send the row on to the next step.
49.   putRow(data.outputRowMeta, r);
51.   return true;
52. }

复制代码

TODO区域就是代码编辑区域，其它是默认脚本函数
点击”确定“，然后再次打开Java脚本，就能看到输入输出字段信息了

完整实现地层计算并补充最后一层的Java脚本代码逻辑如下

1. // 这里是需要用的 java API 所导入的包
2. import java.util.ArrayList;
3. import java.util.HashMap;
4. import java.util.List;
5. import java.util.Map;
6. import java.lang.*;
7. import java.math.BigDecimal;
8. import java.util.*;
10. // 核心处理过程入口
11. public boolean processRow(StepMetaInterface smi, StepDataInterface sdi) throws KettleException {
13. if (first) {
14.     first = false;
15.             logBasic("----------------------字段-------------------------");
16.         String projectCount = "project_count";
17.         String knumber = "knumber";
18.         String depth = "depth";
19.         String dep = "dep";
20.         String layerorder = "layerorder";
21.         String id = "id";
23.                 logBasic("----------------------获取输入流-------------------------");
24.         // 输入数据流 input 是消息步骤中设置的标签名
25.         RowSet infoStream = findInfoRowSet("input");
28.                 Object[] infoRow = null;
29.             int infoRowCount = 0;
31.                 logBasic("----------------------遍历数据流，将其加载到map-------------------------");
32.         // 遍历数据流，将其加载到map，便于操作
33.         // 根据 项目索引+钻孔索引 分组
34.         Map<String, ArrayList<Object[]>> groups = new HashMap<String, ArrayList<Object[]>>();
35.         while((infoRow = getRowFrom(infoStream)) != null){
36.             // 获取字段值
37.                 String prjCode = get(TransformClassBase.Fields.In, projectCount).getString(infoRow);
38.             String drillCode = get(TransformClassBase.Fields.In, knumber).getString(infoRow);
39.             String groupKey = prjCode + drillCode;
41.             if (!groups.containsKey(groupKey)) {
42.                 logBasic("----------------------创建分组-------------------------");
43.                 groups.put(groupKey,new ArrayList<Object[]>());
44.             }
45.                         logBasic("----------------------添加数据到分组-------------------------");
46.             ArrayList<Object[]> objects = (ArrayList<Object[]>)groups.get(groupKey);
47.             objects.add(infoRow);
49.             logBasic("----------------------添加数据到输出流-------------------------");
50.                         // 将当前行拷贝一份
51.             Object[] row=infoRow;
52.             // 创建一个输出行
53.              row = createOutputRow(infoRow, data.outputRowMeta.size());
54.             //putRow(infoStream.getRowMeta(), row);
55.             // 将输出行添加到输出数据集
56.             putRow(data.outputRowMeta, row);
58.                       infoRowCount++;
59.             }
60.            
61.         logBasic("----------------------分组完成，处理最后一条数据-------------------------");
62.         // 将最后一条数据拷贝一份，场地分层索引+1，层底深度dep 赋值为 钻孔深度 depth，然后将此行数数据添加
63.                 Object[] keys = groups.keySet().toArray();
64.                 for (int i = 0; i < keys.length; i++)  {
65.                          String s = keys[i].toString();
66.                          logBasic("----------------------当前分组-----------------------："+ s);       
67.                   ArrayList<Object[]> list = (ArrayList<Object[]>) groups.get(s);
68.                
69.                        
70.               Object[] last = (Object[])list.get(list.size() - 1);         
71.               Object[] newLast=last;
72.             // 设置 layerorder 的值
73.                      String layerorderVal = get(TransformClassBase.Fields.In, layerorder).getString(last);
74.                           BigDecimal v = new BigDecimal(layerorderVal);
75.               v = v.add(new BigDecimal(1));
76.               get(TransformClassBase.Fields.Out, layerorder).setValue(newLast, v);
78.             // 设置 dep 的值
79.               String layerDepVal = get(TransformClassBase.Fields.In, depth).getString(last);
80.                    BigDecimal v2 = new BigDecimal(layerDepVal);
81.               get(TransformClassBase.Fields.Out, dep).setValue(newLast, v2);
82.             
83.                          // 设置id
84.               String idVal = UUID.randomUUID().toString();
85.               get(TransformClassBase.Fields.Out, id).setValue(newLast, idVal);
86.               
87.                           
88.                           logBasic("----------------------添加数据到输出流-------------------------");
89.                           newLast=createOutputRow(newLast, data.outputRowMeta.size());
90.               // 将新的一行数据添加到输出数据集
91.               putRow(data.outputRowMeta, newLast);
93.         }        
94.         
98.     /* TODO: Your code here. (Using info fields)
100.     FieldHelper infoField = get(Fields.Info, "info_field_name");
102.     RowSet infoStream = findInfoRowSet("info_stream_tag");
104.     Object[] infoRow = null;
106.     int infoRowCount = 0;
108.     // Read all rows from info step before calling getRow() method, which returns first row from any
109.     // input rowset. As rowMeta for info and input steps varies getRow() can lead to errors.
110.     while((infoRow = getRowFrom(infoStream)) != null){
112.       // do something with info data
113.       infoRowCount++;
114.     }
115.     */
116.   }
119.   Object[] r = getRow();
120.          logBasic("----------------------getRow-----------------------："+ r);       
123.   if (r == null) {
124.     setOutputDone();
125.     return false;
126.   }
128.   // It is always safest to call createOutputRow() to ensure that your output row's Object[] is large
129.   // enough to handle any new fields you are creating in this step.
130.   r = createOutputRow(r, data.outputRowMeta.size());
131. logBasic("----------------------createOutputRow-----------------------："+ r);       
133.   /* TODO: Your code here. (See Sample)
135.   // Get the value from an input field
136.   String foobar = get(Fields.In, "a_fieldname").getString(r);
138.   foobar += "bar";
139.    
140.   // Set a value in a new output field
141.   get(Fields.Out, "output_fieldname").setValue(r, foobar);
143.   */
144.   // Send the row on to the next step.
145.   putRow(data.outputRowMeta, r);
147.   return true;
148. }

复制代码

至此 Java脚本处理完成。

痛(坑)点总结：
1.脚本编辑区是个文本编辑框，不能像IDEA一样帮助写代码，只能通过日志进行输出验证逻辑
2.建议通用的不涉及pentaho的java代码操作，可以在IDEA中完成，然后拷贝到脚本编辑区。例如需要导入的包就是在IDEA中通过智能导入，然后拷贝的

验证一下数据，图中标记的行，就是根据前2行数据计算而来，然后进行补充的。在数据源中只记录了前2行数据。


免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！