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
tar -zxvf 安装包路径 -C 目标路径
Linux
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
运行 Spoon.bat
MacOS
运行 Spoon.sh
Linux
运行 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是整个脚本处理入口
其默认脚本结构如下
public boolean processRow(StepMetaInterface smi, StepDataInterface sdi) throws KettleException {
if (first) {
first = false;
// 代码逻辑区域
/* TODO: Your code here. (Using info fields)
FieldHelper infoField = get(Fields.Info, "info_field_name");
RowSet infoStream = findInfoRowSet("info_stream_tag");
Object[] infoRow = null;
int infoRowCount = 0;
// Read all rows from info step before calling getRow() method, which returns first row from any
// input rowset. As rowMeta for info and input steps varies getRow() can lead to errors.
while((infoRow = getRowFrom(infoStream)) != null){
// do something with info data
infoRowCount++;
}
*/
}
Object[] r = getRow();
if (r == null) {
setOutputDone();
return false;
}
// It is always safest to call createOutputRow() to ensure that your output row's Object[] is large
// enough to handle any new fields you are creating in this step.
r = createOutputRow(r, data.outputRowMeta.size());
/* TODO: Your code here. (See Sample)
// Get the value from an input field
String foobar = get(Fields.In, "a_fieldname").getString(r);
foobar += "bar";
// Set a value in a new output field
get(Fields.Out, "output_fieldname").setValue(r, foobar);
*/
// Send the row on to the next step.
putRow(data.outputRowMeta, r);
return true;
}
TODO区域就是代码编辑区域，其它是默认脚本函数
点击”确定“，然后再次打开Java脚本，就能看到输入输出字段信息了
完整实现地层计算并补充最后一层的Java脚本代码逻辑如下
// 这里是需要用的 java API 所导入的包
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.lang.*;
import java.math.BigDecimal;
import java.util.*;
// 核心处理过程入口
public boolean processRow(StepMetaInterface smi, StepDataInterface sdi) throws KettleException {
if (first) {
first = false;
logBasic("----------------------字段-------------------------");
String projectCount = "project_count";
String knumber = "knumber";
String depth = "depth";
String dep = "dep";
String layerorder = "layerorder";
String id = "id";
logBasic("----------------------获取输入流-------------------------");
// 输入数据流 input 是消息步骤中设置的标签名
RowSet infoStream = findInfoRowSet("input");
Object[] infoRow = null;
int infoRowCount = 0;
logBasic("----------------------遍历数据流，将其加载到map-------------------------");
// 遍历数据流，将其加载到map，便于操作
// 根据 项目索引+钻孔索引 分组
Map<String, ArrayList<Object[]>> groups = new HashMap<String, ArrayList<Object[]>>();
while((infoRow = getRowFrom(infoStream)) != null){
// 获取字段值
String prjCode = get(TransformClassBase.Fields.In, projectCount).getString(infoRow);
String drillCode = get(TransformClassBase.Fields.In, knumber).getString(infoRow);
String groupKey = prjCode + drillCode;
if (!groups.containsKey(groupKey)) {
logBasic("----------------------创建分组-------------------------");
groups.put(groupKey,new ArrayList<Object[]>());
}
logBasic("----------------------添加数据到分组-------------------------");
ArrayList<Object[]> objects = (ArrayList<Object[]>)groups.get(groupKey);
objects.add(infoRow);
logBasic("----------------------添加数据到输出流-------------------------");
// 将当前行拷贝一份
Object[] row=infoRow;
// 创建一个输出行
row = createOutputRow(infoRow, data.outputRowMeta.size());
//putRow(infoStream.getRowMeta(), row);
// 将输出行添加到输出数据集
putRow(data.outputRowMeta, row);
infoRowCount++;
}
logBasic("----------------------分组完成，处理最后一条数据-------------------------");
// 将最后一条数据拷贝一份，场地分层索引+1，层底深度dep 赋值为 钻孔深度 depth，然后将此行数数据添加
Object[] keys = groups.keySet().toArray();
for (int i = 0; i < keys.length; i++) {
String s = keys[i].toString();
logBasic("----------------------当前分组-----------------------："+ s);
ArrayList<Object[]> list = (ArrayList<Object[]>) groups.get(s);
Object[] last = (Object[])list.get(list.size() - 1);
Object[] newLast=last;
// 设置 layerorder 的值
String layerorderVal = get(TransformClassBase.Fields.In, layerorder).getString(last);
BigDecimal v = new BigDecimal(layerorderVal);
v = v.add(new BigDecimal(1));
get(TransformClassBase.Fields.Out, layerorder).setValue(newLast, v);
// 设置 dep 的值
String layerDepVal = get(TransformClassBase.Fields.In, depth).getString(last);
BigDecimal v2 = new BigDecimal(layerDepVal);
get(TransformClassBase.Fields.Out, dep).setValue(newLast, v2);
// 设置id
String idVal = UUID.randomUUID().toString();
get(TransformClassBase.Fields.Out, id).setValue(newLast, idVal);
logBasic("----------------------添加数据到输出流-------------------------");
newLast=createOutputRow(newLast, data.outputRowMeta.size());
// 将新的一行数据添加到输出数据集
putRow(data.outputRowMeta, newLast);
}
/* TODO: Your code here. (Using info fields)
FieldHelper infoField = get(Fields.Info, "info_field_name");
RowSet infoStream = findInfoRowSet("info_stream_tag");
Object[] infoRow = null;
int infoRowCount = 0;
// Read all rows from info step before calling getRow() method, which returns first row from any
// input rowset. As rowMeta for info and input steps varies getRow() can lead to errors.
while((infoRow = getRowFrom(infoStream)) != null){
// do something with info data
infoRowCount++;
}
*/
}
Object[] r = getRow();
logBasic("----------------------getRow-----------------------："+ r);
if (r == null) {
setOutputDone();
return false;
}
// It is always safest to call createOutputRow() to ensure that your output row's Object[] is large
// enough to handle any new fields you are creating in this step.
r = createOutputRow(r, data.outputRowMeta.size());
logBasic("----------------------createOutputRow-----------------------："+ r);
/* TODO: Your code here. (See Sample)
// Get the value from an input field
String foobar = get(Fields.In, "a_fieldname").getString(r);
foobar += "bar";
// Set a value in a new output field
get(Fields.Out, "output_fieldname").setValue(r, foobar);
*/
// Send the row on to the next step.
putRow(data.outputRowMeta, r);
return true;
}
至此 Java脚本处理完成。
痛(坑)点总结：
1.脚本编辑区是个文本编辑框，不能像IDEA一样帮助写代码，只能通过日志进行输出验证逻辑
2.建议通用的不涉及pentaho的java代码操作，可以在IDEA中完成，然后拷贝到脚本编辑区。例如需要导入的包就是在IDEA中通过智能导入，然后拷贝的
验证一下数据，图中标记的行，就是根据前2行数据计算而来，然后进行补充的。在数据源中只记录了前2行数据。
本文来自博客园，作者：宣君{https://www.nhit.icu/}，转载请注明原文链接：https://www.cnblogs.com/ycit/p/17687558.html
热门相关：总裁别再玩了 重生之嫡女祸妃 嫡嫁千金 一等狂妃：邪王，请接招！ 嫡嫁千金