FlinkSQL内置了这么多函数你都使用过吗？

前言

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

一、系统内置函数

Flink Table API 和 SQL 为用户提供了一组用于数据转换的内置函数。SQL 中支持的很多函数，Table API 和 SQL 都已经做了实现，其它还在快速开发扩展中。

以下是一些典型函数的举例，全部的内置函数，可以参考官网介绍。

类型	TableApi	SQLAPI
比较函数	ANY1 === ANY2	value1 = value2
比较函数	NY1 > ANY2	value1 > value2
逻辑函数	BOOLEAN1 || BOOLEAN2	boolean1 OR boolean2
逻辑函数	BOOLEAN.isFalse	boolean IS FALSE
逻辑函数	!BOOLEAN	NOT boolean
算术函数	NUMERIC1 + NUMERIC2	numeric1 + numeric2
算术函数	NUMERIC1.power(NUMERIC2)	POWER(numeric1, numeric2)
字符串函数	STRING1 + STRING2	string1 || string2
字符串函数	STRING.upperCase()	UPPER(string)
字符串函数	STRING.charLength()	CHAR_LENGTH(string)
时间函数	STRING.toDate	DATE string
时间函数	STRING.toTimestamp	TIMESTAMP string
时间函数	currentTime()	CURRENT_TIME
时间函数	NUMERIC.days	INTERVAL string range
时间函数	NUMERIC.minutes
聚合函数	FIELD.count	COUNT(*)
聚合函数	FIELD.sum0	SUM([ ALL | DISTINCT ] expression)
聚合函数		RANK()
聚合函数		ROW_NUMBER()

二、Flink UDF

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

2.1 注册用户自定义函数 UDF

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

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

2.2 标量函数(Scalar Functions)

用户定义的标量函数，可以将 0、1 或多个标量值，映射到新的标量值。

为了定义标量函数，必须在 org.apache.flink.table.functions 中扩展基类 Scalar Function，并实现(一个或多个)求值(evaluation，eval)方法。标量函数的行为由求值方法决定，求值方法必须公开声明并命名为 eval(直接 def 声明，没有 override)。求值方法的参数类型和返回类型，确定了标量函数的参数和返回类型。

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

准备数据

sensor_1,1547718199,35.8
sensor_6,1547718201,15.4
sensor_7,1547718202,6.7
sensor_10,1547718205,38.1
sensor_1,1547718206,32
sensor_1,1547718208,36.2
sensor_1,1547718210,29.7
sensor_1,1547718213,30.9

代码如下

packageudf

importorg.apache.flink.streaming.api.scala._
importorg.apache.flink.table.api.DataTypes
importorg.apache.flink.table.api.scala._
importorg.apache.flink.table.descriptors.{Csv,FileSystem,Schema}
importorg.apache.flink.table.functions.ScalarFunction
importorg.apache.flink.types.Row

/**
*@Packageudf
*@File：FlinkSqlUdfHashCode.java
*@author大数据老哥
*@date2020/12/2921:58
*@versionV1.0
*/
objectFlinkSqlUdfHashCode{
defmain(args:Array[String]):Unit={
//1.构建运行环境
valenv=StreamExecutionEnvironment.getExecutionEnvironment
env.setParallelism(1)//设置并行度为1
//2.构建TableEnv
valtableEnv=StreamTableEnvironment.create(env)
//3.构建数据源
tableEnv.connect(newFileSystem().path("./data/sensor.txt"))
.withFormat(newCsv())
.withSchema(newSchema()
.field("id",DataTypes.STRING())
.field("timestamp",DataTypes.INT())
.field("temperature",DataTypes.DOUBLE())
).createTemporaryTable("sensor")
//转为表
valtableSensor=tableEnv.from("sensor")
//床架转换对象
valcode=newHashCode()
//使用tableAPI进行测试
valtableRes=tableSensor.select('id,code('id))
tableEnv.registerFunction("code",code)//注册udf
valtableSql=tableEnv.sqlQuery(
"""
|select
|id,
|code(id)
|from
|sensor
|""".stripMargin)
//输出
tableRes.toAppendStream[Row].print("tableAPI")
tableSql.toAppendStream[Row].print("tableSql")

env.execute("FlinkSqlUdfHashCode")
}

classHashCode()extendsScalarFunction{
defeval(s:String):String={
s.hashCode.toString
}
}
}
运行结果

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()，或者带有 ON TRUE 条件的左连接。

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

数据准备

hello|word,hello|spark
hello|Flink,hello|java,hello|大数据老哥

