介绍

标量函数，是指 返回一个值 的函数，实现将0，1，或者多个标量值转化为一个新值。

使用

1. 定义函数类

1. 定义函数类，继承 org.apache.flink.table.functions.ScalarFunction 类

2. 必须定义 public 的 eval() 方法，形参类型、返回值类型自定义（可以定义多个方法）

注意：使用该方法，函数类的 eval() 方法，形参 不能 是基本数据类型，如：int、double 等，否则报错

2. 注册函数

废弃方法：

stEnv.registerFunction("scoreFun",new ScoreFun());

提示：使用该方法，函数类的 eval() 方法，形参可以是基本数据类型，如：int、double 等

推荐方法：

stEnv.createTemporarySystemFunction("函数名",函数类.class);

stEnv.createTemporarySystemFunction("函数名",new 函数类());

3. 调用函数

在 SQL 中调用该函数，方式：

函数名( 列名 )

案例

根据分数显示优良中差

数据文件

在工程的 data 目录下创建 student.csv，内容如下：

李雷,90
韩梅梅,82
lucy,76
lili,63
张三,39

主类

package udf;

import org.apache.flink.api.common.RuntimeExecutionMode;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.EnvironmentSettings;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.TableResult;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.table.functions.ScalarFunction;
import org.apache.flink.types.Row;

public class StdUDF {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setRuntimeMode(RuntimeExecutionMode.AUTOMATIC);//自动模式，根据数据源自行判断

        //构建Table环境
        EnvironmentSettings bsSettings = EnvironmentSettings.newInstance().useBlinkPlanner().inStreamingMode().build();
        StreamTableEnvironment stEnv = StreamTableEnvironment.create(env, bsSettings);

        /*
        创建表
        注意：表名不要出现关键字，如：order等，否则会报错
        要读取 csv文件，需要依赖flink-csv
         */
        TableResult table=stEnv.executeSql("CREATE TABLE t_student (" +
                "  name STRING," +
                "  score int" +
                ") WITH ( " +
                "  'connector' = 'filesystem'," + // 连接器，表示文件系统
                "  'path' = 'file:///D:\\bigdata_std\\flink_sql\\data\\student.csv'," + // 路径指定到文件夹 或 文件都可以
                "  'format' = 'csv'," + // 文件格式
                "  'csv.field-delimiter' = ','" + // csv列之间间隔符号
                ")");

        // 该方法废弃
//        stEnv.registerFunction("scoreFun",new ScoreFun());
        /*
        创建临时系统函数，与 createTemporaryFunction() 相比，可以直接调用
         注意：eval() 方法形参必须是对象类型，不能是 int、long这类基本数据类型
         */
        stEnv.createTemporarySystemFunction("scoreFun",ScoreFun.class);
//        stEnv.createTemporarySystemFunction("scoreFun",new ScoreFun());
        // 根据分数显示优良中差
        Table res=stEnv.sqlQuery("select name,scoreFun(score) from t_student ");
        System.out.println("Schema信息:");
        //打印Schema
        res.printSchema();

        //将计算后的数据，append到ds
        DataStream<Row> dsRes2 = stEnv.toAppendStream(res, Row.class);
        dsRes2.print("---");

        env.execute();
    }

    public static class ScoreFun extends ScalarFunction {

        public String eval(Integer d) {
            String res = "";
            if(d>=90){
                res = "优";
            }else if(d>=80){
                res = "良";
            }else if(d>=70){
                res = "良";
            }else if(d>=60){
                res = "差";
            }else {
                res = "不及格";
            }
            return res;
        }
    }
}

参考：
https://blog.csdn.net/qq_22973811/article/details/124032836
https://blog.csdn.net/qq_21383435/article/details/106698462

原文出处：/show_1IX4NqD8HVOA.html