4.2.11 Flink-流处理框架-Table API 与 SQL-函数（Functions）之用户自定义函数 UDF

比眉伴天荒 2022-11-09 14:12 145阅读 0赞

**目录**

1.写在前面

2.标量函数（Scalar Functions）：一对一

3.表函数（Table Functions）：一对多

4.聚合函数（Aggregate Functions）

5.表聚合函数（Table Aggregate Functions）

--------------------

# 1.写在前面 #

用户定义函数（User-defined Functions，UDF）是一个重要的特性，因为它们显著地扩展了查询（Query）的表达能力。一些系统内置函数无法解决的需求，我们可以用 UDF 来自定义实现。

在大多数情况下，用户定义的函数必须先注册，然后才能在查询中使用。不需要专门为 Scala 的 Table API 注册函数。 函数通过调用 **registerFunction（）**方法在 TableEnvironment 中注册。当用户定义的函数被注册时，它被插入到 TableEnvironment 的函数目录中，这样 Table API 或 SQL 解析器就可以识别并正确地解释它。

# 2.标量函数（Scalar Functions）：一对一 #

用户定义的标量函数，可以将 0、1 或多个标量值，映射到新的标量值。 为了定义标量函数，必须在 org.apache.flink.table.functions 中扩展基类 **ScalarFunction**， 并实现（一个或多个）求值（evaluation，eval）方法。标量函数的行为由求值方法决定， **求值方法必须公开声明并命名为 eval（直接 def 声明，没有 override）**。求值方法的参数类型和返回类型，确定了标量函数的参数和返回类型。

在下面的代码中，我们定义自己的 HashCode 函数，在 TableEnvironment 中注册它，并在查询中调用它。

// 自定义一个标量函数
    public static class HashCode extends ScalarFunction {
        private int factor = 13;
        public HashCode(int factor) {    
            this.factor = factor;
        }
     
        public int eval(String s) {
            return s.hashCode() * factor;
        }
    }

主函数中调用，计算 sensor id 的哈希值（前面部分照抄，流环境、表环境、读取 source、 建表）。

import com.atguigu.apitest.beans.SensorReading;
    import org.apache.flink.streaming.api.datastream.DataStream;
    import org.apache.flink.streaming.api.datastream.DataStreamSource;
    import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
    import org.apache.flink.table.api.Table;
    import org.apache.flink.table.api.java.StreamTableEnvironment;
    import org.apache.flink.table.functions.ScalarFunction;
    import org.apache.flink.types.Row;
    
    public class UdfTest1_ScalarFunction {
        public static void main(String[] args) throws Exception{
            StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
            env.setParallelism(1);
    
            StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);
    
            // 1. 读取数据
            DataStreamSource<String> inputStream = env.readTextFile("D:\\Projects\\BigData\\FlinkTutorial\\src\\main\\resources\\sensor.txt");
    
            // 2. 转换成POJO
            DataStream<SensorReading> dataStream = inputStream.map(line -> {
                String[] fields = line.split(",");
                return new SensorReading(fields[0], new Long(fields[1]), new Double(fields[2]));
            });
    
            // 3. 将流转换成表
            Table sensorTable = tableEnv.fromDataStream(dataStream, "id, timestamp as ts, temperature as temp");
    
            // 4. 自定义标量函数，实现求id的hash值
            // 4.1 table API
            HashCode hashCode = new HashCode(23);
            // 需要在环境中注册UDF
            tableEnv.registerFunction("hashCode", hashCode);
            Table resultTable = sensorTable.select("id, ts, hashCode(id)");
    
            // 4.2 SQL
            tableEnv.createTemporaryView("sensor", sensorTable);
            Table resultSqlTable = tableEnv.sqlQuery("select id, ts, hashCode(id) from sensor");
    
            // 打印输出
            tableEnv.toAppendStream(resultTable, Row.class).print("result");
            tableEnv.toAppendStream(resultSqlTable, Row.class).print("sql");
    
            env.execute();
        }
    
        // 实现自定义的ScalarFunction
        public static class HashCode extends ScalarFunction{
            private int factor = 13;
    
            public HashCode(int factor) {
                this.factor = factor;
            }
    
            public int eval(String str){
                return str.hashCode() * factor;
            }
        }
    }

# 3.表函数（Table Functions）：一对多 #

与用户定义的标量函数类似，用户定义的表函数，可以将 0、1 或多个标量值作为输入 参数；与标量函数不同的是，它可以返回任意数量的行作为输出，而不是单个值。 为了定义一个表函数，必须扩展 org.apache.flink.table.functions 中的基类 **TableFunction** 并实现（一个或多个）求值方法。表函数的行为由其求值方法决定，**求值方法必须是 public 的，并命名为 eval**。求值方法的参数类型，决定表函数的所有有效参数。 返回表的类型由 TableFunction 的泛型类型确定。**求值方法使用 protected collect（T）方法发出输出行**。

