爲Pentaho數據集成中的數據流填充數據缺口，有可能嗎？

Question

我有一個帶有貨幣兌換EUR-USD的CSV文件。該文件從Bank of Canada下載。自2013年10月10日起，我從CSV下載數據。

儘管如此，數據中仍然存在空白，沒有轉換率的日子。

我一直在爭取（用勺子水壺第一天）來找出一個簡單的（但一般的）方式來彌補差距，比如說最後一個非空值。我設法完成這個的唯一方法是鏈接4「獲取前面的行字段」，並使用計算器中的NVL獲取第一個非空值。但是，只有在流中的差距不超過4行時纔有效。

的形象代表的轉型：

我的第一個問題簡化爲：是否有做插/外插在縫隙流的一般方法？

我試圖使用「修改後的JavaScript值」，但API仍然逃脫我。此外，似乎這一步只有MapReduce組合的Map部分，我可能需要兩者。

所以，我的第二個問題是：有沒有一種方法來編程一個不是Java的語言（Scala，Clojure，Jython或JS）的MapReduce組合？

Answer 1

您可以使用以下三個步驟的組合：

1）分析查詢 - 讓您行之前或當前行後取的場N的值;在你的情況下，你會想要提前1日（下一個可用日期）的日期

2）計算器 - 確定該行的上一個日期，用它來計算日期之間的天;

3）計算的場number_of_clones作爲DBD-1（天的缺失數;

4）使用該字段在克隆行步驟到多個行多次必要的;添加克隆編號字段

5）將克隆編號添加爲日期到日期，並獲取它引用的日期。

此外，分析查詢步驟允許你「分組依據」字段中指定一個字段作爲，所以，如果你有X-率美元，那麼你有X-率英鎊，最終美元X-率一天將作爲下一個值檢索null。

下面是一個示例KTR文件：

數據網格步在那裏產生了一些數據差距幾行：

的分析查詢獲取下一個日期，對於相同的貨幣值

然後計算器步驟計算有多少行丟失。需要注意的是每一種貨幣的最後一天將有null的價值，所以我們需要的調整，並用0代替（NVL（A，B）返回B：要爲null，否則）

克隆行：取一排並創建副本。

的clone_number場使我們可以計算該行指實際日期

最後，這裏的數據。你想要的字段是new_date，currency和exchange_rate。使用選擇值來重新排序字段列表並擺脫那些您不再需要的字段。

正如您所看到的，現在我們有2014-01-03和2014-01-04的數據，使用之前已知的值。

Answer 2

儘管這並不完全符合您的要求，但您可以通過使用具有通用功能的用戶定義Java類組件實現您的目標。用此組件的單個實例替換Get previous row fields到Non-values in row的步驟。在部分類 - 組件的處理器插入下面的代碼：

Object[] previousRow; 

public boolean processRow(StepMetaInterface smi, StepDataInterface sdi) throws KettleException 
{ 
    // First, get a row from the default input hop 
    Object[] r = getRow(); 

    // If the row object is null, we are done processing. 
    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. 
    Object[] outputRow = createOutputRow(r, data.outputRowMeta.size()); 

    // copy all input fields to the output fields 

    for (int i=0; i < getInputRowMeta().size(); i++) { 
     logBasic(data.inputRowMeta.getString(r, i)); 
     if (data.inputRowMeta.getString(r, i) == null && (previousRow != null)) { 
      // if the current field is empty take it from the previous row 
      outputRow[i] = previousRow[i]; 
     } 
     else { 
      // otherwise use the current row 
      outputRow[i] = r[i]; 
     } 

    } 

    putRow(data.outputRowMeta, outputRow); 
    // store the current row as future previous row 
    previousRow = data.outputRowMeta.cloneRow(outputRow); 

    return true; 
} 

的JANINO類始終保持前一行的副本，以填補當前行的空字段。

以下測試設置演示組件的使用。在最簡單的情況下，我們處理來自一個CSV文件中讀取流：

輸入文件配置如下：

，幷包含下列數據

NUMBER;STRING;DATE;CURRENCY 
1;A;01.02.2014;12,5 
2;B;;13,5 
;;03.12.2001; 
4;;; 
5;C;; 
6;;20.03.2005;18,2 
7;D;; 

的配置用戶定義的Java類分量如下：

輸出文本文件包含增強的行 「無間隙」：

NUMBER;STRING;DATE;CURRENCY 
1;A;01.02.2014; 012,50 
2;B;01.02.2014; 013,50 
2;B;03.12.2001; 013,50 
4;B;03.12.2001; 013,50 
5;C;03.12.2001; 013,50 
6;C;20.03.2005; 018,20 
7;D;20.03.2005; 018,20 

注：

• 的成分是測試這四個數據類型，但在priciple它應該適用於所有。
• 它與實際字段數無關。
• 一旦一個字段被填充，它永遠不會被「填充」，這對您的設置來說是不錯的（我猜），但這可能不適用於其他設置。
• 該機制僅適用於字段爲null。只包含空格的字符串可能會破壞它，因此請確保在將所有字符串管道輸送到組件之前修剪所有字符串。

該代碼是使用http://wiki.pentaho.com/display/EAI/User+Defined+Java+Class作爲教程編寫的。

附錄

由@manu提供的鏈接包含followng代碼。它包含數字格式的特定處理。請注意，它不再是完全通用的。

Object[] previousRow; 
RowMetaInterface outputMeta; 

public boolean processRow(StepMetaInterface smi, StepDataInterface sdi) throws KettleException 
{ 
    // First, get a row from the default input hop 
    Object[] r = getRow(); 

    // If the row object is null, we are done processing. 
    if (r == null) { 
     setOutputDone(); 
     return false; 
    } 

    if (outputMeta == null) { 
     outputMeta = data.outputRowMeta.clone(); 
    for(int i=0; i < outputMeta.size(); i++) { 
     ValueMetaInterface meta = outputMeta.getValueMeta(i); 
     if (meta.getType() == ValueMetaInterface.TYPE_NUMBER) { 
      meta.setPrecision(4); 
      meta.setConversionMask("#.####"); 
     } 
    } 
} 

// 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. 
Object[] outputRow = createOutputRow(r, data.outputRowMeta.size()); 

// copy all input fields to the output fields 

for (int i=0; i < getInputRowMeta().size(); i++) { 
    if ((r[i] == null) && (previousRow != null)) { 
     // if the current field is empty take it from the previous row 
     outputRow[i] = previousRow[i]; 
    } 
    else { 
     // otherwise use the current row 
     outputRow[i] = r[i]; 
    }  
} 

putRow(outputMeta, outputRow); 
// store the current row as future previous row 
previousRow = outputMeta.cloneRow(outputRow); 

return true; 
}