哈託普豬腳 - 與條件

Question

聯盟我對豬完全陌生。我想聯合使用iid字段的兩個文件A和B，但我不希望輸出具有A沒有的任何IID（來自B）。這似乎很簡單，但我無法弄清楚如何正確地做到這一點。

這裏是一個只有工會我的示例代碼：

a = load '$input' as (iid:int, field:chararray, v1:chararray, v2:chararray, v3:chararray); 
b = load '$data' as (iid:int, field:chararray, v1:chararray, v2:chararray, v3:chararray); 
out = union onschema a,b; 
singled = distinct out; 
ordered = order singled by iid; 
store ordered into '$output'; 

下面是一個只有3列樣本數據來形容我的期望。請注意，字段實際上是製表符分隔的。

樣品數據答：

1 Name Tom Linkon 
1 Title Professor 
2 Name Whatever 
2 Title Worker 

數據採樣B：

1 City New York 
2 City Columbus 
3 City Fake fake 
4 City Blah Bla 

樣本輸出

1 Name Tom Linkon 
1 Title Professor 
1 City New York 
2 Name Whatever 
2 Title Worker 
2 City Columbus 

非常感謝您的幫助！

Answer 1

使用COGROUP組織記錄使用相同的密鑰，但要避免JOIN的不良跨產品。然後FILTER用b的記錄袋是否爲空，拆回兩個關係，並做UNION：

a = load '$input' as (iid:int, field:chararray, v1:chararray, v2:chararray, v3:chararray); 
b = load '$data' as (iid:int, field:chararray, v1:chararray, v2:chararray, v3:chararray); 
c = COGROUP a BY iid, b BY iid; 
c_filt = FILTER c BY NOT IsEmpty(b); 
a_new = FOREACH c_filt GENERATE group AS iid, FLATTEN(a); 
b_new = FOREACH c_filt GENERATE group AS iid, FLATTEN(b); 
out = UNION ONSCHEMA a_new, b_new; 
singled = DISTINCT out; 
STORE (ORDER singled BY iid) INTO '$output'; 

不過，我不是這個解決方案的粉絲 - 這是太多的線條和新關係如此簡單的操作。真正需要的是將兩個袋子合併成一個袋子的方式。豬顯然不提供這個（但如果是這樣，請回答this SO question）。你可以寫一個簡單的UDF要做到這一點，雖然：

public class MERGE extends EvalFunc<DataBag> { 
    public DataBag exec(Tuple input) throws IOException { 
     DataBag b = new DefaultDataBag(); 
     try { 
      if (input != null) 
       for (int i = 0; i < input.size(); i++) 
        b.addAll((DataBag) input.get(i)); 
     } catch (Exception e) { return null; } 
     return b; 
    } 
} 

有了這個UDF在手，溶液變成：

a = load '$input' as (iid:int, field:chararray, v1:chararray, v2:chararray, v3:chararray); 
b = load '$data' as (iid:int, field:chararray, v1:chararray, v2:chararray, v3:chararray); 
c = FOREACH (COGROUP a BY iid, b BY iid) GENERATE group AS iid, MERGE(a,b) AS bag; 
out = FOREACH c { 
    uniq = DISTINCT bag; 
    GENERATE iid, FLATTEN(bag); 
}; 
STORE (ORDER out BY iid) INTO '$output'; 

這種方法的另一個好處是，如果你有幾個輸入端，你在COGROUP之後不需要做幾個FOREACH。只需添加更多的參數，以MERGE：

c = FOREACH (COGROUP a BY iid, b BY iid, ..., z BY iid) 
    GENERATE group AS iid, MERGE(a,b,...,z) AS bag; 

Answer 2

這兒，這應該解決您的問題：

f1 = LOAD '/user/hadoop/f1' USING PigStorage('\t') AS (id_f1:int, key_f1:chararray, value_f1:chararray); 
f2 = LOAD '/user/hadoop/f2' USING PigStorage('\t') AS (id_f2:int, key_f2:chararray, value_f2:chararray); 
f3 = JOIN f1 by id_f1 LEFT OUTER, f2 BY id_f2; 
f4 = FOREACH f3 GENERATE id_f1, key_f1, value_f1; 
f5 = FOREACH f3 GENERATE id_f2, key_f2, value_f2; 
f6 = UNION f4, f5; 
f7 = DISTINCT f6; 
f8 = ORDER f7 BY $0; 
DUMP f8;