在 Apache Flink 中注册聚合 UDF

Question

我正在尝试按照这里的步骤创建一个基本的 Flink 聚合 UDF。我已经添加了依赖项((并实现了

public class MyAggregate extends AggregateFunction<Long, TestAgg> {..}

我已经实现了强制性方法以及其他一些方法：accumulate, merge, etc .所有这些都没有错误。现在根据文档，我应该能够将其注册为

    StreamExecutionEnvironment sEnv = StreamExecutionEnvironment.getExecutionEnvironment();
    StreamTableEnvironment sTableEnv = StreamTableEnvironment.getTableEnvironment(sEnv);
    sTableEnv.registerFunction("MyMin", new MyAggregate());

但是，registerFucntion似乎只想要一个ScalarFunction作为输入。我收到不兼容的类型错误：The method registerFunction(String, ScalarFunction) in the type TableEnvironment is not applicable for the arguments (String, MyAggregate)

任何帮助都会很棒。

Answer 1

您需要

导入所选语言的StreamTableEnvironment，这在您的情况下org.apache.flink.table.api.java.StreamTableEnvironment。

org.apache.flink.table.api.StreamTableEnvironment是StreamTableEnvironment的Java和Scala变体的常见抽象类。我们注意到 API 的这一部分让用户感到困惑，我们将来会对其进行改进。