[[361272]]

前言

Flink Table 和 SQL 內(nèi)置了很多 SQL 中支持的函數(shù);如果有無(wú)法滿足的需要，則可以實(shí)現(xiàn)用戶自定義的函數(shù)(UDF)來(lái)解決。

一、系統(tǒng)內(nèi)置函數(shù)

Flink Table API 和 SQL 為用戶提供了一組用于數(shù)據(jù)轉(zhuǎn)換的內(nèi)置函數(shù)。SQL 中支持的很多函數(shù)，Table API 和 SQL 都已經(jīng)做了實(shí)現(xiàn)，其它還在快速開(kāi)發(fā)擴(kuò)展中。

以下是一些典型函數(shù)的舉例，全部的內(nèi)置函數(shù)，可以參考官網(wǎng)介紹。

二、Flink UDF

用戶定義函數(shù)(User-defined Functions，UDF)是一個(gè)重要的特性，因?yàn)樗鼈冿@著地?cái)U(kuò)展了查詢(Query)的表達(dá)能力。一些系統(tǒng)內(nèi)置函數(shù)無(wú)法解決的需求，我們可以用 UDF 來(lái)自定義實(shí)現(xiàn)。

2.1 注冊(cè)用戶自定義函數(shù) UDF

在大多數(shù)情況下，用戶定義的函數(shù)必須先注冊(cè)，然后才能在查詢中使用。不需要專門(mén)為Scala 的 Table API 注冊(cè)函數(shù)。

函數(shù)通過(guò)調(diào)用 registerFunction()方法在 TableEnvironment 中注冊(cè)。當(dāng)用戶定義的函數(shù)被注冊(cè)時(shí)，它被插入到 TableEnvironment 的函數(shù)目錄中，這樣 Table API 或 SQL 解析器就可以識(shí)別并正確地解釋它。

2.2 標(biāo)量函數(shù)(Scalar Functions)

用戶定義的標(biāo)量函數(shù)，可以將 0、1 或多個(gè)標(biāo)量值，映射到新的標(biāo)量值。

為了定義標(biāo)量函數(shù)，必須在 org.apache.flink.table.functions 中擴(kuò)展基類 Scalar Function，并實(shí)現(xiàn)(一個(gè)或多個(gè))求值(evaluation，eval)方法。標(biāo)量函數(shù)的行為由求值方法決定，求值方法必須公開(kāi)聲明并命名為 eval(直接 def 聲明，沒(méi)有 override)。求值方法的參數(shù)類型和返回類型，確定了標(biāo)量函數(shù)的參數(shù)和返回類型。

在下面的代碼中，我們定義自己的 HashCode 函數(shù)，在 TableEnvironment 中注冊(cè)它，并在查詢中調(diào)用它。

準(zhǔn)備數(shù)據(jù)

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 

代碼如下

package udf 
 
import org.apache.flink.streaming.api.scala._ 
import org.apache.flink.table.api.DataTypes 
import org.apache.flink.table.api.scala._ 
import org.apache.flink.table.descriptors.{Csv, FileSystem, Schema} 
import org.apache.flink.table.functions.ScalarFunction 
import org.apache.flink.types.Row 
 
/** 
* @Package udf 
* @File ：FlinkSqlUdfHashCode.java 
* @author 大數(shù)據(jù)老哥 
* @date 2020/12/29 21:58 
* @version V1.0 
*/ 
object FlinkSqlUdfHashCode { 
 def main(args: Array[String]): Unit = { 
   //1.構(gòu)建運(yùn)行環(huán)境 
   val env = StreamExecutionEnvironment.getExecutionEnvironment 
   env.setParallelism(1) // 設(shè)置并行度為1 
   //2.構(gòu)建TableEnv 
   val tableEnv = StreamTableEnvironment.create(env) 
   //3.構(gòu)建數(shù)據(jù)源 
   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") 
   // 轉(zhuǎn)為表 
   val tableSensor = tableEnv.from("sensor") 
   // 床架轉(zhuǎn)換對(duì)象 
   val code = new HashCode() 
   //使用tableAPI 進(jìn)行測(cè)試 
   val tableRes = tableSensor.select('id, code('id)) 
   tableEnv.registerFunction("code",code) // 注冊(cè)u(píng)df 
   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 
   } 
 } 
} 
運(yùn)行結(jié)果 

2.3 表函數(shù)(Table Functions)

與用戶定義的標(biāo)量函數(shù)類似，用戶定義的表函數(shù)，可以將 0、1 或多個(gè)標(biāo)量值作為輸入?yún)?shù);