编写代码

packageudf

importorg.apache.flink.streaming.api.scala._
importorg.apache.flink.table.api.scala._
importorg.apache.flink.table.functions.TableFunction
importorg.apache.flink.types.Row

/**
*@Packageudf
*@File：FlinkSqlUDFTableFunction.java
*@author大数据老哥
*@date2020/12/2923:10
*@versionV1.0
*/
objectFlinkSqlUDFTableFunction{
defmain(args:Array[String]):Unit={
//1.构建运行环境
valenv=StreamExecutionEnvironment.getExecutionEnvironment
env.setParallelism(1)//设置并行度为1
//2.构建TableEnv
valtableEnv=StreamTableEnvironment.create(env)
//3.构建数据源
valdata=env.readTextFile("./data/words.txt")
//解析数据
valwordData:DataStream[String]=data.flatMap(_.split(","))
//类型转换
valtableWord=tableEnv.fromDataStream(wordData,'id)
//调用TableFunction
valsplit=newSplit()
//TableAPI方式一
valresTable1=tableWord.
joinLateral(split('id)as('word,'length))
.select('id,'word,'length)
//TableAPI方式二
valresTable2=tableWord.
leftOuterJoinLateral(split('id)as('word,'length))
.select('id,'word,'length)
//将数据注册成表
tableEnv.createTemporaryView("sensor",tableWord)
tableEnv.registerFunction("split",split)

//SQL方式一
valtableSQL1=tableEnv.sqlQuery(
"""
|select
|id,
|word,
|length
|from
|sensor,LATERALTABLE(split(id))ASnewsensor(word,length)
|""".stripMargin)
//SQL方式二
valTableSQL2=tableEnv.sqlQuery(
"""
|select
|id,
|word,
|length
|from
|sensor
|LEFTJOINLATERALTABLE(split(id))ASnewsensor(word,length)ONTRUE
|""".stripMargin)
//调用数据
resTable1.toAppendStream[Row].print("resTable1")
resTable2.toAppendStream[Row].print("resTable2")
tableSQL1.toAppendStream[Row].print("tableSQL1")
TableSQL2.toAppendStream[Row].print("TableSQL2")


env.execute("FlinkSqlUDFTableFunction")
}

classSplit()extendsTableFunction[(String,Int)]{
defeval(str:String):Unit={
str.split("\\|").foreach(
word=>collect((word,word.length))
)
}
}
}

2.4 聚合函数(Aggregate Functions)

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

上图中显示了一个聚合的例子。

假设现在有一张表，包含了各种饮料的数据。该表由三列(id、name 和 price)、五行组成数据。现在我们需要找到表中所有饮料的最高价格，即执行 max()聚合，结果将是一个数值。AggregateFunction 的工作原理如下：

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

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

• retract()
• merge()
• resetAccumulator()

接下来我们写一个自定义AggregateFunction,计算一个每个price的平均值。

数据准备

1,Latte,6
2,Milk,3
3,Breve,5
4,Mocha,8
5,Tea,4

代码如下

packageudf

importorg.apache.calcite.rel.`type`.{RelDataType,RelDataTypeFactory}
importorg.apache.flink.streaming.api.scala._
importorg.apache.flink.table.api.DataTypes
importorg.apache.flink.table.api.scala._
importorg.apache.flink.table.descriptors.{Csv,FileSystem,Schema}
importorg.apache.flink.table.functions.AggregateFunction
importorg.apache.flink.types.Row

importjava.util

/**
*@Packageudf
*@File：FlinkSQUDFAggregateFunction.java
*@author大数据老哥
*@date2020/12/3022:06
*@versionV1.0
*/
objectFlinkSQUDFAggregateFunction{
defmain(args:Array[String]):Unit={
//1.构建运行环境
valenv=StreamExecutionEnvironment.getExecutionEnvironment
env.setParallelism(1)//设置并行度为1
//2.构建TableEnv
valtableEnv=StreamTableEnvironment.create(env)
//3.构建数据源
tableEnv.connect(newFileSystem().path("./data/datas"))
.withFormat(newCsv)
.withSchema(newSchema()
.field("id",DataTypes.STRING())
.field("name",DataTypes.STRING())
.field("price",DataTypes.DOUBLE())
).createTemporaryTable("datas")
valAvgTemp=newAvgTemp()
valtable=tableEnv.from("datas")

valresTableApi=table.groupBy('id)
.aggregate(AvgTemp('price)as'sumprice)
.select('id,'sumprice)

tableEnv.registerFunction("avgTemp",AvgTemp)
valtablesql=tableEnv.sqlQuery(
"""
|select
|id,avgTemp(price)
|fromdatasgroupbyid
|""".stripMargin)
resTableApi.toRetractStream[Row].print("resTableApi")
tablesql.toRetractStream[Row].print("tablesql")
env.execute("FlinkSQUDFAggregateFunction")
}

classAvgTempAcc{
varsum:Double=0.0
varcount:Int=0
}

classAvgTempextendsAggregateFunction[Double,AvgTempAcc]{
overridedefgetValue(acc:AvgTempAcc):Double={
acc.sum/acc.count
}

overridedefcreateAccumulator():AvgTempAcc=newAvgTempAcc()
}

defaccumulate(accumulator:AvgTempAcc,price:Double):Unit={
accumulator.sum+=price

accumulator.count+=1
}

}

2.5表聚合函数(Table Aggregate Functions)

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

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

• 为首先，它同样需要一个累加器(Accumulator)，它是保存聚合中间结果的数据结构。通过调用 TableAggregateFunction 的 createAccumulator()方法可以创建空累加器。
• 为随后，对每个输入行调用函数的 accumulate()方法来更新累加器。
• 为处理完所有行后，将调用函数的 emitValue()方法来计算并返回最终结果。除了上述方法之外，还有一些可选择实现的方法。
• retract()
• merge()
• resetAccumulator()
• emitValue()
• emitUpdateWithRetract()

接下来我们写一个自定义 TableAggregateFunction，用来提取每个 price 最高的两个平均值。

数据准备

1,Latte,6
2,Milk,3
3,Breve,5
4,Mocha,8
5,Tea,4

代码如下

packageudf

importorg.apache.flink.streaming.api.scala._
importorg.apache.flink.table.api.DataTypes
importorg.apache.flink.table.api.scala._
importorg.apache.flink.table.descriptors.{Csv,FileSystem,Schema}
importorg.apache.flink.table.functions.TableAggregateFunction
importorg.apache.flink.types.Row
importorg.apache.flink.util.Collector
importudf.FlinkSQUDFAggregateFunction.AvgTemp

/**
*@Packageudf
*@File：FlinkSqlUDFTableAggregateFunction.java
*@author大数据老哥
*@date2020/12/3022:53
*@versionV1.0
*/
objectFlinkSqlUDFTableAggregateFunction{
defmain(args:Array[String]):Unit={
//1.构建运行环境
valenv=StreamExecutionEnvironment.getExecutionEnvironment
env.setParallelism(1)//设置并行度为1
//2.构建TableEnv
valtableEnv=StreamTableEnvironment.create(env)
//3.构建数据源
tableEnv.connect(newFileSystem().path("./data/datas"))
.withFormat(newCsv)
.withSchema(newSchema()
.field("id",DataTypes.STRING())
.field("name",DataTypes.STRING())
.field("price",DataTypes.DOUBLE())
).createTemporaryTable("datas")
valtable=tableEnv.from("datas")
valtemp=newTop2Temp()
valtableApi=table.groupBy('id)
.flatAggregate(temp('price)as('tmpprice,'rank))
.select('id,'tmpprice,'rank)
tableEnv.registerFunction("temp",temp)


tableApi.toRetractStream[Row].print()

env.execute("FlinkSqlUDFTableAggregateFunction")
}

classTop2TempAcc{
varhighestPrice:Double=Int.MinValue
varsecodeHighestPrice:Double=Int.MinValue
}

classTop2TempextendsTableAggregateFunction[(Double,Int),Top2TempAcc]{
overridedefcreateAccumulator():Top2TempAcc=newTop2TempAcc

defaccumulate(acc:Top2TempAcc,temp:Double):Unit={
if(temp>acc.highestPrice){
acc.secodeHighestPrice=acc.highestPrice
acc.highestPrice=temp
}elseif(temp>acc.secodeHighestPrice){
acc.highestPrice=temp
}
}

defemitValue(acc:Top2TempAcc,out:Collector[(Double,Int)]):Unit={
out.collect(acc.highestPrice,1)
out.collect(acc.secodeHighestPrice,2)
}
}

}

总结

好了今天的内容就分享到这里了。上述主要讲解了一个系统自己带的函数，但是往往企业中不光只需要这些函数，有好多需求是本身函数是无法完成的。这时候就要用到我们的自定义函数了。他可以根据我们自己的需要进行编写代码来实现我们想要的功能。