1

使用 pentaho 水壶（也称为 pdi），我有一个合并两个数据流的“加入行（笛卡尔积）”步骤。

第一个和第二个流都附加了一个数值。例如，

Stream 1 - Values 1, 3, 5
Stream 2 - Values 2, 4, 6

我想加入两个流以获得以下输出：

(1, 2)
(3, 4)
(5, 6)

我将正确的输出描述为让流 1 选择大于流 1 的值的最小值。

在 Join Rows 步骤中，我可以指定流 2 的值大于流 1 的值。不幸的是，这会产生以下不正确的结果：

(1, 2)
(1, 4)
(1, 6)
(3, 4)
(3, 6)
(5, 6)

我应该使用不同的步骤来代替 Kettle 中的“加入行”吗？还是我错过了连接行步骤的设置？

注意：我还研究了使用 Stream Lookup 步骤，但它仅适用于 equals 而不适用于我的逻辑。

谢谢。

4

1 回答 1

1

你已经成功了一半。

  • 您有两个输入：Stream1 (1, 3, 5) 和 Stream2(2, 4, 6)
  • 您在 value(stream2) > value(stream1) 上加入行（确保在加入之前对它们进行排序）
  • 您对 {value(Stream1), value(Stream2) 上的结果流进行排序，这给了您
(1, 2)
(1, 4)
(1, 6)
(3, 4)
(3, 6)
(5, 6)
  • 放置“添加值字段更改序列”步骤并将“如果以下字段的值更改则初始化序列”设置为 value(Stream1)。结果流是：
(Stream1, Stream2, result)
(1, 2, 1)
(1, 4, 2)
(1, 6, 3)
(3, 4, 1)
(3, 6, 2)
(5, 6, 1)
  • 放置一个过滤步骤并过滤“结果= 1”。
  • 来自过滤器“真”分支的结果流是期望的结果。

我用解决方案上传了“example.ktr”（我使用了 Kettle 4.3. 版本）：

例子.ktr

于 2012-12-07T12:17:41.140 回答