Flink-TableAPI与SQL-函数与自定义函数

àì夳堔傛蜴生んèń 2022-09-10 02:29 199阅读 0赞

## 函数(Functions) ##

Flink Table API 和 SQL 为用户提供了一组用于数据转换的内置函数

SQL 中支持的很多函数,Table API 和 SQL 都已经做了实现

**比较函数**

**SQL:**

value1 = value2

value1 > value2

**Table API:**

ANY1 === ANY2

ANY1 > ANY2

**逻辑函数**

**SQL:**

boolean1 OR boolean2

boolean IS FALSE （取反，如果是 null is false 的话结果是false）

NOT boolean （取反，如果是 not null 的话结果是null）

**Table API:**

BOOLEAN1 || BOOLEAN2

BOOLEAN.isFalse

!BOOLEAN

**算数函数**

**SQL:**

numeric1 + numeric2

POWER(numeric1, numeric2)

**Table API:**

NUMERIC1 + NUMERIC2

NUMERIC1.power(NUMERIC2)

**字符串函数**

**SQL:**

string1 || string2 （字符串连接）

UPPER(string)

CHAR\_LENGTH(string)

**Table API:**

STRING1 + STRING2

STRING.upperCase()

STRING.charLength()

**时间函数**

**SQL:**

DATE string

TIMESTAMP string

CURRENT\_TIME

INTERVAL string range

**Table API:**

STRING.toDate

STRING.toTimestamp

currentTime()

NUMERIC.days

NUMERIC.minutes

**聚合函数**

**SQL:**

COUNT(\*)

SUM(expression)

RANK()

ROW\_NUMBER()

**Table API:**

FIELD.count

FIELD.sum0 （对都是空值和零值求和的话结果是0，如果sum的话是null）

**目前还有很多聚合函数TableAPI还不支持**

### 用户自定义函数(UDF) ###

用户定义函数(User-defined Functions,UDF)是一个重要的特性,它们显著地扩展了查询的表达能力

在大多数情况下,用户定义的函数必须先注册,然后才能在查询中使用

函数通过调用 registerFunction()方法在 TableEnvironment 中注册。当用户定义的函数被注册时,它被插入到 TableEnvironment 的函数目录中,这样Table API 或 SQL 解析器就可以识别并正确地解释它

#### 标量函数(Scalar Functions) ####

用户定义的标量函数,可以将0、1或多个标量值,映射到新的标量值，标量函数只返回一个标量值，一对一相当于map。

为了定义标量函数,必须在 org.apache.flink.table.functions 中扩展基类Scalar Function,并实现(一个或多个)求值(eval)方法

标量函数的行为由求值方法决定,求值方法必须公开声明并命名为 eval

