apache-flink - 创建具有通用返回类型的 FlinkSQL UDF

Question

我想定义函数 MAX_BY，它接受 T 类型的值和 Number 类型的排序参数，并根据窗口返回最大元素订购(T 类型)。我试过了

public class MaxBy<T> extends AggregateFunction<T, Tuple2<T, Number>> {

    @Override
    public T getValue(Tuple2<T, Number> tuple) {
        return tuple.f0;
    }

    @Override
    public Tuple2<T, Number> createAccumulator() {
        return Tuple2.of(null, 0L);
    }

    public void accumulate(Tuple2<T, Number> acc, T value, Number order) {
        if (order.doubleValue() > acc.f1.doubleValue()) {
            acc.f0 = value;
            acc.f1 = order;
        }
    }
}

但我无法使用 TableEnvironment.registerFunction 注册此类函数。在 Flink 下面使用 TypeInformation 来匹配 SQL 查询中的类型，并且使用这样的定义它无法确定类型(至少我是这么认为的)。我看到可以提供多个 accumulate 函数，但仍然 - 我认为每个重载方法的返回类型必须相同。

内置聚合函数的工作方式与我想要实现的类似 - MAX 可以采用任意列类型并返回相同类型。这就是为什么我想我也应该能够做到这一点。

Answer 1

不幸的是，Flink 不支持具有灵活返回类型的聚合函数。对于 MAX 函数，内部实现定义了独立于类型的核心逻辑，然后为每个支持的类型创建一个实现(see code)。

在内部，MAX 然后根据类型映射到正确的实现。

如果您将函数定义并注册为用户定义的聚合函数，我认为这是不可能的。