Flink基础系列33-Table API和Flink SQL之函数

原创只是甲 2021-11-08

759

一. 函数

1.1 系统内置函数
1.2 UDF

1.2.1 注册用户自定义函数 UDF
1.2.2 标量函数（Scalar Functions）
1.2.3 表函数（Table Functions）
1.2.4 聚合函数（Aggregate Functions）
1.2.5 表聚合函数（Table Aggregate Functions）

二.案例

2.1 Scalar Function
2.2 Table Function
2.3 Aggregate Function

参考:

一. 函数

Flink Table 和 SQL 内置了很多 SQL 中支持的函数；如果有无法满足的需要，则可以实现用户自定义的函数（UDF）来解决。

1.1 系统内置函数

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
NOT boolean

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([ ALL | DISTINCT ] expression)
RANK()
ROW_NUMBER()

Table API：
FIELD.count
FIELD.sum0

1.2 UDF

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

1.2.1 注册用户自定义函数 UDF

在大多数情况下，用户定义的函数必须先注册，然后才能在查询中使用。不需要专门为Scala 的 Table API 注册函数。

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

1.2.2 标量函数（Scalar Functions）

用户定义的标量函数，可以将 0、1 或多个标量值，映射到新的标量值。为了定义标量函数，必须在 org.apache.flink.table.functions 中扩展基类 Scalar Function，并实现（一个或多个）求值（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、建表）：

public static void main(String[] args) throws Exception {
// 1. 创建环境

    StreamExecutionEnvironment env =StreamExecutionEnvironment.getExecutionEnvironment();

    env.setParallelism(1);

    StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);


    // 2. 读取文件，得到 DataStream
    String filePath = "..\\sensor.txt";
    DataStream<String> inputStream = env.readTextFile(filePath);


    // 3. 转换成 Java Bean，并指定 timestamp 和 watermark

    DataStream<SensorReading> dataStream = inputStream.map( line -> {
String[] fields = line.split(",");

    return new SensorReading(fields[0], new Long(fields[1]), new Double(fields[2]));
} );


    // 4. 将 DataStream 转换为 Table
    Table sensorTable = tableEnv.fromDataStream(dataStream, "id, timestamp as ts, temperature");

    // 5. 调用自定义 hash 函数，对 id 进行 hash 运算
    HashCode hashCode = new HashCode(23);
    tableEnv.registerFunction("hashCode", hashCode);
    Table resultTable = sensorTable.select("id, ts, hashCode(id)");

    //	sql
    tableEnv.createTemporaryView("sensor", sensorTable);
Table resultSqlTable = tableEnv.sqlQuery("select id, ts, hashCode(id) from sensor");
    tableEnv.toAppendStream(resultTable, Row.class).print("result");
    tableEnv.toRetractStream(resultSqlTable, Row.class).print("sql");

    env.execute("scalar function test");

1.2.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（<TableFunction>），或者带有 ON TRUE 条件的左连接。

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

// 自定义 TableFunction

public static class Split extends TableFunction<Tuple2<String, Integer>> {

private String separator = ",";


public Split(String separator) {
this.separator = separator;
}


// 类似 flatmap，没有返回值
public void eval(String str) {
for (String s : str.split(separator)) {
collect(new Tuple2<String, Integer>(s, s.length()));
}

}

}

接下来，就是在代码中调用。首先是 Table API 的方式：

Split split = new Split("_"); tableEnv.registerFunction("split", split); Table resultTable = sensorTable
.joinLateral( "split(id) as (word, length)")
.select("id, ts, word, length");

然后是 SQL 的方式：

tableEnv.createTemporaryView("sensor", sensorTable);

Table resultSqlTable = tableEnv.sqlQuery("select id, ts, word, length " +
"from sensor, lateral table( split(id) ) as splitId(word, length)");

1.2.4 聚合函数（Aggregate Functions）

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

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

AggregateFunction 的工作原理如下。
⚫ 首先，它需要一个累加器，用来保存聚合中间结果的数据结构（状态）。可以通过调用 AggregateFunction 的 createAccumulator（）方法创建空累加器。
⚫ 随后，对每个输入行调用函数的 accumulate（）方法来更新累加器。
⚫ 处理完所有行后，将调用函数的 getValue（）方法来计算并返回最终结果。 AggregationFunction 要求必须实现的方法：
• createAccumulator()
• accumulate()
• getValue()

除了上述方法之外，还有一些可选择实现的方法。其中一些方法，可以让系统执行查询更有效率，而另一些方法，对于某些场景是必需的。例如，如果聚合函数应用在会话窗口
（session group window）的上下文中，则 merge（）方法是必需的。
• retract()
• merge()
• resetAccumulator()

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

