如何使用FlinkSQL内置函数
本篇内容介绍了“如何使用FlinkSQL内置函数”的有关知识，在实际案例的操作过程中，不少人都会遇到这样的困境，接下来就让小编带领大家学习一下如何处理这些情况吧！希望大家仔细阅读，能够学有所成！
前言
Flink Table 和 SQL 内置了很多 SQL 中支持的函数;如果有无法满足的需要，则可以实现用户自定义的函数(UDF)来解决。
一、系统内置函数
Flink Table API 和 SQL 为用户提供了一组用于数据转换的内置函数。SQL 中支持的很多函数，Table API 和 SQL 都已经做了实现，其它还在快速开发扩展中。
以下是一些典型函数的举例，全部的内置函数，可以参考官网介绍。
二、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.8sensor_6,1547718201,15.4sensor_7,1547718202,6.7sensor_10,1547718205,38.1sensor_1,1547718206,32sensor_1,1547718208,36.2sensor_1,1547718210,29.7sensor_1,1547718213,30.9
代码如下
packageudfimportorg.apache.flink.streaming.api.scala._importorg.apache.flink.table.api.DataTypesimportorg.apache.flink.table.api.scala._importorg.apache.flink.table.descriptors.{Csv,FileSystem,Schema}importorg.apache.flink.table.functions.ScalarFunctionimportorg.apache.flink.types.Row/***@Packageudf*@File：FlinkSqlUdfHashCode.java*@author大数据老哥*@date2020/12/2921:58*@versionV1.0*/objectFlinkSqlUdfHashCode{defmain(args:Array[String]):Unit={//1.构建运行环境valenv=StreamExecutionEnvironment.getExecutionEnvironmentenv.setParallelism(1)//设置并行度为1//2.构建TableEnvvaltableEnv=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)//注册udfvaltableSql=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|sparkhello|Flink,hello|java,hello|大数据老哥
编写代码
packageudfimportorg.apache.flink.streaming.api.scala._importorg.apache.flink.table.api.scala._importorg.apache.flink.table.functions.TableFunctionimportorg.apache.flink.types.Row/***@Packageudf*@File：FlinkSqlUDFTableFunction.java*@author大数据老哥*@date2020/12/2923:10*@versionV1.0*/objectFlinkSqlUDFTableFunction{defmain(args:Array[String]):Unit={//1.构建运行环境valenv=StreamExecutionEnvironment.getExecutionEnvironmentenv.setParallelism(1)//设置并行度为1//2.构建TableEnvvaltableEnv=StreamTableEnvironment.create(env)//3.构建数据源valdata=env.readTextFile("./data/words.txt")//解析数据valwordData:DataStream[String]=data.flatMap(_.split(","))//类型转换valtableWord=tableEnv.fromDataStream(wordData,'id)//调用TableFunctionvalsplit=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,62,Milk,33,Breve,54,Mocha,85,Tea,4
代码如下
packageudfimportorg.apache.calcite.rel.`type`.{RelDataType,RelDataTypeFactory}importorg.apache.flink.streaming.api.scala._importorg.apache.flink.table.api.DataTypesimportorg.apache.flink.table.api.scala._importorg.apache.flink.table.descriptors.{Csv,FileSystem,Schema}importorg.apache.flink.table.functions.AggregateFunctionimportorg.apache.flink.types.Rowimportjava.util/***@Packageudf*@File：FlinkSQUDFAggregateFunction.java*@author大数据老哥*@date2020/12/3022:06*@versionV1.0*/objectFlinkSQUDFAggregateFunction{defmain(args:Array[String]):Unit={//1.构建运行环境valenv=StreamExecutionEnvironment.getExecutionEnvironmentenv.setParallelism(1)//设置并行度为1//2.构建TableEnvvaltableEnv=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.0varcount:Int=0}classAvgTempextendsAggregateFunction[Double,AvgTempAcc]{overridedefgetValue(acc:AvgTempAcc):Double={acc.sum/acc.count}overridedefcreateAccumulator():AvgTempAcc=newAvgTempAcc()}defaccumulate(accumulator:AvgTempAcc,price:Double):Unit={accumulator.sum+=priceaccumulator.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,62,Milk,33,Breve,54,Mocha,85,Tea,4
代码如下
packageudfimportorg.apache.flink.streaming.api.scala._importorg.apache.flink.table.api.DataTypesimportorg.apache.flink.table.api.scala._importorg.apache.flink.table.descriptors.{Csv,FileSystem,Schema}importorg.apache.flink.table.functions.TableAggregateFunctionimportorg.apache.flink.types.Rowimportorg.apache.flink.util.Collectorimportudf.FlinkSQUDFAggregateFunction.AvgTemp/***@Packageudf*@File：FlinkSqlUDFTableAggregateFunction.java*@author大数据老哥*@date2020/12/3022:53*@versionV1.0*/objectFlinkSqlUDFTableAggregateFunction{defmain(args:Array[String]):Unit={//1.构建运行环境valenv=StreamExecutionEnvironment.getExecutionEnvironmentenv.setParallelism(1)//设置并行度为1//2.构建TableEnvvaltableEnv=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.MinValuevarsecodeHighestPrice:Double=Int.MinValue}classTop2TempextendsTableAggregateFunction[(Double,Int),Top2TempAcc]{overridedefcreateAccumulator():Top2TempAcc=newTop2TempAccdefaccumulate(acc:Top2TempAcc,temp:Double):Unit={if(temp>acc.highestPrice){acc.secodeHighestPrice=acc.highestPriceacc.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)}}}
“如何使用FlinkSQL内置函数”的内容就介绍到这里了，感谢大家的阅读。如果想了解更多行业相关的知识可以关注亿速云网站，小编将为大家输出更多高质量的实用文章！
声明：本站所有文章资源内容，如无特殊说明或标注，均为采集网络资源。如若本站内容侵犯了原著者的合法权益，可联系本站删除。
