FlinkSQL split报错No match found for function signature 使用UDF ScalarFunction/TableFunction

约定不等于承诺〃 2021-09-25 05:20 1797阅读 0赞

**1 用TableFunction报错，用ScalarFunction正常**

stEnv.registerFunction("SplitLine",new SplitLine("_"));
            //报错SqlValidatorException: No match found for function signature SplitLine(<CHARACTER>)
            Table result = stEnv.sqlQuery("SELECT name,name2,SplitLine(name2) as name3,qtty," +
                    "case " +
                    "     when coalesce(name,'')='' then qtty*0 " +
                    "     when name = 'a苹果' then qtty*1 " +
                    "     when name = 'b梨' then qtty*2 " +
                    "      when name = 'c西瓜' then qtty*3 " +
                    "      else qtty*4 " +
                    " end as total" +
                    " FROM tableA where qtty > 5");
    
    
    
    public static class SplitLine extends TableFunction<String> {
            private String separator = " ";
            public SplitLine(String separator) {
                this.separator = separator;
            }
            public String eval(String str) {
                return str.split(separator)[0];
            }
        }
    
        public static class SplitLine2 extends ScalarFunction {
            private String separator = " ";
            public SplitLine2(String separator) {
                this.separator = separator;
            }
            public String eval(String str) {
                return str.split(separator)[0];
            }
        }

stEnv.registerFunction("SplitLine2",new SplitLine2("_"));
    
            //报错SqlValidatorException: No match found for function signature SplitLine(<CHARACTER>)
            Table result = stEnv.sqlQuery("SELECT name,name2,SplitLine2(name2) as name3,qtty," +
                    "case " +
                    "     when coalesce(name,'')='' then qtty*0 " +
                    "     when name = 'a苹果' then qtty*1 " +
                    "     when name = 'b梨' then qtty*2 " +
                    "      when name = 'c西瓜' then qtty*3 " +
                    "      else qtty*4 " +
                    " end as total" +
                    " FROM tableA where qtty > 5");
    
    Wed Mar 10 10:24:40 CST 2021,提交元素:Order{name='e火龙果', name2='e_火龙果', name3='null', qtty=6, total=null}
    6> Order{name='e火龙果', name2='e_火龙果', name3='e', qtty=6, total=24}
    Wed Mar 10 10:24:41 CST 2021,提交元素:Order{name='c西瓜', name2='c_西瓜', name3='null', qtty=7, total=null}
    7> Order{name='c西瓜', name2='c_西瓜', name3='c', qtty=7, total=21}
    Wed Mar 10 10:24:42 CST 2021,提交元素:Order{name='b梨', name2='b_梨', name3='null', qtty=1, total=null}
    Wed Mar 10 10:24:43 CST 2021,提交元素:Order{name='a苹果', name2='a_苹果', name3='null', qtty=6, total=null}
    9> Order{name='a苹果', name2='a_苹果', name3='a', qtty=6, total=6}
    Wed Mar 10 10:24:44 CST 2021,提交元素:Order{name='a苹果', name2='a_苹果', name3='null', qtty=4, total=null}
    Wed Mar 10 10:24:45 CST 2021,提交元素:Order{name='a苹果', name2='a_苹果', name3='null', qtty=1, total=null}
    Wed Mar 10 10:24:46 CST 2021,提交元素:Order{name='e火龙果', name2='e_火龙果', name3='null', qtty=8, total=null}
    12> Order{name='e火龙果', name2='e_火龙果', name3='e', qtty=8, total=32}

**2 在sql中用\[0\]报错**

