以下是脚本的一个版本：

inpt = load '/pig_data/pig_fun/input/from_senders.txt' as (email_id:chararray, from:chararray, to:bag{recipients:tuple(recipient:chararray)});

pivot = foreach inpt generate from, FLATTEN(to);
pivot = foreach pivot generate from, to::recipient as recipient;
dump pivot;
/*
(user1@example.com,friend1@example.com)
(user1@example.com,friend2@example.com)
(user1@example.com,friend3@myusers.com)
(user1@example.com,friend1@example.com)
(user1@example.com,friend4@example.com)
(user1@example.com,friend5@example.com)
(user2@example.com,friend2@example.com)
(user2@example.com,friend4@example.com)

* /

grp = group pivot by (from, recipient);
with_count = foreach grp generate FLATTEN(group), COUNT(pivot) as count;
dump with_count;
/*
(user1@example.com,friend1@example.com,2)
(user1@example.com,friend2@example.com,1)
(user1@example.com,friend3@myusers.com,1)
(user1@example.com,friend4@example.com,1)
(user1@example.com,friend5@example.com,1)
(user2@example.com,friend2@example.com,1)
(user2@example.com,friend4@example.com,1)

* /

to_bag = group with_count by from;
result = foreach to_bag {
    order_by_count = order with_count by count desc;
    generate group as from, order_by_count.(recipient, count);
};
dump result;
/*
(user1@example.com,{(friend1@example.com,2),(friend2@example.com,1),(friend3@myusers.com,1),(friend4@example.com,1),(friend5@example.com,1)})
(user2@example.com,{(friend2@example.com,1),(friend4@example.com,1)})

* /

希望有帮助。