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 Presley <ti...@foxmail.com> on 2021/02/24 01:37:42 UTC

ValidationException: SQL validation failed.No match found for function signature count_uadf()

请教这位兄dei,这个问题您是怎么解决的?_(:з」∠)_
--------------------------------------------------------------------
问题我自己已经解决。

&gt; 在 2020年12月17日,下午9:00,丁浩浩 <[hidden email]&gt; 写道:
&gt;
&gt; flink版本:1.11.1
&gt; udaf函数代码来自于阿里云官网文档
&gt; ----------------------------
&gt; 以下是代码
&gt; public class TestSql {
&gt; &nbsp; &nbsp;public static void main(String[] args) throws Exception {
&gt; &nbsp; &nbsp; &nbsp; &nbsp;StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
&gt; &nbsp; &nbsp; &nbsp; &nbsp;StreamTableEnvironment tableEnv = FlinkUtils.getTableEnv(env);
&gt; &nbsp; &nbsp; &nbsp; &nbsp;//env.setParallelism(3);
&gt; &nbsp; &nbsp; &nbsp; &nbsp;tableEnv.createTemporarySystemFunction("count_uadf", CountUdaf.class);
&gt;
&gt; &nbsp; &nbsp; &nbsp; &nbsp;Properties configs = CommonUtils.getConfigs();
&gt; &nbsp; &nbsp; &nbsp; &nbsp;//注册clazz源表
&gt; &nbsp; &nbsp; &nbsp; &nbsp;FlinkUtils.registerMysqlTable2FlinkTable(
&gt; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp;tableEnv,configs.getProperty("url"),
&gt; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp;configs.getProperty("user.name"), configs.getProperty("password"),
&gt; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp;“test", "clazz_lesson");
&gt;
&gt; &nbsp; &nbsp; &nbsp; &nbsp;Table table = tableEnv.sqlQuery("select count_uadf(clazz_number),clazz_number from clazz_lesson group by clazz_number");
&gt; &nbsp; &nbsp; &nbsp; &nbsp;//Table table = tableEnv.sqlQuery("select number,collect(extension_value) from clazz_extension group by number ");
&gt; &nbsp; &nbsp; &nbsp; &nbsp;tableEnv.toRetractStream(table, Row.class).print();
&gt; &nbsp; &nbsp; &nbsp; &nbsp;env.execute();
&gt;
&gt;
&gt; &nbsp; &nbsp;}
&gt; }
&gt;
&gt; ------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------
&gt;
&gt; public class CountUdaf extends AggregateFunction<Long, CountUdaf.CountAccum&gt; {
&gt; &nbsp; &nbsp;//定义存放count UDAF状态的accumulator的数据的结构。
&gt; &nbsp; &nbsp;public static class CountAccum {
&gt; &nbsp; &nbsp; &nbsp; &nbsp;public long total;
&gt; &nbsp; &nbsp;}
&gt;
&gt; &nbsp; &nbsp;@Override
&gt; &nbsp; &nbsp;//初始化count UDAF的accumulator。
&gt; &nbsp; &nbsp;public CountAccum createAccumulator() {
&gt; &nbsp; &nbsp; &nbsp; &nbsp;CountAccum acc = new CountAccum();
&gt; &nbsp; &nbsp; &nbsp; &nbsp;acc.total = 0;
&gt; &nbsp; &nbsp; &nbsp; &nbsp;return acc;
&gt; &nbsp; &nbsp;}
&gt; &nbsp; &nbsp;@Override
&gt; &nbsp; &nbsp;//getValue提供了如何通过存放状态的accumulator计算count UDAF的结果的方法。
&gt; &nbsp; &nbsp;public Long getValue(CountAccum accumulator) {
&gt; &nbsp; &nbsp; &nbsp; &nbsp;return accumulator.total;
&gt; &nbsp; &nbsp;}
&gt;
&gt;
&gt; &nbsp; &nbsp;//accumulate提供了如何根据输入的数据更新count UDAF存放状态的accumulator。
&gt; &nbsp; &nbsp;public void accumulate(CountAccum accumulator, Long iValue) {
&gt; &nbsp; &nbsp; &nbsp; &nbsp;accumulator.total++;
&gt; &nbsp; &nbsp;}
&gt; &nbsp; &nbsp;public void merge(CountAccum accumulator, Iterable<CountAccum&gt; its) {
&gt; &nbsp; &nbsp; &nbsp; &nbsp;for&nbsp;(CountAccum other : its) {
&gt; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp;accumulator.total += other.total;
&gt; &nbsp; &nbsp; &nbsp; &nbsp;}
&gt; &nbsp; &nbsp;}
&gt; }
&gt;
&gt; 以下是堆栈信息
&gt;
&gt; ---------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------
&gt; Exception in thread "main" org.apache.flink.table.api.ValidationException: SQL validation failed. From line 1, column 8 to line 1, column 31:&nbsp;No&nbsp;match&nbsp;found&nbsp;for&nbsp;function&nbsp;signature&nbsp;count_uadf(<NUMERIC&gt;)
&gt; at org.apache.flink.table.planner.calcite.FlinkPlannerImpl.org$apache$flink$table$planner$calcite$FlinkPlannerImpl$$validate(FlinkPlannerImpl.scala:146)
&gt; at org.apache.flink.table.planner.calcite.FlinkPlannerImpl.validate(FlinkPlannerImpl.scala:108)
&gt; at org.apache.flink.table.planner.operations.SqlToOperationConverter.convert(SqlToOperationConverter.java:185)
&gt; at org.apache.flink.table.planner.delegation.ParserImpl.parse(ParserImpl.java:78)
&gt; at org.apache.flink.table.api.internal.TableEnvironmentImpl.sqlQuery(TableEnvironmentImpl.java:664)
&gt; at com.gaotu.data.performance.flink.job.sql.TestSql.main(TestSql.java:34)
&gt; Caused by: org.apache.calcite.runtime.CalciteContextException: From line 1, column 8 to line 1, column 31:&nbsp;No&nbsp;match&nbsp;found&nbsp;for&nbsp;function&nbsp;signature&nbsp;count_uadf(<NUMERIC&gt;)
&gt; at sun.reflect.NativeConstructorAccessorImpl.newInstance0(Native Method)
&gt; at sun.reflect.NativeConstructorAccessorImpl.newInstance(NativeConstructorAccessorImpl.java:62)
&gt; at sun.reflect.DelegatingConstructorAccessorImpl.newInstance(DelegatingConstructorAccessorImpl.java:45)
&gt; at java.lang.reflect.Constructor.newInstance(Constructor.java:423)
&gt; at org.apache.calcite.runtime.Resources$ExInstWithCause.ex(Resources.java:457)
&gt; at org.apache.calcite.sql.SqlUtil.newContextException(SqlUtil.java:839)
&gt; at org.apache.calcite.sql.SqlUtil.newContextException(SqlUtil.java:824)
&gt; at org.apache.calcite.sql.validate.SqlValidatorImpl.newValidationError(SqlValidatorImpl.java:5089)
&gt; at org.apache.calcite.sql.validate.SqlValidatorImpl.handleUnresolvedFunction(SqlValidatorImpl.java:1882)
&gt; at org.apache.calcite.sql.SqlFunction.deriveType(SqlFunction.java:305)
&gt; at org.apache.calcite.sql.SqlFunction.deriveType(SqlFunction.java:218)
&gt; at org.apache.calcite.sql.validate.SqlValidatorImpl$DeriveTypeVisitor.visit(SqlValidatorImpl.java:5858)
&gt; at org.apache.calcite.sql.validate.SqlValidatorImpl$DeriveTypeVisitor.visit(SqlValidatorImpl.java:5845)
&gt; at org.apache.calcite.sql.SqlCall.accept(SqlCall.java:139)
&gt; at org.apache.calcite.sql.validate.SqlValidatorImpl.deriveTypeImpl(SqlValidatorImpl.java:1800)
&gt; at org.apache.calcite.sql.validate.SqlValidatorImpl.deriveType(SqlValidatorImpl.java:1785)
&gt; at org.apache.calcite.sql.validate.SqlValidatorImpl.expandSelectItem(SqlValidatorImpl.java:481)
&gt; at org.apache.calcite.sql.validate.SqlValidatorImpl.validateSelectList(SqlValidatorImpl.java:4255)
&gt; at org.apache.calcite.sql.validate.SqlValidatorImpl.validateSelect(SqlValidatorImpl.java:3523)
&gt; at org.apache.calcite.sql.validate.SelectNamespace.validateImpl(SelectNamespace.java:60)
&gt; at org.apache.calcite.sql.validate.AbstractNamespace.validate(AbstractNamespace.java:84)
&gt; at org.apache.calcite.sql.validate.SqlValidatorImpl.validateNamespace(SqlValidatorImpl.java:1110)
&gt; at org.apache.calcite.sql.validate.SqlValidatorImpl.validateQuery(SqlValidatorImpl.java:1084)
&gt; at org.apache.calcite.sql.SqlSelect.validate(SqlSelect.java:232)
&gt; at org.apache.calcite.sql.validate.SqlValidatorImpl.validateScopedExpression(SqlValidatorImpl.java:1059)
&gt; at org.apache.calcite.sql.validate.SqlValidatorImpl.validate(SqlValidatorImpl.java:766)
&gt; at org.apache.flink.table.planner.calcite.FlinkPlannerImpl.org$apache$flink$table$planner$calcite$FlinkPlannerImpl$$validate(FlinkPlannerImpl.scala:141)
&gt; ... 5 more
&gt; Caused by: org.apache.calcite.sql.validate.SqlValidatorException:&nbsp;No&nbsp;match&nbsp;found&nbsp;for&nbsp;function&nbsp;signature&nbsp;count_uadf(<NUMERIC&gt;)
&gt; at sun.reflect.NativeConstructorAccessorImpl.newInstance0(Native Method)
&gt; at sun.reflect.NativeConstructorAccessorImpl.newInstance(NativeConstructorAccessorImpl.java:62)
&gt; at sun.reflect.DelegatingConstructorAccessorImpl.newInstance(DelegatingConstructorAccessorImpl.java:45)
&gt; at java.lang.reflect.Constructor.newInstance(Constructor.java:423)
&gt; at org.apache.calcite.runtime.Resources$ExInstWithCause.ex(Resources.java:457)
&gt; at org.apache.calcite.runtime.Resources$ExInst.ex(Resources.java:550)
&gt; ... 27 more