與標(biāo)量函數(shù)不同的是，它可以返回任意數(shù)量的行作為輸出，而不是單個(gè)值。為了定義一個(gè)表函數(shù)，必須擴(kuò)展 org.apache.flink.table.functions 中的基類 TableFunction并實(shí)現(xiàn)(一個(gè)或多個(gè))求值方法。表函數(shù)的行為由其求值方法決定，求值方法必須是 public的，并命名為 eval。求值方法的參數(shù)類型，決定表函數(shù)的所有有效參數(shù)。

返回表的類型由 TableFunction 的泛型類型確定。求值方法使用 protected collect(T)方法發(fā)出輸出行。

在 Table API 中，Table 函數(shù)需要與.joinLateral 或.leftOuterJoinLateral 一起使用。

joinLateral 算子，會(huì)將外部表中的每一行，與表函數(shù)(TableFunction，算子的參數(shù)是它的表達(dá)式)計(jì)算得到的所有行連接起來(lái)。

而 leftOuterJoinLateral 算子，則是左外連接，它同樣會(huì)將外部表中的每一行與表函數(shù)計(jì)算生成的所有行連接起來(lái);并且，對(duì)于表函數(shù)返回的是空表的外部行，也要保留下來(lái)。

在 SQL 中，則需要使用 Lateral Table()，或者帶有 ON TRUE 條件的左連接。

下面的代碼中，我們將定義一個(gè)表函數(shù)，在表環(huán)境中注冊(cè)它，并在查詢中調(diào)用它。

數(shù)據(jù)準(zhǔn)備

hello|word,hello|spark 
hello|Flink,hello|java,hello|大數(shù)據(jù)老哥 

編寫(xiě)代碼

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 大數(shù)據(jù)老哥 
 * @date 2020/12/29 23:10 
 * @version V1.0 
 */ 
