如何在 Pig 相同模式中加入 2 个数据集

Question

您好，我对 Pig 编程比较陌生，遇到了一个我很难解决的问题：

我有 2 个数据集

A: (accountId:chararray, title:chararray,genre:chararray)

("A123", "Harry Potter", "Action/Adventure")
("A123", "Sherlock Holmes", "Mystery")
("B456", "James Bond", "Action")
("B456", "Hamlet", "Drama")

B: (accountId:chararray, title:chararray, 流派:chararray)

("B456", "Percy Jackson", "Action/Adventure")
("B456", "Elementary", "Mystery")
("A123", "Divergent", "Action")
("A123", "Downton Abbey", "Drama")

我想要的结果应该是

(accountId:charray, {(),(),...}

(A123, {("A123", "Harry Potter", "Action/Adventure"),
        ("A123", "Sherlock Holmes", "Mystery"),
        ("A123", "Divergent", "Action"),
        ("A123", "Downton Abbey", "Drama")
        })

(B456, {("B456", "James Bond", "Action"),
        ("B456", "Hamlet", "Drama"),
        ("B456", "Percy Jackson", "Action/Adventure"),
        ("B456", "Elementary", "Mystery")
        })

目前我在做：

ANS = JOIN A BY accountId, B BY accountId;

但结果看起来像

SCHEMA: (accountId:chararray, {(accountId:chararray, title:chararray,genre:chararray), ...})

(B456, {("B456", "James Bond", "Action"),
        ("B456", "Hamlet", "Drama")}
       "B456", {
        ("B456", "Percy Jackson", "Action/Adventure"),
        ("B456", "Elementary", "Mystery")
        })

知道我可能做错了什么。

Answer 1

试试这个：

-- IMPORTANT: register datafu.jar
define BagConcat datafu.pig.bags.BagConcat();
A = load 'A' using PigStorage(',') as (id:chararray, title:chararray, genre:chararray);      
B = load 'B' using PigStorage(',') as (id:chararray, title:chararray, genre:chararray);   
C = cogroup A by id, B by id;
D = foreach C generate BagConcat(A, B);
dump D;

JOIN 将简单地从您的两个关系中加入行。你想要完成两件事：

• 对每个关系中属于同一帐户的所有行进行分组
• 加入两个“分组”关系（仅获取两个关系中都存在的 ID）

这两个动作由 COGROUP 执行。我读到的最好的解释在这里：http://joshualande.com/cogroup-in-pig/

您的关系现在将包含组键 (ID) 和两个包（一个来自 A，一个来自 B），每个包都包含原始关系中的行；将它们“合并”到一个包中的方法是使用 datafu.jar 中的 BagConcat 函数。 datafu 是一个 PIG UDF 库，里面充满了好东西。你可以在这里阅读：http://datafu.incubator.apache.org/docs/datafu/guide/bag-operations.html