阅读 111

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

前言

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

一、系统内置函数

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

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

类型TableApiSQLAPI比较函数ANY1 === ANY2value1 = value2比较函数NY1 > ANY2value1 > value2逻辑函数BOOLEAN1 || BOOLEAN2boolean1 OR boolean2逻辑函数BOOLEAN.isFalseboolean IS FALSE逻辑函数!BOOLEANNOT boolean算术函数NUMERIC1 + NUMERIC2numeric1 + numeric2算术函数NUMERIC1.power(NUMERIC2)POWER(numeric1, numeric2)字符串函数STRING1 + STRING2string1 || string2字符串函数STRING.upperCase()UPPER(string)字符串函数STRING.charLength()CHAR_LENGTH(string)时间函数STRING.toDateDATE string时间函数STRING.toTimestampTIMESTAMP string时间函数currentTime()CURRENT_TIME时间函数NUMERIC.daysINTERVAL string range时间函数NUMERIC.minutes聚合函数FIELD.countCOUNT(*)聚合函数FIELD.sum0SUM([ 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.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 udfimport org.apache.flink.streaming.api.scala._import org.apache.flink.table.api.DataTypesimport org.apache.flink.table.api.scala._import org.apache.flink.table.descriptors.{Csv, FileSystem, Schema}import org.apache.flink.table.functions.ScalarFunctionimport org.apache.flink.types.Row/*** @Package udf* @File ：FlinkSqlUdfHashCode.java* @author 大数据老哥* @date 2020/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("temperature", 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)       |from       |sensor       |""".stripMargin)   // 输出   tableRes.toAppendStream[Row].print("tableAPI")   tableSql.toAppendStream[Row].print("tableSql")   env.execute("FlinkSqlUdfHashCode") } class HashCode() extends ScalarFunction {   def eval(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|大数据老哥

编写代码

package udfimport org.apache.flink.streaming.api.scala._import org.apache.flink.table.api.scala._import org.apache.flink.table.functions.TableFunctionimport org.apache.flink.types.Row/** * @Package udf * @File ：FlinkSqlUDFTableFunction.java * @author 大数据老哥 * @date 2020/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()    // Table API 方式一    val resTable1 = tableWord.      joinLateral(split('id) as('word,'length))      .select('id,'word,'length )    //  Table API  方式二    val resTable2 = tableWord.      leftOuterJoinLateral(split('id) as('word,'length))      .select('id,'word,'length )    // 将数据注册成表     tableEnv.createTemporaryView("sensor",tableWord)     tableEnv.registerFunction("split",split)    // SQL 方式一    val tableSQL1 = tableEnv.sqlQuery(      """        |select        |id,        |word,        |length        |from        |sensor ,LATERAL TABLE(split(id)) AS newsensor(word, length)        |""".stripMargin)    //  SQL 方式二    val TableSQL2 = tableEnv.sqlQuery(      """        |select        |id,        |word,        |length        |from        |sensor        | LEFT JOIN LATERAL TABLE(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() extends TableFunction[(String,Int)] {    def eval(str: String): Unit = {      str.split("\\|").foreach(        word => collect((word, word.length))      )    }  }}

2.4 聚合函数（Aggregate Functions)

       假设现在有一张表，包含了各种饮料的数据。该表由三列（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

代码如下

package udfimport org.apache.calcite.rel.`type`.{RelDataType, RelDataTypeFactory}import org.apache.flink.streaming.api.scala._import org.apache.flink.table.api.DataTypesimport org.apache.flink.table.api.scala._import org.apache.flink.table.descriptors.{Csv, FileSystem, Schema}import org.apache.flink.table.functions.AggregateFunctionimport org.apache.flink.types.Rowimport java.util/** * @Package udf * @File ：FlinkSQUDFAggregateFunction.java * @author 大数据老哥 * @date 2020/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 datas 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()  }  def accumulate(accumulator: AvgTempAcc, price: Double): Unit = {    accumulator.sum += price    accumulator.count += 1  }}

2.5表聚合函数（Table Aggregate Functions）

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

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

  • 为处理完所有行后，将调用函数的 emitValue（）方法来计算并返回最终结果。除了上述方法之外，还有一些可选择实现的方法。

  • retract()

  • merge()

  • resetAccumulator()

  • emitValue()

  • emitUpdateWithRetract()

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

数据准备

1,Latte,62,Milk,33,Breve,54,Mocha,85,Tea,4

代码如下

package udfimport org.apache.flink.streaming.api.scala._import org.apache.flink.table.api.DataTypesimport org.apache.flink.table.api.scala._import org.apache.flink.table.descriptors.{Csv, FileSystem, Schema}import org.apache.flink.table.functions.TableAggregateFunctionimport org.apache.flink.types.Rowimport org.apache.flink.util.Collectorimport udf.FlinkSQUDFAggregateFunction.AvgTemp/** * @Package udf * @File ：FlinkSqlUDFTableAggregateFunction.java * @author 大数据老哥 * @date 2020/12/30 22:53 * @version V1.0 */object FlinkSqlUDFTableAggregateFunction {  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 table = tableEnv.from("datas")    val temp = new Top2Temp()    val 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 extends TableAggregateFunction[(Double, Int), Top2TempAcc] {    override def createAccumulator(): Top2TempAcc = new Top2TempAcc    def accumulate(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)    }  }}

总结

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



作者：大数据老哥
链接：https://www.jianshu.com/p/55db5ed7b5eb

文章分类
后端
文章标签
flinksql
函数
版权声明：本站是系统测试站点，无实际运营。本文内容由互联网用户自发贡献，该文观点仅代表作者本人。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。如发现本站有涉嫌抄袭侵权/违法违规的内容， 请发送邮件至 XXXXXXo@163.com 举报，一经查实，本站将立刻删除。
相关推荐