// 定义 AggregateFunction 的 Accumulator

public static class AvgTempAcc {

double sum = 0.0;

int count = 0;

}

// 自定义一个聚合函数，求每个传感器的平均温度值，保存状态(tempSum,  tempCount)

public static class AvgTemp extends AggregateFunction<Double, AvgTempAcc>{

@Override

public Double getValue(AvgTempAcc accumulator) {

return accumulator.sum / accumulator.count;

}

@Override

public AvgTempAcc createAccumulator() {

return new AvgTempAcc();

}

// 实现一个具体的处理计算函数，accumulate

public void accumulate( AvgTempAcc accumulator, Double temp) {
accumulator.sum += temp;

accumulator.count += 1;

}
}

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

// 创建一个聚合函数实例

AvgTemp avgTemp = new AvgTemp();


// Table API 的调用

tableEnv.registerFunction("avgTemp", avgTemp); Table resultTable = sensorTable
.groupBy("id")
.aggregate("avgTemp(temperature) as avgTemp")
.select("id, avgTemp");

// sql
tableEnv.createTemporaryView("sensor", sensorTable);
Table resultSqlTable = tableEnv.sqlQuery("select id, avgTemp(temperature) " +
"from sensor group by id");

tableEnv.toRetractStream(resultTable, Row.class).print("result"); tableEnv.toRetractStream(resultSqlTable, Row.class).print("sql");

1.2.5 表聚合函数（Table Aggregate Functions）

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

比如现在我们需要找到表中所有饮料的前 2 个最高价格，即执行 top2（）表聚合。我们需要检查 5 行中的每一行，得到的结果将是一个具有排序后前 2 个值的表。
用户定义的表聚合函数，是通过继承 TableAggregateFunction 抽象类来实现的。

TableAggregateFunction 的工作原理如下。
⚫ 首先，它同样需要一个累加器（Accumulator），它是保存聚合中间结果的数据结构。
通过调用 TableAggregateFunction 的 createAccumulato（r）方法可以创建空累加器。

⚫ 随后，对每个输入行调用函数的 accumulate（）方法来更新累加器。

⚫ 处理完所有行后，将调用函数的 emitValue（）方法来计算并返回最终结果。

AggregationFunction 要求必须实现的方法：
• createAccumulator()
• accumulate()

除了上述方法之外，还有一些可选择实现的方法。
• retract()
• merge()
• resetAccumulator()
• emitValue()
• 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");

二.案例

2.1 Scalar Function

代码:

package org.flink.tableapi.udf;

import org.flink.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;

/**
 * @author 只是甲
 * @date   2021-09-30
 */

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("C:\\Users\\Administrator\\IdeaProjects\\FlinkStudy\\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.registerTable("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;
        }
    }
}

测试记录:

2.2 Table Function

代码:

package org.flink.tableapi.udf;


import org.flink.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;

/**
 * @author 只是甲
 * @date   2021-09-30
 */

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("C:\\Users\\Administrator\\IdeaProjects\\FlinkStudy\\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.registerTable("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()));
            }
        }
    }
}

测试记录：

2.3 Aggregate Function

代码:

package org.flink.tableapi.udf;


import org.flink.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;


/**
 * @author 只是甲
 * @date   2021-09-30
 */

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("C:\\Users\\Administrator\\IdeaProjects\\FlinkStudy\\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.registerTable("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;
        }
    }
}

测试记录：

参考:

https://www.bilibili.com/video/BV1qy4y1q728
https://ashiamd.github.io/docsify-notes/#/study/BigData/Flink/%E5%B0%9A%E7%A1%85%E8%B0%B7Flink%E5%85%A5%E9%97%A8%E5%88%B0%E5%AE%9E%E6%88%98-%E5%AD%A6%E4%B9%A0%E7%AC%94%E8%AE%B0?id=_11-table-api%e5%92%8cflink-sql

flink

最后修改时间：2021-11-26 18:18:22

「喜欢这篇文章，您的关注和赞赏是给作者最好的鼓励」

关注作者

文章被以下合辑收录

大数据及数据仓库（共239篇）

DBA转大数据之路

Flink基础系列33-Table API和Flink SQL之函数

Table of Contents

一. 函数

1.1 系统内置函数

1.2 UDF

1.2.1 注册用户自定义函数 UDF

1.2.2 标量函数（Scalar Functions）

1.2.3 表函数（Table Functions）

1.2.4 聚合函数（Aggregate Functions）

1.2.5 表聚合函数（Table Aggregate Functions）

二.案例

2.1 Scalar Function

2.2 Table Function

2.3 Aggregate Function

参考:

文章被以下合辑收录

评论