如何使用FlinkSQL內(nèi)置函數(shù)

本篇內(nèi)容介紹了“如何使用FlinkSQL內(nèi)置函數(shù)”的有關(guān)知識，在實際案例的操作過程中，不少人都會遇到這樣的困境，接下來就讓小編帶領(lǐng)大家學習一下如何處理這些情況吧！希望大家仔細閱讀，能夠?qū)W有所成！

前言

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

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

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

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

二、Flink UDF

用戶定義函數(shù)(User-defined  Functions，UDF)是一個重要的特性，因為它們顯著地擴展了查詢(Query)的表達能力。一些系統(tǒng)內(nèi)置函數(shù)無法解決的需求，我們可以用 UDF  來自定義實現(xiàn)。

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

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

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

2.2 標量函數(shù)(Scalar Functions)

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

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

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

準備數(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)建運行環(huán)境    val env = StreamExecutionEnvironment.getExecutionEnvironment    env.setParallelism(1) // 設置并行度為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)換對象    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    }  } } 運行結(jié)果

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

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

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

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

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

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

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

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

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

數(shù)據(jù)準備

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

編寫代碼

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)建運行環(huán)境     val env = StreamExecutionEnvironment.getExecutionEnvironment     env.setParallelism(1) // 設置并行度為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ù)注冊成表      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)可以把一個表中的數(shù)據(jù)，聚合成一個標量值。用戶定義的聚合函數(shù)，是通過繼承 AggregateFunction  抽象類實現(xiàn)的。

上圖中顯示了一個聚合的例子。

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

• 首先，它需要一個累加器，用來保存聚合中間結(jié)果的數(shù)據(jù)結(jié)構(gòu)(狀態(tài))?？梢酝ㄟ^調(diào)用 AggregateFunction 的  createAccumulator()方法創(chuàng)建空累加器。

• 隨后，對每個輸入行調(diào)用函數(shù)的 accumulate() 方法來更新累加器。

• 處理完所有行后，將調(diào)用函數(shù)的 getValue() 方法來計算并返回最終結(jié)果。AggregationFunction 要求必須實現(xiàn)的方法：

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

• retract()

• merge()

• resetAccumulator()

接下來我們寫一個自定義AggregateFunction,計算一個每個price的平均值。

數(shù)據(jù)準備

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)建運行環(huán)境     val env = StreamExecutionEnvironment.getExecutionEnvironment     env.setParallelism(1) // 設置并行度為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)，可以把一個表中數(shù)據(jù)，聚合為具有多行和多列的結(jié)果表。這跟 AggregateFunction  非常類似，只是之前聚合結(jié)果是一個標量值，現(xiàn)在變成了一張表。

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

• 為首先，它同樣需要一個累加器(Accumulator)，它是保存聚合中間結(jié)果的數(shù)據(jù)結(jié)構(gòu)。通過調(diào)用 TableAggregateFunction 的  createAccumulator()方法可以創(chuàng)建空累加器。

• 為隨后，對每個輸入行調(diào)用函數(shù)的 accumulate()方法來更新累加器。

• 為處理完所有行后，將調(diào)用函數(shù)的 emitValue()方法來計算并返回最終結(jié)果。除了上述方法之外，還有一些可選擇實現(xiàn)的方法。

• retract()

• merge()

• resetAccumulator()

• emitValue()

• emitUpdateWithRetract()

接下來我們寫一個自定義 TableAggregateFunction，用來提取每個 price 最高的兩個平均值。

數(shù)據(jù)準備

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)建運行環(huán)境     val env = StreamExecutionEnvironment.getExecutionEnvironment     env.setParallelism(1) // 設置并行度為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)     }   }  }

“如何使用FlinkSQL內(nèi)置函數(shù)”的內(nèi)容就介紹到這里了，感謝大家的閱讀。如果想了解更多行業(yè)相關(guān)的知識可以關(guān)注億速云網(wǎng)站，小編將為大家輸出更多高質(zhì)量的實用文章！