Pig 中 GROUP BY 中的条件过滤器答案

【问题标题】：Conditional Filter in GROUP BY in PigPig 中 GROUP BY 中的条件过滤器
【发布时间】：2013-08-13 15:56:11
【问题描述】：

我有以下数据集，如果它们具有相同的键，我需要将它们合并为一个。同时，我需要在分组的多个元组中进行选择。

例如

A = LOAD 'data.txt' AS (f1:int, f2:chararray, f3:int, f4:int);
G = GROUP A BY (f1, f2);
DUMP G;

((1,N1),{(1,N1,1,10),(1,N1,2,15)})
((2,N1),{(2,N1,1,10)})
((3,N1),{(3,N1,1,10),(3,N1,2,15)})
((4,N2),{(4,N2,1,10)})
((5,N3),{(5,N3,1,10),(5,N3,2,20)})

现在，我想选择收集袋中是否有多个元组，我只想过滤那些具有 f3==2 的元组。这是我想要的最终数据：

((1,N1),{(1,N1,2,15)})   -- f3==2, f3==1 is removed from this set
((2,N1),{(2,N1,1,10)})
((3,N1),{(3,N1,2,15)})   -- f3==2, f3==1 is removed from this bag
((4,N2),{(4,N2,1,10)})
((5,N3),{(5,N3,2,10)})

知道如何实现吗？

【问题讨论】：

我的方法是计算分组包的计数，然后过滤计数 > 1 的行和计数 ==1 的行。对于 count == 1 的集合，我直接将其展平，而对于第二个集合，我使用了嵌套过滤器。但是现在，pig 正在抱怨函数重载。我收到此错误 - ERROR 1065: Found more than one load function to use: [PigStorage, org.apache.pig.builtin.PigStorage]

标签： hadoop apache-pig hadoop-streaming

【解决方案1】：

我按照上面评论中指定的方式做了。这是我的做法。

A = LOAD 'group.txt' USING PigStorage(',') AS (f1:int, f2:chararray, f3:int, f4:int);
G = GROUP A BY (f1, f2);

CNT = FOREACH G GENERATE group, COUNT($1) AS cnt, $1;

SPLIT CNT INTO
    CNT1 IF (cnt > 1),
    CNT2 IF (cnt == 1);


M1 = FOREACH CNT1 {
        row = FILTER $2 BY (f3 == 2);
        GENERATE FLATTEN(row);
     };

M2 = FOREACH CNT2 GENERATE FLATTEN($2);

O = UNION M1, M2;

DUMP O;

(2,N1,1,10)
(4,N2,1,10)
(1,N1,2,15)
(3,N1,2,15)
(5,N3,2,20)

【讨论】：

这太棒了。让我免于痛苦。