共計 10823 個字符，預(yù)計需要花費 28 分鐘才能閱讀完成。
本篇內(nèi)容介紹了“如何使用 FlinkSQL 內(nèi)置函數(shù)”的有關(guān)知識，在實際案例的操作過程中，不少人都會遇到這樣的困境，接下來就讓丸趣 TV 小編帶領(lǐng)大家學(xué)習(xí)一下如何處理這些情況吧！希望大家仔細(xì)閱讀，能夠?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)介紹。
類型 TableApiSQLAPI 比較函數(shù) ANY1 === ANY2value1 = value2 比較函數(shù) NY1 ANY2value1 value2 邏輯函數(shù) BOOLEAN1 || BOOLEAN2boolean1 OR boolean2 邏輯函數(shù) BOOLEAN.isFalseboolean IS FALSE 邏輯函數(shù)!BOOLEANNOT boolean 算術(shù)函數(shù) NUMERIC1 + NUMERIC2numeric1 + numeric2 算術(shù)函數(shù) NUMERIC1.power(NUMERIC2)POWER(numeric1, numeric2)字符串函數(shù) STRING1 + STRING2string1 || string2 字符串函數(shù) STRING.upperCase()UPPER(string)字符串函數(shù) STRING.charLength()CHAR_LENGTH(string)時間函數(shù) STRING.toDateDATE string 時間函數(shù) STRING.toTimestampTIMESTAMP string 時間函數(shù) currentTime()CURRENT_TIME 時間函數(shù) NUMERIC.daysINTERVAL string range 時間函數(shù) NUMERIC.minutes 聚合函數(shù) FIELD.countCOUNT(*)聚合函數(shù) FIELD.sum0SUM([ALL | DISTINCT] expression)聚合函數(shù) RANK()聚合函數(shù) ROW_NUMBER()
二、Flink UDF
用戶定義函數(shù) (User-defined Functions，UDF) 是一個重要的特性，因為它們顯著地擴展了查詢 (Query) 的表達(dá)能力。一些系統(tǒng)內(nèi)置函數(shù)無法解決的需求，我們可以用 UDF 來自定義實現(xiàn)。
2.1 注冊用戶自定義函數(shù) UDF
在大多數(shù)情況下，用戶定義的函數(shù)必須先注冊，然后才能在查詢中使用。不需要專門為 Scala 的 Table API 注冊函數(shù)。
函數(shù)通過調(diào)用 registerFunction()方法在 TableEnvironment 中注冊。當(dāng)用戶定義的函數(shù)被注冊時，它被插入到 TableEnvironment 的函數(shù)目錄中，這樣 Table API 或 SQL 解析器就可以識別并正確地解釋它。
2.2 標(biāo)量函數(shù)(Scalar Functions)
用戶定義的標(biāo)量函數(shù)，可以將 0、1 或多個標(biāo)量值，映射到新的標(biāo)量值。
為了定義標(biāo)量函數(shù)，必須在 org.apache.flink.table.functions 中擴展基類 Scalar Function，并實現(xiàn) (一個或多個) 求值 (evaluation，eval) 方法。標(biāo)量函數(shù)的行為由求值方法決定，求值方法必須公開聲明并命名為 eval(直接 def 聲明，沒有 override)。求值方法的參數(shù)類型和返回類型，確定了標(biāo)量函數(shù)的參數(shù)和返回類型。
在下面的代碼中，我們定義自己的 HashCode 函數(shù)，在 TableEnvironment 中注冊它，并在查詢中調(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)建運行環(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)換對象 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)
與用戶定義的標(biāo)量函數(shù)類似，用戶定義的表函數(shù)，可以將 0、1 或多個標(biāo)量值作為輸入?yún)?shù);
與標(biāo)量函數(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ù)是它的表達(dá)式) 計算得到的所有行連接起來。
而 leftOuterJoinLateral 算子，則是左外連接，它同樣會將外部表中的每一行與表函數(shù)計算生成的所有行連接起來; 并且，對于表函數(shù)返回的是空表的外部行，也要保留下來。
在 SQL 中，則需要使用 Lateral Table()，或者帶有 ON TRUE 條件的左連接。
下面的代碼中，我們將定義一個表函數(shù)，在表環(huán)境中注冊它，并在查詢中調(diào)用它。
數(shù)據(jù)準(zhǔn)備
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) // 設(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ù)注冊成表 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ù)，聚合成一個標(biāo)量值。用戶定義的聚合函數(shù)，是通過繼承 AggregateFunction 抽象類實現(xiàn)的。
上圖中顯示了一個聚合的例子。
假設(shè)現(xiàn)在有一張表，包含了各種飲料的數(shù)據(jù)。該表由三列 (id、name 和 price)、五行組成數(shù)據(jù)。現(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ù)應(yīng)用在會話窗口 (session group window) 上下文中，則 merge()方法是必需的。
retract()
merge()
resetAccumulator()
接下來我們寫一個自定義 AggregateFunction, 計算一個每個 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)建運行環(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)，可以把一個表中數(shù)據(jù)，聚合為具有多行和多列的結(jié)果表。這跟 AggregateFunction 非常類似，只是之前聚合結(jié)果是一個標(biāo)量值，現(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ù)準(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)建運行環(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) } } }“如何使用 FlinkSQL 內(nèi)置函數(shù)”的內(nèi)容就介紹到這里了，感謝大家的閱讀。如果想了解更多行業(yè)相關(guān)的知識可以關(guān)注丸趣 TV 網(wǎng)站，丸趣 TV 小編將為大家輸出更多高質(zhì)量的實用文章！
