java - 如何为 group by 编写 pig 代数 udf

Question

我想编写一个 pig 代码来执行分组并生成 31 个字段的总和，但在此之前我需要做一些自定义处理，为此我编写了一个 eval 函数。我想如果我可以将 GROUP 和 SUM 操作包含到 UDF 中，我可以让它运行得更快。要做到这一点，我可以使用代数 UDF 如果是，我的 inital()、intermed() 和 final() 的返回模式会是什么样子，如果不是，我还能如何实现它。下面是我的代码，谢谢。

a = LOAD './a' using PigStorage('|') AS (val:int, grp1, grp2, amt1:long, amt2:long, amt3 ... amt31:long);

b = FOREACH a GENERATE myudfs.Custom(val) AS custom_val, grp1, grp2, amt1 ... amt31;

c = GROUP b BY (custom_val,grp1, grp2);

d = FOREACH c GENERATE group, SUM(b.amt1) ... SUM(b.amt31);

store d into './op';

Answer 1

如何在 UDF 中执行 GROUP...？

GROUP 在 Pig 中被翻译成一个 MapReduce 作业(该作业的中间键将从 custom_val、grp1、grp2 组合而来)。

迭代(FOREACH)特定组的整个元组列表的能力正在 Reducer 中完成。

代数 UDF 不会“包含 GROUP”，但会作为 GROUP 聚合的一部分执行。所以我认为代数在这里不相关。

我想您在这里可能做的唯一优化是对原始 val 进行分组，并仅在 GROUP 之后调用 myudfs.Custom(val)。

假设您的 UDF 是一个 injective function .

a = LOAD './a' using PigStorage('|') AS (val:int, grp1, grp2, amt1:long, amt2:long, amt3 ... amt31:long);
c = GROUP b BY (val,grp1, grp2);
d = FOREACH c GENERATE myudfs.Custom(group) AS custom_val, SUM(b.amt1) ... SUM(b.amt31);
store d into './op';