[[361272]]

前言

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.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 大數據老哥 
* @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|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 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)

用戶自定義聚合函數(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 

代碼如下

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 大數據老哥 
 * @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)

戶定義的表聚合函數(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 

代碼如下

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 大數據老哥 
 * @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) 
    } 
  } 
 
} 

總結

好了今天的內容就分享到這里了。上述主要講解了一個系統自己帶的函數，但是往往企業中不光只需要這些函數，有好多需求是本身函數是無法完成的。這時候就要用到我們的自定義函數了。他可以根據我們自己的需要進行編寫代碼來實現我們想要的功能。

本文轉載自微信公眾號「大數據老哥」，可以通過以下二維碼關注。轉載本文請聯系大數據老哥公眾號。