Hi,
我debug到了代码里，似乎是个bug。如果使用OldPlanner的话，emitValue和emitUpdateWithRetract仅需定义一个就可以了，并且emitUpdateWithRetract的优先级大于emitValue。但是在Blink
Planner里，只看有没有定义emitValue。你可以去Flink issue上提一下这个bug
在 2022-05-23 18:24:17，"sjf0115" <si_ji_f...@163.com> 写道：
>Flink 版本：1.13.5
>
>
>
>
>函数完整代码如下：
>```
>public class Top2RetractTableAggregateFunction extends
>TableAggregateFunction<Tuple2<Long, Integer>,
>Top2RetractTableAggregateFunction.Top2RetractAccumulator> {
> private static final Logger LOG =
> LoggerFactory.getLogger(Top2RetractTableAggregateFunction.class);
> // Top2 聚合中间结果数据结构
> public static class Top2RetractAccumulator {
> public long beforeFirst = 0;
> public long beforeSecond = 0;
> public long afterFirst = 0;
> public long afterSecond = 0;
> }
>
>
> // 创建 Top2Accumulator 累加器并做初始化
> @Override
> public Top2RetractAccumulator createAccumulator() {
> LOG.info("[INFO] createAccumulator ...........................");
> Top2RetractAccumulator acc = new Top2RetractAccumulator();
> acc.beforeFirst = Integer.MIN_VALUE;
> acc.beforeSecond = Integer.MIN_VALUE;
> acc.afterFirst = Integer.MIN_VALUE;
> acc.afterSecond = Integer.MIN_VALUE;
> return acc;
> }
>
>
> // 接收输入元素并累加到 Accumulator 数据结构
> public void accumulate(Top2RetractAccumulator acc, Long value) {
> LOG.info("[INFO] accumulate ...........................");
> if (value > acc.afterFirst) {
> acc.afterSecond = acc.afterFirst;
> acc.afterFirst = value;
> } else if (value > acc.afterSecond) {
> acc.afterSecond = value;
> }
> }
>
>
> // 带撤回的输出
> public void emitUpdateWithRetract(Top2RetractAccumulator acc,
> RetractableCollector<Tuple2<Long, Integer>> out) {
> LOG.info("[INFO] emitUpdateWithRetract ...........................");
> if (!Objects.equals(acc.afterFirst, acc.beforeFirst)) {
> // 撤回旧记录
> if (acc.beforeFirst != Integer.MIN_VALUE) {
> out.retract(Tuple2.of(acc.beforeFirst, 1));
> }
> // 输出新记录
> out.collect(Tuple2.of(acc.afterFirst, 1));
> acc.beforeFirst = acc.afterFirst;
> }
> if (!Objects.equals(acc.afterSecond, acc.beforeSecond)) {
> // 撤回旧记录
> if (acc.beforeSecond != Integer.MIN_VALUE) {
> out.retract(Tuple2.of(acc.beforeSecond, 2));
> }
> // 输出新记录
> out.collect(Tuple2.of(acc.afterSecond, 2));
> acc.beforeSecond = acc.afterSecond;
> }
> }
>}
>```
>调用完整代码如下：
>```
>// 执行环境
>StreamExecutionEnvironment env =
>StreamExecutionEnvironment.getExecutionEnvironment();
>env.setParallelism(1);
>EnvironmentSettings settings = EnvironmentSettings
> .newInstance()
> .useOldPlanner() // Blink Planner 异常 Old Planner 可以
> .inStreamingMode()
> .build();
>StreamTableEnvironment tEnv = StreamTableEnvironment.create(env, settings);
>
>
>DataStream<Row> sourceStream = env.fromElements(
> Row.of("李雷", "语文", 78),
> Row.of("韩梅梅", "语文", 50),
> Row.of("李雷", "语文", 99),
> Row.of("韩梅梅", "语文", 80),
> Row.of("李雷", "英语", 90),
> Row.of("韩梅梅", "英语", 40),
> Row.of("李雷", "英语", 98),
> Row.of("韩梅梅", "英语", 88)
>);
>
>
>// 注册虚拟表
>tEnv.createTemporaryView("stu_score", sourceStream, $("name"), $("course"),
>$("score"));
>// 注册临时i系统函数
>tEnv.createTemporarySystemFunction("Top2", new
>Top2RetractTableAggregateFunction());
>// 调用函数
>tEnv.from("stu_score")
> .groupBy($("course"))
> .flatAggregate(call("Top2", $("score")).as("score", "rank"))
> .select($("course"), $("score"), $("rank"))
> .execute()
> .print();
>```
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>在 2022-05-23 18:21:42，"sjf0115" <si_ji_f...@163.com> 写道：
>>函数代码如下：<br/>```<br/>public class Top2RetractTableAggregateFunction extends
>>TableAggregateFunction<Tuple2<Long, Integer>,
>>Top2RetractTableAggregateFunction.Top2RetractAccumulator> {<br/>
>>private static final Logger LOG =
>>LoggerFactory.getLogger(Top2RetractTableAggregateFunction.class);<br/> //
>>Top2 聚合中间结果数据结构<br/> public static class Top2RetractAccumulator {<br/>
>> public long beforeFirst = 0;<br/> public long beforeSecond =
>>0;<br/> public long afterFirst = 0;<br/> public long
>>afterSecond = 0;<br/> }<br/><br/> // 创建 Top2Accumulator 累加器并做初始化<br/>
>> @Override<br/> public Top2RetractAccumulator createAccumulator() {<br/>
>> LOG.info("[INFO] createAccumulator ...........................");<br/>
>> Top2RetractAccumulator acc = new Top2RetractAccumulator();<br/>
>>acc.beforeFirst = Integer.MIN_VALUE;<br/> acc.beforeSecond =
>>Integer.MIN_VALUE;<br/> acc.afterFirst = Integer.MIN_VALUE;<br/>
>> acc.afterSecond = Integer.MIN_VALUE;<br/> return acc;<br/>
>>}<br/><br/> // 接收输入元素并累加到 Accumulator 数据结构<br/> public void
>>accumulate(Top2RetractAccumulator acc, Long value) {<br/>
>>LOG.info("[INFO] accumulate ...........................");<br/> if
>>(value > acc.afterFirst) {<br/> acc.afterSecond =
>>acc.afterFirst;<br/> acc.afterFirst = value;<br/> } else if
>>(value > acc.afterSecond) {<br/> acc.afterSecond = value;<br/>
>> }<br/> }<br/><br/> // 带撤回的输出<br/> public void
>>emitUpdateWithRetract(Top2RetractAccumulator acc,
>>RetractableCollector<Tuple2<Long, Integer>> out) {<br/>
>>LOG.info("[INFO] emitUpdateWithRetract ...........................");<br/>
>> if (!Objects.equals(acc.afterFirst, acc.beforeFirst)) {<br/>
>>// 撤回旧记录<br/> if (acc.beforeFirst != Integer.MIN_VALUE) {<br/>
>> out.retract(Tuple2.of(acc.beforeFirst, 1));<br/> }<br/>
>> // 输出新记录<br/> out.collect(Tuple2.of(acc.afterFirst,
>>1));<br/> acc.beforeFirst = acc.afterFirst;<br/> }<br/>
>> if (!Objects.equals(acc.afterSecond, acc.beforeSecond)) {<br/>
>>// 撤回旧记录<br/> if (acc.beforeSecond != Integer.MIN_VALUE) {<br/>
>> out.retract(Tuple2.of(acc.beforeSecond, 2));<br/>
>>}<br/> // 输出新记录<br/>
>>out.collect(Tuple2.of(acc.afterSecond, 2));<br/> acc.beforeSecond
>>= acc.afterSecond;<br/> }<br/>
>>}<br/>}<br/>```<br/>完整调用代码：<br/>```<br/>//
>>执行环境<br/>StreamExecutionEnvironment env =
>>StreamExecutionEnvironment.getExecutionEnvironment();<br/>env.setParallelism(1);<br/>EnvironmentSettings
>> settings = EnvironmentSettings<br/> .newInstance()<br/>
>>.useOldPlanner() // Blink Planner 异常 Old Planner 可以<br/>
>>.inStreamingMode()<br/> .build();<br/>StreamTableEnvironment tEnv =
>>StreamTableEnvironment.create(env, settings);<br/><br/>DataStream<Row>
>>sourceStream = env.fromElements(<br/> Row.of("李雷", "语文", 78),<br/>
>> Row.of("韩梅梅", "语文", 50),<br/> Row.of("李雷", "语文", 99),<br/>
>>Row.of("韩梅梅", "语文", 80),<br/> Row.of("李雷", "英语", 90),<br/>
>>Row.of("韩梅梅", "英语", 40),<br/> Row.of("李雷", "英语", 98),<br/>
>>Row.of("韩梅梅", "英语", 88)<br/>);<br/><br/>//
>>注册虚拟表<br/>tEnv.createTemporaryView("stu_score", sourceStream, $("name"),
>>$("course"), $("score"));<br/>//
>>注册临时i系统函数<br/>tEnv.createTemporarySystemFunction("Top2", new
>>Top2RetractTableAggregateFunction());<br/>//
>>调用函数<br/>tEnv.from("stu_score")<br/> .groupBy($("course"))<br/>
>>.flatAggregate(call("Top2", $("score")).as("score", "rank"))<br/>
>>.select($("course"), $("score"), $("rank"))<br/> .execute()<br/>
>> .print();<br/>```<br/>Flink 版本：1.13.5
>>在 2022-05-23 09:55:40，"Xuyang" <xyzhong...@163.com> 写道：
>>>Hi， 可以将你的taf具体代码发出来吗？还有你的版本，也贴一下。
>>>
>>>
>>>
>>>
>>>--
>>>
>>> Best！
>>> Xuyang
>>>
>>>
>>>
>>>
>>>
>>>在 2022-05-22 22:35:46，"赢峰" <si_ji_f...@163.com> 写道：
>>>>
>>>>
>>>>在自定义表聚合函数 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);
>>>>```
>>>>这是什么地方使用有问题？
>>>>
>>>>
>>>>
>>>>
>>>>
