You are viewing a plain text version of this content. The canonical link for it is here.
Posted to user-zh@flink.apache.org by 赢峰 <si...@163.com> on 2022/05/22 14:33:05 UTC
22/05/2022 22:28:33 自动保存草稿
在自定义表聚合函数 TableAggregateFunction 时使用的是 emitUpdateWithRetract 输出数据。在调用的时候参考文档的使用方式:
```
tEnv.from("stu_score")
.groupBy($("course"))
.flatAggregate(call(Top2RetractTableAggregateFunction.class, $("score")))
.select($("course"), $("f0"), $("f1"))
```
使用默认 blink Planner,会抛出如下异常:
```
Exception in thread "main" org.apache.flink.table.api.ValidationException: Could not find an implementation method 'emitValue' in class 'com.flink.example.table.function.custom.Top2RetractTableAggregateFunction' for function 'Top2' that matches the following signature:
void emitValue(com.flink.example.table.function.custom.Top2RetractTableAggregateFunction.Top2RetractAccumulator, org.apache.flink.util.Collector)
```
但是使用 Old Planner,则会正常输出:
```
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
env.setParallelism(1);
EnvironmentSettings settings = EnvironmentSettings
.newInstance()
.useOldPlanner()
.inStreamingMode()
.build();
StreamTableEnvironment tEnv = StreamTableEnvironment.create(env, settings);
```
这是什么地方使用有问题?