使用猪脚本

Question

我有数据如下

1,ref1,1200,USD,CR 
2,ref1,1200,USD,DR 
3,ref2,2100,USD,DR 
4,ref2,800,USD,CR 
5,ref2,700,USD,CR 
6,ref2,600,USD,CR 

欲组这些记录，其中FIELD2匹配，和（字段3）匹配和字段5是相反的（也就是说，如果LHS是“CR”然后RHS寻找类似条目应该是“DR”，反之亦然）

如何使用猪脚本来实现这一点？

Answer 1

你也可以做到这方法：

data = LOAD 'myData' USING PigStorage(',') AS 
     (field1: int, field2: chararray, 
     field3: int, field4: chararray, 
     field5: chararray) ; 

B = FOREACH (GROUP data BY (field2, field5)) GENERATE group.field2, data ; 

-- Since in B there will always be two sets of field2 (one for CR and one for DR) 
-- grouping by field2 again will get all pairs of CR and DR 
-- (where the sums are equal of course) 
C = GROUP B BY (field2, SUM(field3)) ; 

在最后一步的模式和输出：

C: {group: (field2: chararray,long),B: {(field2: chararray,data: {(field1: int,field2: chararray,field3: int,field4: chararray,field5: chararray)})}} 
((ref1,1200),{(ref1,{(1,ref1,1200,USD,CR)}),(ref1,{(2,ref1,1200,USD,DR)})}) 
((ref2,2100),{(ref2,{(4,ref2,800,USD,CR),(5,ref2,700,USD,CR),(6,ref2,600,USD,CR)}),(ref2,{(3,ref2,2100,USD,DR)})}) 

输出放一点点unwie LDY的权利，但这样会清除掉：

-- Make sure to look at the schema for C above 
D = FOREACH C { 
     -- B is a bag containing tuples in the form: B: {(field2, data)} 
     -- What we want is to just extract out the data field and nothing else 
     -- so we can go over each tuple in the bag and pull out 
     -- the second element (the data we want).      
     justbag = FOREACH B GENERATE FLATTEN($1) ; 
     -- Without FLATTEN the schema for justbag would be: 
     -- justbag: {(data: (field1, ...))} 
     -- FLATTEN makes it easier to access the fields by removing data: 
     -- justbag: {(field1, ...)} 
     GENERATE justbag ;             
} 

进入这个：

D: {justbag: {(data::field1: int,data::field2: chararray,data::field3: int,data::field4: chararray,data::field5: chararray)}} 
({(1,ref1,1200,USD,CR),(2,ref1,1200,USD,DR)}) 
({(4,ref2,800,USD,CR),(5,ref2,700,USD,CR),(6,ref2,600,USD,CR),(3,ref2,2100,USD,DR)}) 

Answer 2

我不知道，我理解您的需求，但你可以加载数据，分成两组（过滤器/分）和协同组如：

data = load ... as (field1: int, field2: chararray, field3: int, field4: chararray, field5: chararray); 

crs= filter data by field5='CR'; 
crs_grp = group crs by field1; 
crs_agg = foreach crs_grp generate group.field1 as field1, sum(crs.field3); 
drs = filter data by field5='DR'; 
drs_grp = group drs by field1; 
drs_agg = foreach drs_grp generate group.field1 as field1, sum(drs.field3); 
g = COGROUP crs_agg BY (field1, field3), drs_agg BY (field1, field3);