object FlinkSqlUDFTableFunction { 
  def main(args: Array[String]): Unit = { 
    //1.構(gòu)建運(yùn)行環(huán)境 
    val env = StreamExecutionEnvironment.getExecutionEnvironment 
    env.setParallelism(1) // 設(shè)置并行度為1 
    //2.構(gòu)建TableEnv 
    val tableEnv = StreamTableEnvironment.create(env) 
    //3.構(gòu)建數(shù)據(jù)源 
    val data = env.readTextFile("./data/words.txt") 
    // 解析數(shù)據(jù) 
    val wordData: DataStream[String] = data.flatMap(_.split(",")) 
    // 類型轉(zhuǎn)換 
    val tableWord = tableEnv.fromDataStream(wordData,'id) 
    // 調(diào)用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 ) 
    // 將數(shù)據(jù)注冊(cè)成表 
     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) 
    // 調(diào)用數(shù)據(jù) 
    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 聚合函數(shù)(Aggregate Functions)

用戶自定義聚合函數(shù)(User-Defined Aggregate Functions，UDAGGs)可以把一個(gè)表中的數(shù)據(jù)，聚合成一個(gè)標(biāo)量值。用戶定義的聚合函數(shù)，是通過(guò)繼承 AggregateFunction 抽象類實(shí)現(xiàn)的。

上圖中顯示了一個(gè)聚合的例子。

假設(shè)現(xiàn)在有一張表，包含了各種飲料的數(shù)據(jù)。該表由三列(id、name 和 price)、五行組成數(shù)據(jù)?，F(xiàn)在我們需要找到表中所有飲料的最高價(jià)格，即執(zhí)行 max()聚合，結(jié)果將是一個(gè)數(shù)值。AggregateFunction 的工作原理如下：

• 首先，它需要一個(gè)累加器，用來(lái)保存聚合中間結(jié)果的數(shù)據(jù)結(jié)構(gòu)(狀態(tài))。可以通過(guò)調(diào)用 AggregateFunction 的 createAccumulator()方法創(chuàng)建空累加器。
• 隨后，對(duì)每個(gè)輸入行調(diào)用函數(shù)的 accumulate() 方法來(lái)更新累加器。
• 處理完所有行后，將調(diào)用函數(shù)的 getValue() 方法來(lái)計(jì)算并返回最終結(jié)果。AggregationFunction 要求必須實(shí)現(xiàn)的方法：

除了上述方法之外，還有一些可選擇實(shí)現(xiàn)的方法。其中一些方法，可以讓系統(tǒng)執(zhí)行查詢更有效率，而另一些方法，對(duì)于某些場(chǎng)景是必需的。例如，如果聚合函數(shù)應(yīng)用在會(huì)話窗口(session group window)上下文中，則 merge()方法是必需的。

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

接下來(lái)我們寫(xiě)一個(gè)自定義AggregateFunction,計(jì)算一個(gè)每個(gè)price的平均值。

數(shù)據(jù)準(zhǔn)備

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

代碼如下

package udf 
 
import org.apache.calcite.rel.`type`.{RelDataType, RelDataTypeFactory} 
import org.apache.flink.streaming.api.scala._ 
import org.apache.flink.table.api.DataTypes 
import org.apache.flink.table.api.scala._ 
import org.apache.flink.table.descriptors.{Csv, FileSystem, Schema} 
import org.apache.flink.table.functions.AggregateFunction 
import org.apache.flink.types.Row 
 
import java.util 
 
/** 
 * @Package udf 
 * @File ：FlinkSQUDFAggregateFunction.java 
 * @author 大數(shù)據(jù)老哥 
 * @date 2020/12/30 22:06 
 * @version V1.0 
 */ 
object FlinkSQUDFAggregateFunction { 
  def main(args: Array[String]): Unit = { 
    //1.構(gòu)建運(yùn)行環(huán)境 
    val env = StreamExecutionEnvironment.getExecutionEnvironment 
    env.setParallelism(1) // 設(shè)置并行度為1 
    //2.構(gòu)建TableEnv 
    val tableEnv = StreamTableEnvironment.create(env) 
    //3.構(gòu)建數(shù)據(jù)源 
    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表聚合函數(shù)(Table Aggregate Functions)

戶定義的表聚合函數(shù)(User-Defined Table Aggregate Functions，UDTAGGs)，可以把一個(gè)表中數(shù)據(jù)，聚合為具有多行和多列的結(jié)果表。這跟 AggregateFunction 非常類似，只是之前聚合結(jié)果是一個(gè)標(biāo)量值，現(xiàn)在變成了一張表。

比如現(xiàn)在我們需要找到表中所有飲料的前 2 個(gè)最高價(jià)格，即執(zhí)行 top2()表聚合。我們需要檢查 5 行中的每一行，得到的結(jié)果將是一個(gè)具有排序后前 2 個(gè)值的表。用戶定義的表聚合函數(shù)，是通過(guò)繼承 TableAggregateFunction 抽象類來(lái)實(shí)現(xiàn)的。TableAggregateFunction 的工作原理如下：

• 為首先，它同樣需要一個(gè)累加器(Accumulator)，它是保存聚合中間結(jié)果的數(shù)據(jù)結(jié)構(gòu)。通過(guò)調(diào)用 TableAggregateFunction 的 createAccumulator()方法可以創(chuàng)建空累加器。
• 為隨后，對(duì)每個(gè)輸入行調(diào)用函數(shù)的 accumulate()方法來(lái)更新累加器。
• 為處理完所有行后，將調(diào)用函數(shù)的 emitValue()方法來(lái)計(jì)算并返回最終結(jié)果。除了上述方法之外，還有一些可選擇實(shí)現(xiàn)的方法。
• retract()
• merge()
• resetAccumulator()
• emitValue()
• emitUpdateWithRetract()

接下來(lái)我們寫(xiě)一個(gè)自定義 TableAggregateFunction，用來(lái)提取每個(gè) price 最高的兩個(gè)平均值。

數(shù)據(jù)準(zhǔn)備

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

代碼如下

package udf 
 
import org.apache.flink.streaming.api.scala._ 
import org.apache.flink.table.api.DataTypes 
import org.apache.flink.table.api.scala._ 
import org.apache.flink.table.descriptors.{Csv, FileSystem, Schema} 
import org.apache.flink.table.functions.TableAggregateFunction 
import org.apache.flink.types.Row 
import org.apache.flink.util.Collector 
import udf.FlinkSQUDFAggregateFunction.AvgTemp 
 
/** 
 * @Package udf 
 * @File ：FlinkSqlUDFTableAggregateFunction.java 
 * @author 大數(shù)據(jù)老哥 
 * @date 2020/12/30 22:53 
 * @version V1.0 
 */ 
object FlinkSqlUDFTableAggregateFunction { 
  def main(args: Array[String]): Unit = { 
    //1.構(gòu)建運(yùn)行環(huán)境 
    val env = StreamExecutionEnvironment.getExecutionEnvironment 
    env.setParallelism(1) // 設(shè)置并行度為1 
    //2.構(gòu)建TableEnv 
    val tableEnv = StreamTableEnvironment.create(env) 
    //3.構(gòu)建數(shù)據(jù)源 
    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) 
    } 
  } 
 
} 

總結(jié)

好了今天的內(nèi)容就分享到這里了。上述主要講解了一個(gè)系統(tǒng)自己帶的函數(shù)，但是往往企業(yè)中不光只需要這些函數(shù)，有好多需求是本身函數(shù)是無(wú)法完成的。這時(shí)候就要用到我們的自定義函數(shù)了。他可以根據(jù)我們自己的需要進(jìn)行編寫(xiě)代碼來(lái)實(shí)現(xiàn)我們想要的功能。

本文轉(zhuǎn)載自微信公眾號(hào)「大數(shù)據(jù)老哥」，可以通過(guò)以下二維碼關(guān)注。轉(zhuǎn)載本文請(qǐng)聯(lián)系大數(shù)據(jù)老哥公眾號(hào)。