在 Table API 中，Table 函数需要与.joinLateral 或.leftOuterJoinLateral 一起使用。 joinLateral 算子，会将外部表中的每一行，与表函数（TableFunction，算子的参数是它的表达式）计算得到的所有行连接起来。 而 leftOuterJoinLateral 算子，则是左外连接，它同样会将外部表中的每一行与表函数计算生成的所有行连接起来；并且，对于表函数返回的是空表的外部行，也要保留下来。 在 SQL 中，则需要使用 Lateral Table（），或者带有 ON TRUE 条件的左连接。

下面的代码中，我们将定义一个表函数，在表环境中注册它，并在查询中调用它。

import com.atguigu.apitest.beans.SensorReading;
    import org.apache.flink.api.java.tuple.Tuple2;
    import org.apache.flink.streaming.api.datastream.DataStream;
    import org.apache.flink.streaming.api.datastream.DataStreamSource;
    import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
    import org.apache.flink.table.api.Table;
    import org.apache.flink.table.api.java.StreamTableEnvironment;
    import org.apache.flink.table.functions.TableFunction;
    import org.apache.flink.types.Row;
    
    public class UdfTest2_TableFunction {
        public static void main(String[] args) throws Exception{
            StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
            env.setParallelism(1);
    
            StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);
    
            // 1. 读取数据
            DataStreamSource<String> inputStream = env.readTextFile("D:\\Projects\\BigData\\FlinkTutorial\\src\\main\\resources\\sensor.txt");
    
            // 2. 转换成POJO
            DataStream<SensorReading> dataStream = inputStream.map(line -> {
                String[] fields = line.split(",");
                return new SensorReading(fields[0], new Long(fields[1]), new Double(fields[2]));
            });
    
            // 3. 将流转换成表
            Table sensorTable = tableEnv.fromDataStream(dataStream, "id, timestamp as ts, temperature as temp");
    
            // 4. 自定义表函数，实现将id拆分，并输出（word, length）
            // 4.1 table API
            Split split = new Split("_");
    
            // 需要在环境中注册UDF
            tableEnv.registerFunction("split", split);
            Table resultTable = sensorTable
                    .joinLateral("split(id) as (word, length)")
                    .select("id, ts, word, length");
    
            // 4.2 SQL
            tableEnv.createTemporaryView("sensor", sensorTable);
            Table resultSqlTable = tableEnv.sqlQuery("select id, ts, word, length " +
                    " from sensor, lateral table(split(id)) as splitid(word, length)");
    
            // 打印输出
            tableEnv.toAppendStream(resultTable, Row.class).print("result");
            tableEnv.toAppendStream(resultSqlTable, Row.class).print("sql");
    
            env.execute();
        }
    
        // 实现自定义TableFunction
        public static class Split extends TableFunction<Tuple2<String, Integer>>{
            // 定义属性，分隔符
            private String separator = ",";
    
            public Split(String separator) {
                this.separator = separator;
            }
    
            // 必须实现一个eval方法，没有返回值
            public void eval( String str ){
                for( String s: str.split(separator) ){
                    collect(new Tuple2<>(s, s.length()));
                }
            }
        }
    }

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1N1eWViaXViaXU_size_16_color_FFFFFF_t_70][]

# 4.聚合函数（Aggregate Functions） #

用户自定义聚合函数（User-Defined Aggregate Functions，UDAGGs）可以把一个表中的数据，聚合成一个标量值。用户定义的聚合函数，是通过**继承 AggregateFunction 抽象类**实 现的。

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1N1eWViaXViaXU_size_16_color_FFFFFF_t_70 1][]

上图中显示了一个聚合的例子。 假设现在有一张表，包含了各种饮料的数据。该表由三列（id、name 和 price）、五行组成数据。现在我们需要找到表中所有饮料的最高价格，即执行 max（）聚合，结果将是一个数值。

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1N1eWViaXViaXU_size_16_color_FFFFFF_t_70 2][]

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1N1eWViaXViaXU_size_16_color_FFFFFF_t_70 3][]

接下来我们写一个自定义  AggregateFunction ，计算一下每个  sensor 的平均温度值

