hadoop - 在 EvalFunc pig UDF 中抛出异常是跳过那一行，还是完全停止？

Question

我有一个用 Java 编写的用户定义函数 (UDF)，用于解析日志文件中的行并将信息返回给 pig，因此它可以完成所有处理。

看起来像这样:

public abstract class Foo extends EvalFunc<Tuple> {
    public Foo() {
        super();
    }

    public Tuple exec(Tuple input) throws IOException {
        try {
            // do stuff with input
        } catch (Exception e) {
            throw WrappedIOException.wrap("Error with line", e);
        }
    }
}

我的问题是:如果它抛出 IOException，它会完全停止，还是会为其余未抛出异常的行返回结果？

例子:我在 pig 身上运行这个

REGISTER myjar.jar
DEFINE Extractor com.namespace.Extractor();

logs = LOAD '$IN' USING TextLoader AS (line: chararray);
events = FOREACH logs GENERATE FLATTEN(Extractor(line));

有了这个输入:

1.5 7 "Valid Line"
1.3 gghyhtt Inv"alid line"" I throw an exceptioN!!
1.8 10 "Valid Line 2"

它会处理这两行并且“logs”会有 2 个元组，还是会在火中烧毁？

Answer 1

如果 UDF 抛出异常，任务将失败并重试。

它将再次失败三次(默认尝试 4 次)，整个作业将失败。

如果您想记录错误并且不想让作业停止，您可以返回一个空值:

public Tuple exec(Tuple input) throws IOException {
    try {
        // do stuff with input
    } catch (Exception e) {
        System.err.println("Error with ...");
        return null;
    }
}

然后在 Pig 中过滤它们:

events_all = FOREACH logs GENERATE Extractor(line) AS line;
events_valid = FILTER events_all by line IS NOT null;
events = FOREACH events_valid GENERATE FLATTEN(line);

在您的示例中，输出将只有两行有效(但要小心这种行为，因为错误仅出现在日志中，不会使您的工作失败!)。

回复评论#1:

实际上，整个结果元组将为空(因此内部没有字段)。

例如，如果您的模式有 3 个字段:

 events_all = FOREACH logs
              GENERATE Extractor(line) AS line:tuple(a:int,b:int,c:int);

有些行不正确我们会得到:

 ()
 ((1,2,3))
 ((1,2,3))
 ()
 ((1,2,3))

如果您不过滤空行并尝试访问一个字段，您将得到一个 java.lang.NullPointerException:

events = FOREACH events_all GENERATE line.a;