Pentaho中缺失值的插值

Question

我正在尝试使用 Pentaho pdi 填充缺失值。

输入：

期望的输出：

目前只找到Filling data gaps in a stream in Pentaho Data Integration, is it possible?，但它填充了最后一个已知值。

可能，我认为我可以使用上述解决方案，我还将下一个金额与下一个日期一起添加到分析查询中。然后，我在克隆步骤中添加了标志，并将原始结果从输入过滤到 Dummy 并将生成的结果（来自计算器）到计算器（目前）。然后，潜在地，我可以将该单独的流转储到数据库中的临时表并运行将执行滚动减法的 sql 查询。我也在调查 javascript 步骤。

我忽略了 Python 或 R Executor 步骤，因为最后我将在 aws vm 上运行该作业，并且我已经预见到安装过程中会遇到的痛苦。

您有什么建议？有没有简单的插值方法？

针对问题进行了更新

Answer 1

您的链接中提供的方法在我的测试中确实有效（尽管我使用LAG 而不是LEAD 来完成您的任务）。在这里，我不打算复制该方法，只是为您提供另一种选择，即使用 JavaScript 构建您可能还可以扩展到其他应用程序的逻辑：

在下面的测试中（在 PDI-8.0 上测试），转换有 5 个步骤，见下文

1. Data Grid 步骤来创建包含三个字段的测试数据：date、account number 和 amount
2. 对行进行排序，根据account number 和date 对行进行排序。这是Analytic Query 步骤所必需的，如果您的源数据已经排序，则跳过此步骤
3. 分析查询步骤，见下文，再创建两个字段：prev_date 和 prev_amount

4. Modified Java Script Value步骤，添加如下代码，此步骤无需其他配置：

   var days_diff = dateDiff(prev_date, date, "d")
   
   if (days_diff > 0) {
       /* retrieve index for two fields: 'date', 'amount' 
        * and modify their values accordingly
        */
       var idx_date = getInputRowMeta().indexOfValue("date")
       var idx_amount = getInputRowMeta().indexOfValue("amount")
   
       /* amount to increment by each row */
       var delta_amount = (amount - prev_amount)/days_diff
   
       for (var i = 1; i < days_diff; i++) {
           newRow = createRowCopy(getOutputRowMeta().size());
           newRow[idx_date]   = dateAdd(prev_date, "d", i);
           newRow[idx_amount] = prev_amount + delta_amount * i;
           putRow(newRow);
       } 
   }
   

5. 选择值步骤删除不需要的字段，即：prev_date、prev_amount

运行转换，您将在Modified Java Script Value 步骤的Preview data 选项卡下显示以下内容：

更新：

根据您的 cmets，您可以执行以下操作，假设您有一个新字段 account_type：

1. 在分析查询步骤中，添加一个新字段prev_account_type，类似于其他两个prev_字段，只是来自不同的Subject：account_type

2. 在Modified Java Script Value步骤中，需要获取account_type的Row index并修改计算delta_amount的逻辑，所以当prev_account_type与当前account_type不同时， delta_amount 为零，见以下代码：

   var days_diff = dateDiff(prev_date, date, "d")
   
   if (days_diff > 0) {
       /* retrieve index for three fields: 'date', 'amount', 'account_type' */
       var idx_date     = getInputRowMeta().indexOfValue("date")
       var idx_amount   = getInputRowMeta().indexOfValue("amount")
       var idx_act_type = getInputRowMeta().indexOfValue("account_type")
   
       /* amount to increment by each row */
       var delta_amount = prev_account_type.equals(account_type) ? (amount - prev_amount)/days_diff : 0;
   
       /* copy the current Row into newRow and modify fields accordingly */
       for (var i = 1; i < days_diff; i++) {
           newRow = createRowCopy(getOutputRowMeta().size());
           newRow[idx_date]     = dateAdd(prev_date, "d", i);
           newRow[idx_amount]   = prev_amount + delta_amount * i;
           newRow[idx_act_type] = prev_account_type;
           putRow(newRow);
       } 
   }
   

注意：调用 Javascript 解释器确实会影响性能，因此如果这对您很重要，请坚持使用您提供的链接中的方法。