如何使用FlinkSQL内置函数
前言
Flink Table 和 SQL 有很多 SQL 支持的内置函数;如果存在无法满足的需求,可以实现用户自定义函数(UDF)来解决。
1.系统内置函数
Flink Table API 和 SQL 为用户提供了一组用于数据转换的内置函数。 SQL支持的许多功能已经在Table API和SQL中实现,其他功能仍在快速开发和扩展中。
以下是一些典型函数的示例。所有内置功能请参考官方网站。
类型 | TableApi | SQLAPI |
---|---|---|
比较函数 | ANY1 === ANY2 | value1 = value2< /td> |
比较函数 | NY1 > ANY2 | value1 > value2 |
逻辑函数 | BOOLEAN1 || BOOLEAN2 | boolean1 OR boolean2 |
逻辑函数 | BOOLEAN.isFalse | boolean IS FALSE |
逻辑函数 | !BOOLEAN | NOT boolean | < /tr>
算术函数 | NUMERIC1 + NUMERIC2 | numeric1 + numeric2 |
NUMERIC1.power(NUMERIC2) | POWER(numeric1, numeric2) | |
字符串函数 | STRING1 + STRING2 | string1 || string2 |
字符串函数 | STRING.upperCase() | UPPER(字符串) |
字符串函数 | STRING.charLength() | CHAR_LENGTH(string) |
时间函数 | < td>STRING.toDate日期字符串 | |
时间函数 | STRING.toTimestamp | TIMESTAMP 字符串 |
时间函数 | currentTime() | CURRENT_TIME |
时间函数 | NUMERIC.days | INTERVAL字符串范围 |
时间函数 | NUMERIC.分钟 | |
聚合函数< /strong> | FIELD.count | COUNT(*) |
聚合函数 | FIELD.sum0 | SUM([ ALL | DISTINCT ]表达式) |
聚合函数 | RANK() | |
聚合函数 | ROW_NUMBER() |
2. Flink UDF
用户定义函数(UDF)是一个这是一个重要的功能,因为它们显着扩展了查询的表达能力。对于一些系统内置函数无法解决的需求,我们可以使用UDF来定制实现。
2.1 注册用户定义函数UDF
大多数情况下,用户定义函数必须先注册才能在查询中使用。无需专门为 Scala 的 Table API 注册函数。
通过调用registerFunction()方法将函数注册到TableEnvironment中。当用户定义的函数被注册时,它被插入到TableEnvironment的函数目录中,以便Table API或SQL解析器可以识别它并正确解释它。
2.2 标量函数
用户定义的标量函数可以将0、1或多个标量值映射到新的标量值。
为了定义标量函数,必须在org.apache.flink.table.fu中扩展基类Scalar Function功能并实施(一个或多个)评估(eval)方法。标量函数的行为由求值方法决定,该方法必须公开声明并命名为 eval(直接 def 声明,无覆盖)。求值方法的参数类型和返回类型决定了标量函数的参数和返回类型。
在下面的代码中,我们定义了自己的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
代码如下
package udf import org.apache.flink.streaming.api.scala._ import org.apache.flink.table.api.DataTypes 导入 org.apache.flink.table.api.scala._ 导入 org.apache.flink.table.descriptors。 {Csv,文件系统,架构}导入 org.apache.flink.table.functions.ScalarFunction import org.apache.flink.types.Row /** * @Package udf * @File: FlinkSqlUdfHashCode.java * @author 大数据兄弟 * @date 2023/12/29 21:58 * @version V1.0 */ object FlinkSqlUdfHashCode { def main(args: Array[String]): Unit = { //1.构建运行环境 val env = StreamExecutionEnvironment.getExecutionEnvironment env.setParallelism(1) //设置并行度为1 //2.构建 TableEnv val tableEnv = StreamTableEnvironment.create(env) //3.构建数据源 tableEnv.connect(new FileSystem().path(" ./data/sensor.txt")) .withFormat(new Csv()) .withSchema(new Schema() .field("id", DataTypes. STRING()) .field("timestamp", DataTypes.INT() ) .field("温度", DataTypes.DOUBLE()) ).createTemporaryTable("sensor") // 转换为表 val tableSensor = tableEnv.from( "sensor")// 床架转换对象 val code = new HashCode() // 使用tableAPI进行测试 val tableRes = tableSensor.select('id, code('id)) tableEnv.registerFunction("code",code) // 注册udf val tableSql = tableEnv.sqlQuery( """ |select |id, |code(id) |sensor |""".stripMargin) // 输出 tableRes.toAppendStream[R ow].print("tableAPI" ) tableSql.toAppendStream[ Row].print("tableSql") env.execute("FlinkSqlUdfHashCode") } class HashCode() extends ScalarFunction { def eval(s: String): String = { s.hashCode.toString } } } 运行results
2.3 表函数
与用户定义的标量函数类似,用户定义的表函数可以取 0、1 或多个标量值作为输入参数;
与标量函数不同,它可以返回任意数量的行作为输出,而不是单个值。为了定义一个表函数,你必须扩展 org.apache.flink.table.functions 中的基类 TableFunction 并实现(一个或多个)评估方法。表函数的行为由其求值方法决定,该方法必须是公共的且名为 eval。评估方法的参数类型决定了所有的val表函数的id参数。
返回的表的类型由TableFunction的泛型类型决定。评估方法使用受保护的collect(T)方法发出输出行。
在Table API中,Table函数需要与.joinLateral或.leftOuterJoinLateral一起使用。
joinLateral 运算符会将外部表中的每一行与表函数(TableFunction,运算符的参数为其表达式)计算出的所有行进行连接
leftOuterJoinLateral 运算符是左外连接,它将还将外部表中的每一行与表函数计算生成的所有行连接起来;并且,表函数返回空。表的外部行也应该保留。
在 SQL 中,您需要使用 Lateral Table(),或者使用 ON TRUE 条件的左连接。
在下面的代码中,我们将定义一个表函数,将其注册到表环境中,并在查询中调用它。
数据准备
hello|word,hello|spark hello|Flink,hello|java,hello|大数据兄弟
编写代码强>
package udf import org.apache.flink.streaming.api.scala._ import org.apache.flink.table.api.scala._ import org. apache.flink.table.functions.TableFunction import org.apache.flink.types.Row /** * @Package udf * @File: FlinkSqlUDFTableFunction.java * @author 大数据兄弟 * @date 2023/12/29 23: 10 * @version V1.0 */ object FlinkSqlUDFTableFunction { def main(args: Array[String]): Unit = { //1.构建运行环境 val env = StreamExecutionEnvironment.getExecutionEnvironment env.setParallelism(1) //设置并行度为1 //2.构建 TableEnv val tableEnv = StreamTableEnvironment.create(env) //3.构建数据源 val data = env.readTextFile("./data/words.txt") // 解析数据 val wordData: DataStream[String] = data.flatMap(_.split(",")) // 类型转换 val tableWord = tableEnv.fromDataStream(wordData,'id) // 调用TableFunction val split = new Split() // 表 API 方法 1 val resTable1 = tableWord. joinLateral(split('id) as('word,'length)) .select('id,'word,'length) // Table API 方法 2 val resTable2 = tableWord. leftOuterJoinLateral(split('id) as('word,'length)) .select('id ,'word,'length ) // 将数据注册到表中 tableEnv.createTemporaryView("sensor",tableWord) tableEnv.registerFunction(" split",split) // SQL 方法 1 val tableSQL1 = tableEnv.sqlQuery( """ | select |id,|word, |length |from |sensor,LATERAL TABLE(split(id)) AS newsensor(word, length) |""".stripMargin) // SQL 方法 2 val TableSQL2 = tableEnv.sqlQuery( """ id, sp;左连接横向T ABLE(split(id)) AS newsensor(word, length) ON TRUE |"""。 stripMargin) // 调用数据 resTable1.toAppendStream[Row].print("resTable1") resTable2.toAppendStream[Row ].print("resTable2") tableSQL1.toAppendStream[Row].print("tableSQL1") TableSQL2.toAppendStream[Row].print("TableSQL2") env.execute("FlinkSqlUDFTableFunction") } class Split() 扩展TableFunction[ (String,Int)] { def eval(str: String): Unit = str.split("\\|").foreach( word =>collect((word, word.length)) ) } } }< /pre>2.4 聚合函数(Aggregate Functions)
用户定义的聚合函数(User-Defined Aggregate Functions,UDAGGs)可以将表中的数据聚合成标量价值。用户定义的聚合函数是通过继承AggregateFunction抽象类来实现的。
聚合的示例如上图所示。
假设现在有一个包含各种饮料数据的表。该表由三列(id、名称和价格)和五行数据组成。现在我们需要找到表中所有饮料的最高价格,即进行max()聚合,结果为wi将是一个数值。 AggregateFunction的工作原理如下:
首先,它需要一个累加器来保存聚合中间结果的数据结构(状态)。可以通过调用 AggregateFunction 的 createAccumulator() 方法来创建一个空累加器。
然后为每个输入行调用该函数的accumulate() 方法来更新累加器。
处理完所有行后,调用函数的 getValue() 方法计算并返回最终结果。 AggregationFunction需要必须实现的方法:
除了上述方法之外,还有一些可选的方法。其中一些方法可以使系统更高效地执行查询,而其他方法则是某些场景所必需的。例如,如果聚合函数应用于会话组窗口的上下文中,则需要 merge() 方法。
retract()
merge()
resetAccumulator()
接下来我们编写一个自定义AggregateFunction,计算每个价格的平均值。
数据准备
1、拿铁、6 2、牛奶、3 3、Breve、5 4、摩卡、8 5、茶、4 pre>代码如下
package udf import org.apache.calcite.rel.`type`.{RelDataType, RelDataTypeFactory} import org.apache.flink.streaming.api.scala. _ 导入 org.apache.flink.table.api.DataTypes 导入 org.apache.flink.table.api.scala._ 导入 org.apache.flink.table.descriptors.{Csv,文件系统,模式} 导入 org.apache。 flink.table .functions.AggregateFunction import org.apache.flink.types.Row import java.util /** * @Package udf * @File: FlinkSQUDFAggregateFunction.java * @author 大数据兄弟 * @date 2023/12/30 22 :06 * @version V1.0 */ object FlinkSQUDFAggregateFunction { def main(args: Array[String]): Unit = { //1.构建运行环境 val env = StreamExecutionEnvironment.getExecutionEnvironment env.setParallelism(1) // 设置并行度为1 //2。构建 TableEnv val tableEnv = StreamTableEnvironment.create(env ) //3.构建数据源 tableEnv.connect(new FileSystem().path("./data/datas")) .withFormat(new Csv) .withSchema(new Schema() .field("id", DataTypes.STRING ()) 。 field("name", DataTypes.STRING()) .field("price", DataTypes.DOUBLE()) ).createTemporaryTable("datas") val AvgTemp = new AvgTemp() val table = tableEnv.from("datas" ) val resTableApi = table.groupBy('id) .aggregate(AvgTemp('price) as 'sumprice) 。 select('id, 'sumprice) tableEnv.registerFunction("avgTemp",AvgTemp) val tablesql = tableEnv.sqlQuery( """ |select |id,avgTemp(price) |from data group by id |""".stripMargin) resTableApi.toRetractStream[Row].print("resTableApi") tablesql.toRetractStream[Row].print("tablesql") env.execute("FlinkSQUDFAggregateFunction") } class AvgTempAcc { var Sum: Double = 0.0 var count: Int = 0 } class AvgTemp extends AggregateFunction[Double, AvgTempAcc] { override def getValue(acc: AvgTempAcc): Double = { acc.sum / acc.count } override def createAccumulator(): AvgTempAcc = new AvgTempAcc( ) } defaccumulate(accumulator: AvgTempAcc,price:Double): Unit = {accumulator.sum += 价格accumulator.count += 1 } }2.5 表聚合函数
用户定义的表聚合函数(UDTAGG)可以将表中的数据聚合成多行多列的结果表。这与 AggregateFunction 非常相似,只不过聚合结果以前是标量值,现在是表。
例如,现在我们需要找到表中所有饮料价格最高的前2个,即进行top2()表聚合。我们需要检查 5 行中的每一行,结果将是一个表格,其中前 2 个值已排序。用户自定义的表聚合函数是通过继承TableAggregateFunction抽象类来实现的。 TableAggregateFunction的工作原理如下:
首先,它还需要一个累加器(Accumulator),它是一个数据st存储聚合中间结果的结构。可以通过调用 TableAggregateFunction 的 createAccumulator() 方法来创建一个空累加器。
然后,为每个输入行调用函数的accumulate()方法来更新累加器。
处理完所有行后,将调用函数的emitValue()方法计算并返回最终结果。除了上述方法之外,还有一些替代的实现方法。
retract()
merge()
resetAccumulator()
emitValue()
emitUpdateWithRetract()
接下来我们编写一个自定义 TableAggregateFunction 来提取每个价格的两个最高平均值。
数据准备
1、拿铁、6 2、牛奶、3 3、Breve、5 4、摩卡、8 5、茶、4 pre>代码如下
package udf import org.apache.flink.streaming.api.scala._ import org.apache.flink.table.api.DataTypes import org.apache. flink.table.api.scala._导入org.apache。flink.table.descriptors.{Csv,文件系统,模式}导入org.apache.flink.table.functions.TableAggregateFunction导入org.apache.flink.types.Row导入org.apache.flink.util.Collector导入udf.FlinkSQUDFAggregateFunction。 AvgTemp /** * @Package udf * @File: FlinkSqlUDFTableAggregateFunction.java * @author 大数据兄弟 * @date 2023/12/30 22:53 * @version V1.0 */ object FlinkSqlUDFTableAggregateFunction { def main(args: Array[字符串]): 单位 = { //1.构建运行环境 val env = StreamExecutionEnvironment.getExecutionEnvironment env.setParallelism(1) // 设置并行度为1 //2.构建 TableEnv val tableEnv = StreamTableEnvironment.create(env) //3.构建数据源 tableEnv.connect(new FileSystem().path("./data/datas")) .withFormat(new Csv) .withSchema(new Schema() .field( "id", DataTypes.STRING()) .field("name", DataTypes.STRING()) .field("price", DataTypes.DOUBLE()) ).createTemporaryTable("datas") val table = tableEnv.from("datas") val temp = new Top2Temp () 弗吉尼亚州l tableApi = table.groupBy('id) .flatAggregate (temp('price) as('tmpprice, 'rank)) .select('id, 'tmpprice, 'rank) tableEnv.registerFunction("temp",temp) tableApi .toRetractStream[Row].print() env.execute ("FlinkSqlUDFTableAggregateFunction") } class Top2TempAcc { var HighestPrice: Double = Int.MinValue var secodeHighestPrice: Double = Int.MinValue } class Top2Temp 扩展 TableAggregateFunction[(Double, Int), Top2TempAcc ] { override def createAccumulator(): Top2TempAcc = new Top2TempAcc defaccumulate(acc: Top2TempAcc, temp: Double): Unit = { if (temp > acc.highestPrice) { acc.secodeHighestPrice = acc.highestPrice acc.highestPrice = temp } else if (temp > acc.secodeHighestPrice) { acc.highestPrice = temp } } def emitValue(acc: Top2TempAcc, out: Collector[(Double, Int)]): Unit = { out.collect(acc.highestPrice, 1) out. collect(acc.secodeHighestPrice, 2) } } }这里介绍“如何使用FlinkSQL内置函数”。感谢您的阅读。如果哟想了解更多行业资讯,可以关注网站,小编将为大家输出更多优质实用文章!
2. 本站积分货币获取途径以及用途的解读,想在本站混的好,请务必认真阅读!
3. 本站强烈打击盗版/破解等有损他人权益和违法作为,请各位会员支持正版!
4. 编程技术 > 如何使用FlinkSQL内置函数