版本: 下个版本

数据流控制

案例说明

数据流（data stream）是一组有序，有起点和终点的字节的数据序列。本案例将展现如何操作和控制数据流，并通过相应组件控制流的流量和流向。

数据准备

本案例中“控制数据流”、“数据分发”和“通过‘Switch/Case’组件控制数据流”三个小节将使用 Pentaho+BI+Platform+Tracking.xls；

本案例中“合并数据”和“通过‘追加流’组件合并数据流”两个小节将使用 progress_b_bouchard.xls 、progress_other_programmers.xls两个数据。

控制数据流

本小节主要展现数据流的控制，每个步骤中的数据以流的方式流向下一步骤。本节将使用6种组件：Excel输入、过滤记录、值映射、排序记录、增加序列和Excel 输出。

Excel输入组件读取Excel文件中的数据；
过滤记录组件控制流的流量（记录数量）；
值映射组件将字段中的数据映射成指定值；
排序记录组件对流中的记录进行排序；
增加序列组件给每条记录添加一个序号；
Excel输出组件将处理的数据输出到Excel文件中；

具体操作如下：

新建转换，从组件库的输入目录下拖拽Excel输入组件至画布。双击组件，填写文件Tab页中的参数：选择表格类型（引擎）为 Excel XSLX，XLS，点击浏览，选择Pentaho+BI+Platform+Tracking.xls文件，点击确定。点击增加，将文件添加到选中的文件列表中，参数设置如下图所示：

完整配置如下图所示：

2. 切换到工作表标签，在工作表列表的右键菜单点击获取工作表名称按钮获取工作表，选择工作表 Sheet1，在起始行和起始列中分别输入 3 和 1。配置如下图所示：

3. 选择字段标签，右击空白文本框并选择获取来自头部的字段。配置结果如下图所示：

4. 选中Excel 输入组件，右击并选择预览查看数据。结果如下图所示：

5. 从“流程”中拖入过滤记录组件至画布，并连接上一步骤。左边的 field 选择 Assignee，操作符选择 =，右边的 value 输入 Unassigned，点击确定。过滤分配状态为未分配的记录。配置如下图所示：

6. 从“转换”中拖入值映射组件并连接上个步骤，连接线类型选择True输出。双击该组件，录入优先级映射规则。使用的字段名选择 Priority，其他配置如下图所示：

7. 拖入排序记录组件并连接上个步骤，使用字段 Priority_Order、Summary 对记录进行排序，配置如下图所示:

8. 拖入增加序列组件并连接上个步骤，生成新的数据列 nr，配置如下图所示：

9. 从“输出”中拖拽三个Excel 输出组件至画布，分别与增加序列组件建立连接，弹出的提示框选择“复制发送”。完整的转换如下图所示：

10. 运行转换，结果如下图所示：

11. 分别预览三个Excel 输出的数据，结果应该是一模一样。如下图所示：

数据分发

本小节主要是在上节的基础上再展现数据流中“复制”和“随机分发”的不同之处。具体操作如下：

新建转换，然后打开上节“控制数据流”中创建的转换并复制所有步骤到新转换中。依次连接步骤，在连接最后三个Excel 输出步骤时，选择随机分发。同时将Excel 输出组件的参数如果文件已存在的值设置为使用现有文件输出，运行转换，结果如下图所示：

2. 分别预览三个Excel 输出步骤，可以看到每个输出步骤的数据都不相同。这时因为，选择随机分发后，数据将会分成 n（Excel 输出步骤的个数）等份随机分发至每个Excel 输出步骤。

通过按值分发(Switch/Case)组件控制数据流

本小节主要展示用按值分发组件来控制数据流。当数据流向有多个选择，并且数据流向有条件限制时，可以使用按值分发制定多个条件，来限制数据流的流向。具体操作如下：

在“输入”中拖拽Excel 输入组件至画布，双击该组件，按照下图所示步骤进行配置：

结果如下图

选择工作表标签，并点击文本框下面的获取工作表名称按钮获取工作表，然后分别在起始行和起始列输入 3 和 1。配置如下图所示：

3. 选择字段标签，右击空白文本框并选择获取来自头部的字段，结果如下图所示：