public static void main(String[] args) throws Exception { 
    
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
    
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);
    
        // 从文件中读取数据
        String path = "/home/lxj/workspace/Flink/src/main/resources/sensor.txt";
        DataStream<String> inputDataStream = env.readTextFile(path);
    
        // 转换成POJO
        DataStream<SensorReading> dataStream = inputDataStream.map(line -> { 
            String[] fields = line.split(",");
            return new SensorReading(fields[0], new Long(fields[1]), new Double(fields[2]));
        });
    
        // 将流转换成表
        Table sensorTable = tableEnv.fromDataStream(dataStream, "id, timestamp as ts, temperature as temp");
    
        // 自定义标量函数，实现求id的hash值
        // tableAPI
        HashCode hashCode = new HashCode(23);
        // 需要在环境中注册UDF
        tableEnv.registerFunction("hashCode", hashCode);
    
        Table hashCodeResult = sensorTable.select("id, ts, hashCode(id)");
    
        // SQL
        tableEnv.createTemporaryView("sensor", sensorTable);
    
        Table hashCodeSqlResult = tableEnv.sqlQuery("select id, ts, hashCode(id) from sensor");
    
        tableEnv.toAppendStream(hashCodeResult, Row.class).print("table");
        // table> sensor_1,1547718199,-1036124876
        // table> sensor_6,1547718201,-1036124761
        // table> sensor_7,1547718202,-1036124738
        // table> sensor_10,1547718205,-2055098980
        // table> sensor_1,1547710000,-1036124876
        // table> sensor_1,1547719999,-1036124876
        // table> sensor_1,1547715555,-1036124876
    
        tableEnv.toAppendStream(hashCodeSqlResult, Row.class).print("sql");
        // sql> sensor_1,1547718199,-1036124876
        // sql> sensor_6,1547718201,-1036124761
        // sql> sensor_7,1547718202,-1036124738
        // sql> sensor_10,1547718205,-2055098980
        // sql> sensor_1,1547710000,-1036124876
        // sql> sensor_1,1547719999,-1036124876
        // sql> sensor_1,1547715555,-1036124876
    
        env.execute();
    }
    
    // 实现自定义的ScalarFunction
    public static class HashCode extends ScalarFunction { 
    
        private int factor = 13;
    
        public HashCode(int factor) { 
    
            this.factor = factor;
        }
    
        // 必须有一个叫eval的public的方法
        public int eval(String str) { 
    
            return str.hashCode() * factor;
        }
    
        public int getFactor() { 
            return factor;
        }
    
        public void setFactor(int factor) { 
            this.factor = factor;
        }
    
    }

#### 表函数(Table Functions) ####

用户定义的表函数,也可以将0、1或多个标量值作为输入参数;与标量函数不同的是,它可以返回任意数量的行作为输出,而不是单个值。相当于把一行或者多行的数据炸裂出多行，一对多相当于flatMap。

为了定义一个表函数,必须扩展 org.apache.flink.table.functions 中的基类TableFunction 并实现(一个或多个)求值方法

表函数的行为由其求值方法决定,求值方法必须是 public 的,并命名为 eval

// 将流转换成表
    Table sensorTable = tableEnv.fromDataStream(dataStream, "id, timestamp as ts, temperature as temp");
    
    // 自定义表函数，实现将id拆分，并输出（word, length）
    // tableAPI
    Split split = new Split("_");
    // 需要在环境中注册UDF
    tableEnv.registerFunction("split", split);
    
    Table result = sensorTable.joinLateral("split(id) as (word, length)")
        .select("id, ts, word, length");
    
    // SQL
    tableEnv.createTemporaryView("sensor", sensorTable);
    
    Table sqlTable = tableEnv.sqlQuery("select id, ts, word, length " +
                                       "from sensor, lateral table(split(id)) as splitid(word, length)");
    
    tableEnv.toAppendStream(result, Row.class).print("table");
    // table> sensor_1,1547718199,sensor,6
    // table> sensor_1,1547718199,1,1
    // table> sensor_6,1547718201,sensor,6
    // table> sensor_6,1547718201,6,1
    // table> sensor_7,1547718202,sensor,6
    // table> sensor_7,1547718202,7,1
    // table> sensor_10,1547718205,sensor,6
    // table> sensor_10,1547718205,10,2
    // table> sensor_1,1547710000,sensor,6
    // table> sensor_1,1547710000,1,1
    // table> sensor_1,1547719999,sensor,6
    // table> sensor_1,1547719999,1,1
    // table> sensor_1,1547715555,sensor,6
    // table> sensor_1,1547715555,1,1
    
    tableEnv.toAppendStream(sqlTable, Row.class).print("sql");
    // sql> sensor_1,1547718199,sensor,6
    // sql> sensor_1,1547718199,1,1
    // sql> sensor_6,1547718201,sensor,6
    // sql> sensor_6,1547718201,6,1
    // sql> sensor_7,1547718202,sensor,6
    // sql> sensor_7,1547718202,7,1
    // sql> sensor_10,1547718205,sensor,6
    // sql> sensor_10,1547718205,10,2
    // sql> sensor_1,1547710000,sensor,6
    // sql> sensor_1,1547710000,1,1
    // sql> sensor_1,1547719999,sensor,6
    // sql> sensor_1,1547719999,1,1
    // sql> sensor_1,1547715555,sensor,6
    // sql> sensor_1,1547715555,1,1