stEnv.registerFunction("SplitLine4", new SplitLine4());
    
            Table result = stEnv.sqlQuery("SELECT name,name2, SplitLine4(name2,'_')[0] as name3,qtty," +
                    "case " +
                    "     when coalesce(name,'')='' then qtty*0 " +
                    "     when name = 'a苹果' then qtty*1 " +
                    "     when name = 'b梨' then qtty*2 " +
                    "      when name = 'c西瓜' then qtty*3 " +
                    "      else qtty*4 " +
                    " end as total" +
                    " FROM tableA where qtty > 5");
    
     public static class SplitLine4 extends ScalarFunction {
            private String separator;
            private Integer index;
    
            public SplitLine4() {
            }
    
            public String[] eval(String str, String separator) {
                if (StringUtils.isNotBlank(str)) {
                    String[] arr = str.split(separator);
                    return arr;
                }
                return new String[10];
            }
        }
    
    
    Wed Mar 10 10:56:05 CST 2021,提交元素:Order{name='c西瓜', name2='c_西瓜', name3='null', qtty=0, total=null}
    Wed Mar 10 10:56:06 CST 2021,提交元素:Order{name='c西瓜', name2='c_西瓜', name3='null', qtty=9, total=null}
    Exception in thread "main" org.apache.flink.runtime.client.JobExecutionException: Job execution failed.
    	at org.apache.flink.runtime.jobmaster.JobResult.toJobExecutionResult(JobResult.java:147)
    	at org.apache.flink.runtime.minicluster.MiniCluster.executeJobBlocking(MiniCluster.java:626)
    	at org.apache.flink.streaming.api.environment.LocalStreamEnvironment.execute(LocalStreamEnvironment.java:117)
    	at org.apache.flink.streaming.api.environment.StreamExecutionEnvironment.execute(StreamExecutionEnvironment.java:1507)
    	at SQL.jd.DataStreamSql2_Select_CaseWhen_Where.main(DataStreamSql2_Select_CaseWhen_Where.java:159)
    Caused by: java.lang.ArrayIndexOutOfBoundsException: -1
    	at DataStreamCalcRule$72.processElement(Unknown Source)
    	at org.apache.flink.table.runtime.CRowProcessRunner.processElement(CRowProcessRunner.scala:66)
    	at org.apache.flink.table.runtime.CRowProcessRunner.processElement(CRowProcessRunner.scala:35)
    	at org.apache.flink.streaming.api.operators.ProcessOperator.processElement(ProcessOperator.java:66)
    	at org.apache.flink.streaming.runtime.tasks.OperatorChain$CopyingChainingOutput.pushToOperator(OperatorChain.java:637)
    	at org.apache.flink.streaming.runtime.tasks.OperatorChain$CopyingChainingOutput.collect(OperatorChain.java:612)
    	at org.apache.flink.streaming.runtime.tasks.OperatorChain$CopyingChainingOutput.collect(OperatorChain.java:592)
    	at org.apache.flink.streaming.api.operators.AbstractStreamOperator$CountingOutput.collect(AbstractStreamOperator.java:727)
    	at org.apache.flink.streaming.api.operators.AbstractStreamOperator$CountingOutput.collect(AbstractStreamOperator.java:705)
    	at org.apache.flink.streaming.api.operators.TimestampedCollector.collect(TimestampedCollector.java:51)
    	at org.apache.flink.table.runtime.CRowWrappingCollector.collect(CRowWrappingCollector.scala:37)
    	at org.apache.flink.table.runtime.CRowWrappingCollector.collect(CRowWrappingCollector.scala:28)
    	at DataStreamSourceConversion$15.processElement(Unknown Source)
    	at org.apache.flink.table.runtime.CRowOutputProcessRunner.processElement(CRowOutputProcessRunner.scala:70)
    	at org.apache.flink.streaming.api.operators.ProcessOperator.processElement(ProcessOperator.java:66)
    	at org.apache.flink.streaming.runtime.io.StreamOneInputProcessor.processElement(StreamOneInputProcessor.java:164)
    	at org.apache.flink.streaming.runtime.io.StreamOneInputProcessor.processInput(StreamOneInputProcessor.java:143)
    	at org.apache.flink.streaming.runtime.tasks.StreamTask.processInput(StreamTask.java:279)
    	at org.apache.flink.streaming.runtime.tasks.StreamTask.run(StreamTask.java:321)
    	at org.apache.flink.streaming.runtime.tasks.StreamTask.runAndHandleCancel(StreamTask.java:286)
    	at org.apache.flink.streaming.runtime.tasks.StreamTask.invoke(StreamTask.java:426)
    	at org.apache.flink.runtime.taskmanager.Task.doRun(Task.java:705)
    	at org.apache.flink.runtime.taskmanager.Task.run(Task.java:530)
    	at java.lang.Thread.run(Thread.java:748)

**3 用无参构造函数注册；下标通过eval()传入 ；这个是正确的！**

stEnv.registerFunction("SplitLine3", new SplitLine3());
    
            Table result = stEnv.sqlQuery("SELECT name,name2, SplitLine3(name2,'_',4) as name3,qtty," + //下标4
                    "case " +
                    "     when coalesce(name,'')='' then qtty*0 " +
                    "     when name = 'a苹果' then qtty*1 " +
                    "     when name = 'b梨' then qtty*2 " +
                    "      when name = 'c西瓜' then qtty*3 " +
                    "      else qtty*4 " +
                    " end as total" +
                    " FROM tableA where qtty > 5");
    
    
    
        public static class SplitLine3 extends ScalarFunction {
            private String separator;
            private Integer index;
    
            public SplitLine3() {
            }
    
            public String eval(String str, String separator, Integer index) {
                if (StringUtils.isNotBlank(str)) {
                    String[] arr = str.split(separator);
                    if (arr.length > index) {
                        return arr[index];
                    }
                }
                return "";
            }
        }
    
    
    输出：
    Wed Mar 10 10:50:49 CST 2021,提交元素:Order{name='d葡萄', name2='d_葡萄', name3='null', qtty=0, total=null}
    Wed Mar 10 10:50:50 CST 2021,提交元素:Order{name='a苹果', name2='a_苹果', name3='null', qtty=8, total=null}
    8> Order{name='a苹果', name2='a_苹果', name3='', qtty=8, total=8}
    Wed Mar 10 10:50:51 CST 2021,提交元素:Order{name='d葡萄', name2='d_葡萄', name3='null', qtty=2, total=null}
    Wed Mar 10 10:50:52 CST 2021,提交元素:Order{name='e火龙果', name2='e_火龙果', name3='null', qtty=9, total=null}
    10> Order{name='e火龙果', name2='e_火龙果', name3='', qtty=9, total=36}
    Wed Mar 10 10:50:53 CST 2021,提交元素:Order{name='a苹果', name2='a_苹果', name3='null', qtty=7, total=null}
    11> Order{name='a苹果', name2='a_苹果', name3='', qtty=7, total=7}
    Wed Mar 10 10:50:54 CST 2021,提交元素:Order{name='', name2='', name3='null', qtty=7, total=null}
    12> Order{name='', name2='', name3='', qtty=7, total=0}
    Wed Mar 10 10:50:55 CST 2021,提交元素:Order{name='e火龙果', name2='e_火龙果', name3='null', qtty=5, total=null}
    Wed Mar 10 10:50:56 CST 2021,提交元素:Order{name='', name2='', name3='null', qtty=5, total=null}