4. 在“流程”中拖入过滤记录组件并连接上一步骤，配置如下图所示：

5. 拖入按值分发组件并连接上一步骤，拖入四个Excel 输出组件，分别和 按值分发连接。整个转换视图如下图所示：

6. 双击按值分发，Switch字段选择Priority，Case值数据类型选择 String，其余参数配置如下图所示：

7. 运行转换，运行结果如下图所示：

8. 预览四个Excel 输出的数据，可以查看每个输出结果对应于按值分发所设定的条件。

合并数据

本小节主要展现通过增加序列组件来合并数据流。

读取数据后，首先过滤【Progress】字段非空的数据，再将【Progress】字段进行降序排序。数据排好序以后，为数据集添加【Programmer】常量字段，并保留所需字段。

读取第二个数据后使用过滤【Progress】非空数据，再给数据排好序。最后，处理好的两个数据集进行增加序列步骤。具体操作如下：

在“输入”中拖入Excel 输入组件，配置如下图所示：

配置结果如下

选择工作表标签，点击获取工作表名称按钮获取工作表，如下图所示：

3. 选择字段标签，右击空白文本框并选择获取来自头部的字段，如下图所示：

4. 从“流程”中拖入过滤记录组件并连接至上一步骤，左边的 field 选择 Progress，操作符选择 IS NOT NULL。配置如下图所示：

5. 使用排序记录组件对记录按 Progress 字段进行升序排序。从“转换”中拖入排序记录组件并连接上一步骤，连接线类型选择True输出，配置如下图所示：

6. 拖入增加常量组件并连接上一步骤，双击步骤进入配置界面，点击新增字段按钮，新增Programmer字段，类型选择String，值为“Benjamin Bouchard”如下图所示：

拖入字段选择组件并连接上一步骤，右键空白区域—>选择获取选择字段，并配置如下图所示：

8. 连接好所有步骤的视图如下图所示：

拖入Excel 输入组件，配置如下图所示：

配置结果如下

选择工作表标签，点击获取工作表名称获取工作表，如下图所示：

11. 选择字段标签，右击空白文本框并选择获取来自头部的字段，如下图所示：

12. 拖入过滤记录字段并连接上一步骤，配置如下图所示：

13. 拖入排序记录组件并连接至上一步骤，配置如下图所示：

14. 在“画布”中拖入增加序列组件并连接至上一步骤，配置如下图所示：

15. 建立从字段选择到增加序列之间的连接，整个转换视图如下图所示：

16. 运行转换，结果如下图所示：两个Excel文件共有 276+125=401 条数据，经过过滤记录过滤后还有 10+9=19 条数据，经过增加序列步骤合并后，共有 19 条数据输出。

17. 选择增加序列，右击并预览数据，可以观察到合并的数据，结果如下图所示：

使用“追加流”组件合并数据流

本小节主要是在上节的基础上再通过追加流组件来演示如何合并数据流。通过追加流组件，能够把数据格式一致的数据流合并在一起。具体操作如下：

新建转换，复制上一节“合并数据”转换中所有步骤到新建的转换中；
删除字段选择、排序记录到增加序列之间的连接；
在“流程”中拖拽追加流组件至画布，连接字段选择、排序记录，然后连接增加序列组件。整个转换视图如下图所示：

双击“追加流”组件，配置如下图所示：

注意：追加流组件的两个数据流中的字段类型、名称及顺序需一致

运行组件，结果如下图所示：

6. 点击增加序列并预览数据，结果如下图所示：

脏数据处理

本小节主要是展示错误数据的处理。在处理真实的数据集时，数据往往并不时完美无瑕的，其中总是夹杂的一些“脏”数据。在数据处理的过程中，不能一味的把错误的数据丢弃，必须制定完善的错误处理步骤来处理那些“脏”数据。

首先通过字段选择步骤将【estimated】的数据类型转为 Integer 类型。但是在第 6 行数据当中存在一个“脏”数据，会阻碍组件的转换。这时，为了处理这个错误，可以将错误输出发送到写日志步骤，然后再做进一步的处理，紧接着用增加常量为错误字段定义默认值。本小节中将【estimated】的默认值设为 180。具体操作如下：