#### 聚合函数(Aggregate Functions) ####

用户自定义聚合函数(User-Defined Aggregate Functions,UDAGGs)可以把一个表中的数据,聚合成一个标量值，多对一

用户定义的聚合函数,是通过继承 AggregateFunction 抽象类实现的  
![在这里插入图片描述][watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBA6a2U56ybTG92ZQ_size_20_color_FFFFFF_t_70_g_se_x_16]

AggregationFunction要求必须实现的方法:

createAccumulator()

accumulate()

getValue()

AggregateFunction 的工作原理如下:

首先,它需要一个累加器(Accumulator),用来保存聚合中间结果的数据结构;可以通过调用 createAccumulator() 方法创建空累加器

随后,对每个输入行调用函数的 accumulate() 方法来更新累加器

处理完所有行后,将调用函数的 getValue() 方法来计算并返回最终结果

// 自定义聚合函数，求当前传感器的平均温度值
    // tableAPI
    AvgTemp avgTemp = new AvgTemp();
    // 需要在环境中注册UDF
    tableEnv.registerFunction("avgTemp", avgTemp);
    
    Table result = sensorTable.groupBy("id")
        .aggregate("avgTemp(temp) as avgTemp")
        .select("id, avgTemp");
    
    // SQL
    tableEnv.createTemporaryView("sensor", sensorTable);
    
    Table sqlResult = tableEnv.sqlQuery("select id, avgTemp(temp) from sensor group by id");
    
    tableEnv.toRetractStream(result, Row.class).print("table");
    // table> (true,sensor_1,35.8)
    // table> (true,sensor_6,15.4)
    // table> (true,sensor_7,6.7)
    // table> (true,sensor_10,38.1)
    // table> (false,sensor_1,35.8)
    // table> (true,sensor_1,36.3)
    // table> (false,sensor_1,36.3)
    // table> (true,sensor_1,35.8)
    // table> (false,sensor_1,35.8)
    // table> (true,sensor_1,36.3)
    
    tableEnv.toRetractStream(sqlResult, Row.class).print("sql");
    // sql> (true,sensor_1,35.8)
    // sql> (true,sensor_6,15.4)
    // sql> (true,sensor_7,6.7)
    // sql> (true,sensor_10,38.1)
    // sql> (false,sensor_1,35.8)
    // sql> (true,sensor_1,36.3)
    // sql> (false,sensor_1,36.3)
    // sql> (true,sensor_1,35.8)
    // sql> (false,sensor_1,35.8)
    // sql> (true,sensor_1,36.3)

#### 表聚合函数(Table Aggregate Functions) ####

用户定义的表聚合函数(User-Defined Table Aggregate Functions,UDTAGGs),可以把一个表中数据,聚合为具有多行和多列的结果表，多对多

用户定义表聚合函数,是通过继承 TableAggregateFunction 抽象类来实现的

![在这里插入图片描述][watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBA6a2U56ybTG92ZQ_size_20_color_FFFFFF_t_70_g_se_x_16 1]

AggregationFunction 要求必须实现的方法:

createAccumulator()

accumulate()

emitValue()

TableAggregateFunction 的工作原理如下:

首先,它同样需要一个累加器(Accumulator),它是保存聚合中间结果的数据结构。通过调用 createAccumulator() 方法可以创建空累加器。

随后,对每个输入行调用函数的 accumulate() 方法来更新累加器。

处理完所有行后,将调用函数的 emitValue() 方法来计算并返回最终结果。

[watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBA6a2U56ybTG92ZQ_size_20_color_FFFFFF_t_70_g_se_x_16]: /images/20220829/aba6fa36aa054282bf984243050d975a.png
[watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBA6a2U56ybTG92ZQ_size_20_color_FFFFFF_t_70_g_se_x_16 1]: /images/20220829/efd6e7ff006049ac99e4fff94473bc92.png