创建具有通用返回类型的 FlinkSQL UDF [英] Create FlinkSQL UDF with generic return type

问题描述

我想定义函数 MAX_BY ，它采用 T 类型的值和 Number 类型的排序参数，并根据到排序(T 类型).我试过了

I would like to define function MAX_BY that takes value of type T and ordering parameter of type Number and returns max element from window according to ordering (of type T). I've tried

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 函数，但仍然 - 我认为每个重载方法的返回类型必须相同.

but I cannot register such function using TableEnvironment.registerFunction. Underneath Flink uses TypeInformation to match types within SQL query and with such definition it cannot determine types (at least that's what I suppose). I saw that it is possible to provide several accumulate functions but still - I think return type must be same for each overloaded method.

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

Built-in aggregation functions work similarly to what I want to achieve - MAX can take arbitrary column type and return the same type. That's why I suppose I should be able to do it as well.

推荐答案

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

Unfortunately, Flink doesn't support aggregation functions with flexible return types. For the MAX function, the internal implementation defines the core logic independent of the the type and then creates an implementation for every supported type (see code).

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

Internally, MAX is then mapped to the right implementation, depending on the type.

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

I don't think that's possible if you define and register a function as user-defined aggregation function.

这篇关于创建具有通用返回类型的 FlinkSQL UDF的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持IT屋！