import com.atguigu.apitest.beans.SensorReading;
    import org.apache.flink.api.java.tuple.Tuple2;
    import org.apache.flink.streaming.api.datastream.DataStream;
    import org.apache.flink.streaming.api.datastream.DataStreamSource;
    import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
    import org.apache.flink.table.api.Table;
    import org.apache.flink.table.api.java.StreamTableEnvironment;
    import org.apache.flink.table.functions.AggregateFunction;
    import org.apache.flink.types.Row;
    
    public class UdfTest3_AggregateFunction {
        public static void main(String[] args) throws Exception{
            StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
            env.setParallelism(1);
    
            StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);
    
            // 1. 读取数据
            DataStreamSource<String> inputStream = env.readTextFile("D:\\Projects\\BigData\\FlinkTutorial\\src\\main\\resources\\sensor.txt");
    
            // 2. 转换成POJO
            DataStream<SensorReading> dataStream = inputStream.map(line -> {
                String[] fields = line.split(",");
                return new SensorReading(fields[0], new Long(fields[1]), new Double(fields[2]));
            });
    
            // 3. 将流转换成表
            Table sensorTable = tableEnv.fromDataStream(dataStream, "id, timestamp as ts, temperature as temp");
    
            // 4. 自定义聚合函数，求当前传感器的平均温度值
            // 4.1 table API
            AvgTemp avgTemp = new AvgTemp();
    
            // 需要在环境中注册UDF
            tableEnv.registerFunction("avgTemp", avgTemp);
            Table resultTable = sensorTable
                    .groupBy("id")
                    .aggregate( "avgTemp(temp) as avgtemp" )
                    .select("id, avgtemp");
    
            // 4.2 SQL
            tableEnv.createTemporaryView("sensor", sensorTable);
            Table resultSqlTable = tableEnv.sqlQuery("select id, avgTemp(temp) " +
                    " from sensor group by id");
    
            // 打印输出
            tableEnv.toRetractStream(resultTable, Row.class).print("result");
            tableEnv.toRetractStream(resultSqlTable, Row.class).print("sql");
    
            env.execute();
        }
    
        // 实现自定义的AggregateFunction
        public static class AvgTemp extends AggregateFunction<Double, Tuple2<Double, Integer>>{
            @Override
            public Double getValue(Tuple2<Double, Integer> accumulator) {
                return accumulator.f0 / accumulator.f1;
            }
    
            @Override
            public Tuple2<Double, Integer> createAccumulator() {
                return new Tuple2<>(0.0, 0);
            }
    
            // 必须实现一个accumulate方法，来数据之后更新状态
            public void accumulate( Tuple2<Double, Integer> accumulator, Double temp ){
                accumulator.f0 += temp;
                accumulator.f1 += 1;
            }
        }
    }

# 5.**表聚合函数（****Table Aggregate Functions****）**  #

用户定义的表聚合函数（User-Defined Table Aggregate Functions，UDTAGGs），可以把一个表中数据，聚合为具有多行和多列的结果表。这跟 AggregateFunction 非常类似，只是之前聚合结果是一个标量值，现在变成了一张表。

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1N1eWViaXViaXU_size_16_color_FFFFFF_t_70 4][]

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1N1eWViaXViaXU_size_16_color_FFFFFF_t_70 5][]

⚫ emitUpdateWithRetract()

接下来我们写一个自定义 TableAggregateFunction，用来提取每个 sensor 最高的两个温度值。

// 先定义一个 Accumulator
    public static class Top2TempAcc {
        double highestTemp = Double.MIN_VALUE;
        double secondHighestTemp = Double.MIN_VALUE;
    }
    // 自定义表聚合函数
    public static class Top2Temp extends TableAggregateFunction<Tuple2<Double, Integer>, Top2TempAcc> {
        @Override
        public Top2TempAcc createAccumulator() {
            return new Top2TempAcc();
        }
        // 实现计算聚合结果的函数 accumulate
        public void accumulate(Top2TempAcc acc, Double temp) {
            if (temp > acc.highestTemp) {
                acc.secondHighestTemp = acc.highestTemp;
                acc.highestTemp = temp;
            } else if (temp > acc.secondHighestTemp) {
                acc.secondHighestTemp = temp;
            }
        }
    // 实现一个输出结果的方法，最终处理完表中所有数据时调用 
    public void emitValue(Top2TempAcc acc, Collector<Tuple2<Double, Integer>> out) {
        out.collect(new Tuple2<>(acc.highestTemp, 1));
        out.collect(new Tuple2<>(acc.secondHighestTemp, 2));
    }
    }

接下来就可以在代码中调用了。

// 创建一个表聚合函数实例
    Top2Temp top2Temp = new Top2Temp();
    tableEnv.registerFunction("top2Temp", top2Temp);
    Table resultTable = sensorTable.groupBy("id").flatAggregate("top2Temp(temperature) as (temp, rank)").select("id, temp, rank");
    tableEnv.toRetractStream(resultTable, Row.class).print("result");

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1N1eWViaXViaXU_size_16_color_FFFFFF_t_70]: /images/20221023/08015458674144bdac33a477d83fdb6f.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1N1eWViaXViaXU_size_16_color_FFFFFF_t_70 1]: /images/20221023/6dcf9c758a4b4821ae0777dfc0df7897.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1N1eWViaXViaXU_size_16_color_FFFFFF_t_70 2]: /images/20221023/6ff90636104a4230aa8ea0294c2b35ad.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1N1eWViaXViaXU_size_16_color_FFFFFF_t_70 3]: /images/20221023/5e6a1cee3af44ccfba673986ec9afed6.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1N1eWViaXViaXU_size_16_color_FFFFFF_t_70 4]: /images/20221023/00a7a0808cd34567bffa1480c6f99dc6.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1N1eWViaXViaXU_size_16_color_FFFFFF_t_70 5]: /images/20221023/53c48afc08ad4236a92f7429705d0498.png