使用 pig 更改元组

Posted 2023-04-13

【中文标题】使用 pig 更改元组

【英文标题】：Change a tuple using pig

【发布时间】：2016-05-02 01:48:24

【问题描述】：

我需要使用 Pig UDF 替换元组的字符。例如，如果我在文件中有一行为“hello world，Hello WORLD，hello\WORLD”，则需要将其转换为“hello_world，hello_world，hello_world”。为此，我尝试了以下 UDF：

package myUDF;
import java.io.IOException;
import org.apache.pig.EvalFunc;
import org.apache.pig.data.Tuple;
import org.apache.pig.data.TupleFactory;
public class ReplaceValues extends EvalFunc<Tuple>

    public Tuple exec(Tuple input) throws IOException 
        if (input == null || input.size() == 0)
            return null;
     try
           String str = (String)input.get(0);
           str=str.replace(" ", "_");
           str=str.replace("/","");
           str=str.replace("\\","");
           TupleFactory tf = TupleFactory.getInstance();
           Tuple t = tf.newTuple();
           t.append(str);
           return t;
    catch(Exception e)
        throw new IOException("Caught exception processing input row ", e);
    
  

但是当通过 pig 脚本调用这个 UDF 时我遇到了问题，请帮助我解决这个问题：

A = load '/user/cloudera/Stage/ActualDataSet.csv' using PigStorage(',') AS (Rank:chararray,NCTNumber:chararray,Title:chararray,Recruitment:chararray);
B = FILTER A by Rank == 'Rank';
C = FOREACH B GENERATE PigUDF.ReplaceValues(B);

错误：Pig 脚本无法解析： 标量投影无效：B：需要从关系中投影列才能用作标量

【问题讨论】：

UDF 代码：public class ReplaceValues extends EvalFunc public Tuple exec(Tuple input) throws IOException if (input == null || input.size() == 0) return null;尝试 String str = (String)input.get(0); str=str.replace(" ", "_"); str=str.replace("/",""); str=str.replace("\\",""); TupleFactory tf = TupleFactory.getInstance();元组 t = tf.newTuple(); t.append(str);返回 t; catch(异常 e) throw(e);

【参考方案1】：

您必须传递您尝试修改的字段而不是关系 B。假设您尝试匹配的字段是 Title，那么您将调用 UDF，如下所示

C = FOREACH B GENERATE B.Rank,B.NCTNumber,PigUDF.ReplaceValues(B.Title),B.Recruitment;

请注意，如果您尝试在整个记录中替换它，那么您的加载语句不正确。您必须将整个记录加载为一个 line:chararray，然后将 line 传递给您的 UDF。

此外，您可以使用REGEX 代替UDF 来匹配和替换您选择的字符串。

【讨论】：

感谢您的意见。我的计划是发送整个元组（在本例中为 Title、NCTNumber、Recuitment ......）并从 UDF 获取清理后的输出。这就是 exec 方法的返回类型为元组的原因，否则我们可以使用 String 来这样做。有没有办法避免为每个元组值多次调用 UDF？另外，我想使用 REGEX/REPLACE 使用 pig 命令，但我找不到正确的语法来替换“\”（反斜杠）。

【参考方案2】：

在您的 Pig 脚本中，您在 UDF 中传递整个 Bag "B"，同时它接受元组作为参数。

改为传递如下所示的 B.Title 之类的字段。

C = FOREACH B GENERATE PigUDF.ReplaceValues(B.Title);

您也可以在同一行中的其他字段上调用此 UDF：

C = FOREACH B GENERATE PigUDF.ReplaceValues(B.Title), PigUDF.ReplaceValues(B.Rank);

【讨论】：

请原谅我的无知，我是这项技术的新手。我使用的是 EvalFunc，这是否意味着我发送的是元组而不是包？我的计划是一次性清理谁排队，而不是为元组的每个数据成员调用 UDF

您不能在EvalFunc() 中发送包。它只需要 Tuple 作为输入，但输出可以是 DataBag。在您的情况下，您正在过滤 Rank 上的记录，因此您可以将剩余的 3 个元组加载为 1 个元组并将其传递给 UDF，否则您将不得不一个一个地传递。详情请参考EvalFunction post

谢谢尼修。那么您是说我需要将 UDF 称为 C = FOREACH B GENERATE PigUDF.ReplaceValues(Title,Rank,....) 以将行作为元组传递，而不是作为 (B)。您也可以提供 pig 命令来替换“\”（反斜杠）吗？