如何使用最后一个输入行转发配置单元UDTF输出

Question

配置单元UDTF单独处理每行，并转发每行的输出。我需要对所有的函数输出做一些操作。有没有办法在处理完所有行之后一次转发所有函数的输出？

提前感谢！

Answer 1

是的，这种方式是存在的。

要使用

1. 的输出进行一些更改，您必须首先存储它。因此，在UDF类中创建字段来收集其中的输出：private List<SomeClass> output = new ArrayList<>();
2. Then在process()方法中您应该使用output.add()而不是forward()。因此，在process()中，output将被填充。

，

1. ，最后是close()方法。它用于在所有处理之后进行清理，通常不需要在其中编写代码。然而，在您的例子中，close()是一个很好的处理output收集并转发它的可能性。

close()的大致代码如下：

@Override
public void close() throws HiveException {
    for(String record: output){
        forwardObjArray[0] = record.substring(3);
        forward(forwardObjArray);
    }
}

forwardObjArray是UDF的另一个私有字段。优化资源使用(避免在每次forward()调用前创建1个元素的数组)。

但是，只有在小数据集上运行UDTF时，这种方法才适用。否则，output集合将非常大。

如果您要在大型输入数据上运行UDTF，我建议使用两个UDTF。由于当前的配置单元版本不允许使用嵌套UDTF，您可以使用子查询：

SELECT func2(processed)
FROM (
    SELECT func1(fiels) AS processed
    FROM table
) AS interm_table;

在func2()中，您将实现用于操作func1()输出的逻辑。