Table result = stEnv.sqlQuery("SELECT name,name2, SplitLine3(name2,'_',0) as name3,qtty," + //下标0
    
    
    Wed Mar 10 10:52:59 CST 2021,提交元素:Order{name='c西瓜', name2='c_西瓜', name3='null', qtty=2, total=null}
    Wed Mar 10 10:53:00 CST 2021,提交元素:Order{name='a苹果', name2='a_苹果', name3='null', qtty=1, total=null}
    Wed Mar 10 10:53:01 CST 2021,提交元素:Order{name='d葡萄', name2='d_葡萄', name3='null', qtty=7, total=null}
    5> Order{name='d葡萄', name2='d_葡萄', name3='d', qtty=7, total=28}
    Wed Mar 10 10:53:02 CST 2021,提交元素:Order{name='b梨', name2='b_梨', name3='null', qtty=4, total=null}
    Wed Mar 10 10:53:03 CST 2021,提交元素:Order{name='d葡萄', name2='d_葡萄', name3='null', qtty=5, total=null}
    Wed Mar 10 10:53:04 CST 2021,提交元素:Order{name='d葡萄', name2='d_葡萄', name3='null', qtty=3, total=null}
    Wed Mar 10 10:53:05 CST 2021,提交元素:Order{name='a苹果', name2='a_苹果', name3='null', qtty=4, total=null}
    Wed Mar 10 10:53:06 CST 2021,提交元素:Order{name='a苹果', name2='a_苹果', name3='null', qtty=0, total=null}
    Wed Mar 10 10:53:07 CST 2021,提交元素:Order{name='a苹果', name2='a_苹果', name3='null', qtty=2, total=null}
    Wed Mar 10 10:53:08 CST 2021,提交元素:Order{name='c西瓜', name2='c_西瓜', name3='null', qtty=2, total=null}
    Wed Mar 10 10:53:09 CST 2021,提交元素:Order{name='', name2='', name3='null', qtty=6, total=null}
    1> Order{name='', name2='', name3='', qtty=6, total=0}
    Wed Mar 10 10:53:10 CST 2021,提交元素:Order{name='c西瓜', name2='c_西瓜', name3='null', qtty=9, total=null}
    2> Order{name='c西瓜', name2='c_西瓜', name3='c', qtty=9, total=27}
    Wed Mar 10 10:53:11 CST 2021,提交元素:Order{name='a苹果', name2='a_苹果', name3='null', qtty=9, total=null}
    3> Order{name='a苹果', name2='a_苹果', name3='a', qtty=9, total=9}

**4 自定义的函数需要先注册才能用**

//SqlValidatorException: No match found for function signature splitLine(<CHARACTER>)
    Table result = stEnv.sqlQuery("SELECT name,splitLine(name2) as name2,qtty," +
            "case " +
            "     when coalesce(name,'')='' then qtty*0 " +
            "     when name = 'a苹果' then qtty*1 " +
            "     when name = 'b梨' then qtty*2 " +
            "      when name = 'c西瓜' then qtty*3 " +
            "      else qtty*4 " +
            " end as total" +
            " FROM tableA where qtty > 5");
    
    
    public String splitLine(String s){
        return s.split("_")[0];
    }

Flink 1.9.2 ,JDK1.8

![watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBA5LqM5Y2B5YWt55S755Sf55qE5Y2a5a6i_size_20_color_FFFFFF_t_70_g_se_x_16][]

[watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBA5LqM5Y2B5YWt55S755Sf55qE5Y2a5a6i_size_20_color_FFFFFF_t_70_g_se_x_16]: /images/20210923/003f016dd6554f11addb0c8b02665297.png