从Pig Latin中的组运算中获取过滤后的元组

Question

我有一些交易数据，比如

txn_id、receiver_userid、sender_userid、金额

1,1,2,50

2,1100

3,1,2500
4,5,3100
5,5,3200
6,5,3300
7,6,2200

8,6,1200

现在我想找到所有从同一个用户那里收到2笔以上交易的接收者用户，我已经开始把我的PIG工作写为

txnrecord = LOAD './txndata' USING PigStorage(',') AS (txn_id:int, receiver_userid:int, sender_userid:int, amount:int);
grptxn1 = GROUP txnrecord BY (receiver_userid, sender_userid);
txncount = FOREACH grptxn1 GENERATE FLATTEN(group) as (receiver_userid, sender_userid), COUNT(txnrecord) as num_txns, SUM(txnrecord.amount) as total_sum;
txncount1 = FILTER txncount by num_txns > 2;
dump txncount1;

上面给了我正确的组聚合，但我的附加要求是

1) 查找聚合的组记录及其关联的元组集(单个txn)，例如-如果我的组聚合说userid 1从userid 2接收了3个事务，我需要将所有三个元组存储在另一个数据文件中。

2) 组聚合不匹配>2事务条件应被忽略(此处应忽略最后两条记录)

3) 我想将序列分配给我的组聚合，并且相同的序列应该用作它们关联的事务元组中的链接键(以标识这三个事务记录与特定的组聚合关联)。

我正在尝试使用各种功能，但到目前为止运气不佳。

感谢您的任何帮助。

Answer 1

您可以携带GROUP by创建的BAG，它们包含所有原始列，以检查它是否执行

DESCRIBE grptxn1;

回答要求1和2：

txnrecord = LOAD './txndata' USING PigStorage(',') AS (txn_id:int, receiver_userid:int, sender_userid:int, amount:int);
grptxn1 = GROUP txnrecord BY (receiver_userid, sender_userid);
txncount = FOREACH grptxn1 GENERATE FLATTEN(group) as (receiver_userid, sender_userid), 
txnrecord, -- carry bags through the filter
COUNT(txnrecord) as num_txns, SUM(txnrecord.amount) as total_sum ;
txncount1 = FILTER txncount by num_txns > 2;
tran_dump = FOREACH  txncount1 GENERATE FLATTEN(txnrecord);
STORE tran_dump INTO 'another data file';
txncount2 = FOREACH txncount1 GENERATE (receiver_userid, sender_userid, num_txns, total_sum);
dump txncount2;

在MapReduce中，如果不让需求3变得非常慢或使用一些网络ID代理，它就不容易实现。可能您不需要它，因为FLATTEN(txnrecord)将转储输入文件中的所有列。