(8) FlinkSQL custom UDF

package com.udf;import org.apache.flink.table.functions.ScalarFunction;/*** Created by lj on 2022-07-25.*/public class TestUDF extends ScalarFunction {public String eval(String value) {return value + "_udf";}}

 public static void main(String[] args) throws Exception {StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();env.setParallelism(1);StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);DataStreamSource streamSource = env.socketTextStream("127.0.0.1", 9999,"\n");SingleOutputStreamOperator waterDS = streamSource.map(new MapFunction() {@Overridepublic WaterSensor map(String s) throws Exception {String[] split = s.split(",");return new WaterSensor(split[0], Long.parseLong(split[1]), Integer.parseInt(split[2]));}});// convert stream to tableTable table = tableEnv.fromDataStream(waterDS,$("id"),$("ts"),$("vc"),$("pt").proctime());tableEnv.createTemporaryView("EventTable", table);/*// 1. Call the custom udf function directly// table.select(call(myFunction.class,$("id"))).execute().print();// 2. Register first before usingtableEnv.createTemporarySystemFunction("MyLength",myFunction.class);//2.1 Using the registered custom function named MyLength// table.select(call("MyLength",$("id"))).execute().print();// 2.2 Use sql to use custom functionstableEnv.sqlQuery("select id, MyLength(id) from "+table).execute().print();* */tableEnv.createTemporarySystemFunction("MyLength",TestUDF.class);Table result = tableEnv.sqlQuery("SELECT " +"id as componentname, " + //window_start, window_end,"COUNT(ts) as componentcount ,SUM(ts) as componentsum, " +"MyLength(cast(COUNT(ts) as string)) as testudf " +"FROM TABLE( " +"TUMBLE( TABLE EventTable , " +"DESCRIPTOR(pt), " +"INTERVAL '10' SECOND)) " +"GROUP BY id , window_start, window_end");tableEnv.toRetractStream(result, Row.class).print("toRetractStream"); ​​//indentation modeenv.execute();}