Apache Pig：组运算符和变量模式

Posted 2023-04-18

【中文标题】Apache Pig：组运算符和变量模式

【英文标题】：Apache Pig: Group operator and variable schema

【发布时间】：2014-11-13 23:48:00

【问题描述】：

我有以下问题： 我的 sample.txt 的片段。第一列是 id，但是每行可以有可变数量的列。

id1 100 200 300 400 500

id2 10 20 30

id1 800 900 600

id3 10 20 30 40 50 60 70 80 90 100

id1 1 2 3 4 5 6 7 8 9

id2 40 50 60 70 80 90

id3 200

sample = LOAD 'sample.txt' [我应该如何在这里指定架构]

sample_grpd = GROUP 样本 $0;

sample_result = FOREACH sample_grpd 生成组，FLATTEN(TOBAG([what should go here]))

按 id 分组，结果为：

id1 100 200 300 400 500 800 900 600 1 2 3 4 5 6 7 8 9

id2 10 20 30 40 50 60 70 80 90

id3 10 20 30 40 50 60 70 80 90 100 200

对此的任何帮助，将不胜感激！

【问题讨论】：

使用原生 pig 实现您的要求非常困难。你只需要去UDF。请根据您的输入尝试以下解决方案，并告诉我它是如何工作的。对于上述问题，将代码更改为“sample = LOAD 'sample.txt' USING PigStorage(' ')”和 FLATTEN(TOBAG(*))。

【参考方案1】：

这是一个棘手的问题，最后我以某种方式使用 UDF 解决了它。

input.txt

id1 100 200 300 400 500
id2 10 20 30
id1 800 900 600
id3 10 20 30 40 50 60 70 80 90 100
id1 1 2 3 4 5 6 7 8 9
id2 40 50 60 70 80 90
id3 200

PigScript：

REGISTER removeduplicate.jar;
A = LOAD 'input.txt' USING PigStorage(' ');
B = GROUP A by $0;
C = FOREACH B GENERATE $0 AS myid:chararray,$1 AS (B:T:(f1:chararray));
D = FOREACH C GENERATE myid,BagToString(B) AS concatString;
E = FOREACH D GENERATE myid,mypackage.REMOVEDUPLICATE(concatString) AS finalString;
F = FOREACH E GENERATE myid,FLATTEN(STRSPLIT(finalString,'_',40)) AS result;
STORE F INTO 'output' USING PigStorage(' ');

输出：

id1 100 200 300 400 500 800 900 600 1 2 3 4 5 6 7 8 9
id2 10 20 30 40 50 60 70 80 90
id3 10 20 30 40 50 60 70 80 90 100 200

UDF CODE下面的java类文件被编译生成为removeduplicate.jarREMOVEDUPLICATE.java

  package mypackage;
    import java.io.IOException;
    import org.apache.commons.lang.StringUtils;
    import org.apache.pig.EvalFunc;
    import org.apache.pig.data.Tuple;

    public class REMOVEDUPLICATE extends EvalFunc<String> 
    @Override
    public String exec(Tuple arg0) throws IOException 
           try
            
                String input = ((String) arg0.get(0));
                    String duplicateString = input.split("\\_")[0]+"_";
                    System.out.println(duplicateString);
                    return(input.replace(duplicateString, ""));
            
            catch(Exception e)
            
                throw new IOException("Caught exception while processing the input row ", e);
            
        
    

【讨论】：

BagToString 是否每个都添加 _？

是的，你是对的，BagToString 函数会将 '_' 作为默认分隔符附加到每个字段。您也可以更改分隔符，例如 BagToString(mystring,delimiter);