ToB企服应用市场:ToB评测及商务社交产业平台

标题: pentaho(keetle)使用手册 [打印本页]
作者: 何小豆儿在此    时间: 2023-9-8 14:47
标题: 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由主要四部分组成
1.2 pentaho安装

Windows

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

MacOS
  1. tar -zxvf 安装包路径 -C 目标路径
复制代码
Linux
  1. tar -zxvf 安装包路径 -C 目标路径
复制代码
目录结构

重点目录以及执行文件说明
在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行数据。


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



欢迎光临 ToB企服应用市场:ToB评测及商务社交产业平台 (https://dis.qidao123.com/) Powered by Discuz! X3.4