Flink重點(diǎn)難點(diǎn)：Flink Table&SQL必知必會(huì)(一)

什么是Table API和Flink SQL

Flink本身是批流統(tǒng)一的處理框架，所以Table API和SQL，就是批流統(tǒng)一的上層處理API。目前功能尚未完善，處于活躍的開發(fā)階段。

Table API是一套內(nèi)嵌在Java和Scala語言中的查詢API，它允許我們以非常直觀的方式，組合來自一些關(guān)系運(yùn)算符的查詢（比如select、filter和join）。而對(duì)于Flink SQL，就是直接可以在代碼中寫SQL，來實(shí)現(xiàn)一些查詢（Query）操作。Flink的SQL支持，基于實(shí)現(xiàn)了SQL標(biāo)準(zhǔn)的Apache Calcite（Apache開源SQL解析工具）。

無論輸入是批輸入還是流式輸入，在這兩套API中，指定的查詢都具有相同的語義，得到相同的結(jié)果。

需要引入的依賴

取決于你使用的編程語言，比如這里，我們選擇 Scala API 來構(gòu)建你的 Table API 和 SQL 程序：

<dependency>
  <groupId>org.apache.flink</groupId>
  <artifactId>flink-table-api-scala-bridge_2.11</artifactId>
  <version>1.11.0</version>
  <scope>provided</scope>
</dependency>

除此之外，如果你想在 IDE 本地運(yùn)行你的程序，你需要添加下面的模塊，具體用哪個(gè)取決于你使用哪個(gè) Planner，我們這里選擇使用 blink planner：

<dependency>
  <groupId>org.apache.flink</groupId>
  <artifactId>flink-table-planner-blink_2.11</artifactId>
  <version>1.11.0</version>
  <scope>provided</scope>
</dependency>

如果你想實(shí)現(xiàn)自定義格式來解析 Kafka 數(shù)據(jù)，或者自定義函數(shù)，使用下面的依賴：

<dependency>
  <groupId>org.apache.flink</groupId>
  <artifactId>flink-table-common</artifactId>
  <version>1.11.0</version>
  <scope>provided</scope>
</dependency>

• flink-table-planner-blink：planner計(jì)劃器，是table API最主要的部分，提供了運(yùn)行時(shí)環(huán)境和生成程序執(zhí)行計(jì)劃的planner；

• flink-table-api-scala-bridge：bridge橋接器，主要負(fù)責(zé)table API和 DataStream/DataSet API的連接支持，按照語言分java和scala。

這里的兩個(gè)依賴，是IDE環(huán)境下運(yùn)行需要添加的；如果是生產(chǎn)環(huán)境，lib目錄下默認(rèn)已經(jīng)有了planner，就只需要有bridge就可以了。

需要注意的是：flink table本身有兩個(gè) planner 計(jì)劃器，在flink 1.11之后，已經(jīng)默認(rèn)使用 blink planner，如果想了解 old planner，可以查閱官方文檔。

兩種planner（old & blink）的區(qū)別

• 批流統(tǒng)一：Blink將批處理作業(yè)，視為流式處理的特殊情況。所以，blink不支持表和DataSet之間的轉(zhuǎn)換，批處理作業(yè)將不轉(zhuǎn)換為DataSet應(yīng)用程序，而是跟流處理一樣，轉(zhuǎn)換為DataStream程序來處理。

• 因?yàn)榕鹘y(tǒng)一，Blink planner也不支持BatchTableSource，而使用有界的StreamTableSource代替。

• Blink planner只支持全新的目錄，不支持已棄用的ExternalCatalog。

• 舊planner和Blink planner的FilterableTableSource實(shí)現(xiàn)不兼容。舊的planner會(huì)把PlannerExpressions下推到filterableTableSource中，而blink planner則會(huì)把Expressions下推。

• 基于字符串的鍵值配置選項(xiàng)僅適用于Blink planner。

• PlannerConfig在兩個(gè)planner中的實(shí)現(xiàn)不同。

• Blink planner會(huì)將多個(gè)sink優(yōu)化在一個(gè)DAG中（僅在TableEnvironment上受支持，而在StreamTableEnvironment上不受支持）。而舊planner的優(yōu)化總是將每一個(gè)sink放在一個(gè)新的DAG中，其中所有DAG彼此獨(dú)立。

• 舊的planner不支持目錄統(tǒng)計(jì)，而Blink planner支持。

1 基本程序結(jié)構(gòu)

Table API 和 SQL 的程序結(jié)構(gòu)，與流式處理的程序結(jié)構(gòu)類似；也可以近似地認(rèn)為有這么幾步：首先創(chuàng)建執(zhí)行環(huán)境，然后定義source、transform和sink。

具體操作流程如下：

val tableEnv = ... // 創(chuàng)建表環(huán)境

// 創(chuàng)建表
tableEnv.connect(...).createTemporaryTable("table1")
// 注冊(cè)輸出表
tableEnv.connect(...).createTemporaryTable("outputTable")

// 使用 Table API query 創(chuàng)建表
val tapiResult = tableEnv.from("table1").select(...)
// 使用 SQL query 創(chuàng)建表
val sqlResult  = tableEnv.sqlQuery("SELECT ... FROM table1 ...")

// 輸出一張結(jié)果表到 TableSink，SQL查詢的結(jié)果表也一樣
TableResult tableResult = tapiResult.executeInsert("outputTable");
tableResult...

// 執(zhí)行
tableEnv.execute("scala_job")

2 創(chuàng)建表環(huán)境

表環(huán)境（TableEnvironment）是flink中集成Table API & SQL的核心概念。它負(fù)責(zé):

• 在內(nèi)部的 catalog 中注冊(cè) Table

• 注冊(cè)外部的 catalog

• 加載可插拔模塊

• 執(zhí)行 SQL 查詢

• 注冊(cè)自定義函數(shù) （scalar、table 或 aggregation）

• 將 DataStream 或 DataSet 轉(zhuǎn)換成 Table

• 持有對(duì) ExecutionEnvironment 或 StreamExecutionEnvironment 的引用

在創(chuàng)建TableEnv的時(shí)候，可以多傳入一個(gè)EnvironmentSettings或者TableConfig參數(shù)，可以用來配置TableEnvironment的一些特性。

Table 總是與特定的 TableEnvironment 綁定。不能在同一條查詢中使用不同 TableEnvironment 中的表，例如，對(duì)它們進(jìn)行 join 或 union 操作。

TableEnvironment 可以通過靜態(tài)方法 BatchTableEnvironment.create() 或者 StreamTableEnvironment.create() 在 StreamExecutionEnvironment 或者 ExecutionEnvironment 中創(chuàng)建，TableConfig 是可選項(xiàng)。TableConfig可用于配置TableEnvironment或定制的查詢優(yōu)化和轉(zhuǎn)換過程(參見 查詢優(yōu)化)。

請(qǐng)確保選擇與你的編程語言匹配的特定的計(jì)劃器BatchTableEnvironment/StreamTableEnvironment。

如果兩種計(jì)劃器的 jar 包都在 classpath 中（默認(rèn)行為），你應(yīng)該明確地設(shè)置要在當(dāng)前程序中使用的計(jì)劃器。

基于blink版本的流處理環(huán)境（Blink-Streaming-Query）：

import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment
import org.apache.flink.table.api.EnvironmentSettings
import org.apache.flink.table.api.bridge.scala.StreamTableEnvironment

val bsEnv = StreamExecutionEnvironment.getExecutionEnvironment
val bsSettings = EnvironmentSettings.newInstance().useBlinkPlanner().inStreamingMode().build()
val bsTableEnv = StreamTableEnvironment.create(bsEnv, bsSettings)

這里只提供了 blink planner 的流處理設(shè)置。有關(guān) old planner 的批處理和流處理的設(shè)置，以及 blink planner 的批處理的設(shè)置，請(qǐng)查閱官方文檔。

3 在Catalog中注冊(cè)表

TableEnvironment 維護(hù)著一個(gè)由標(biāo)識(shí)符（identifier）創(chuàng)建的表 catalog 的映射。標(biāo)識(shí)符由三個(gè)部分組成：catalog 名稱、數(shù)據(jù)庫名稱以及對(duì)象名稱。如果 catalog 或者數(shù)據(jù)庫沒有指明，就會(huì)使用當(dāng)前默認(rèn)值。

Table 可以是虛擬的（視圖 VIEWS）也可以是常規(guī)的（表 TABLES）。視圖 VIEWS可以從已經(jīng)存在的Table中創(chuàng)建，一般是 Table API 或者 SQL 的查詢結(jié)果。表TABLES描述的是外部數(shù)據(jù)，例如文件、數(shù)據(jù)庫表或者消息隊(duì)列。

臨時(shí)表（Temporary Table）和永久表（Permanent Table） 表可以是臨時(shí)的，并與單個(gè) Flink 會(huì)話（session）的生命周期相關(guān)，也可以是永久的，并且在多個(gè) Flink 會(huì)話和群集（cluster）中可見。

永久表需要 catalog（例如 Hive Metastore）以維護(hù)表的元數(shù)據(jù)。一旦永久表被創(chuàng)建，它將對(duì)任何連接到 catalog 的 Flink 會(huì)話可見且持續(xù)存在，直至被明確刪除。

另一方面，臨時(shí)表通常保存于內(nèi)存中并且僅在創(chuàng)建它們的 Flink 會(huì)話持續(xù)期間存在。這些表對(duì)于其它會(huì)話是不可見的。它們不與任何 catalog 或者數(shù)據(jù)庫綁定但可以在一個(gè)命名空間（namespace）中創(chuàng)建。即使它們對(duì)應(yīng)的數(shù)據(jù)庫被刪除，臨時(shí)表也不會(huì)被刪除。

創(chuàng)建表

虛擬表

在 SQL 的術(shù)語中，Table API 的對(duì)象對(duì)應(yīng)于視圖（虛擬表）。它封裝了一個(gè)邏輯查詢計(jì)劃。它可以通過以下方法在 catalog 中創(chuàng)建：

// get a TableEnvironment
val tableEnv = ... // see "Create a TableEnvironment" section

// table is the result of a simple projection query
val projTable: Table = tableEnv.from("X").select(...)

// register the Table projTable as table "projectedTable"
tableEnv.createTemporaryView("projectedTable", projTable)

擴(kuò)展表標(biāo)識(shí)符

表總是通過三元標(biāo)識(shí)符注冊(cè)，包括 catalog 名、數(shù)據(jù)庫名和表名。

用戶可以指定一個(gè) catalog 和數(shù)據(jù)庫作為 “當(dāng)前catalog” 和”當(dāng)前數(shù)據(jù)庫”。有了這些，那么剛剛提到的三元標(biāo)識(shí)符的前兩個(gè)部分就可以被省略了。如果前兩部分的標(biāo)識(shí)符沒有指定， 那么會(huì)使用當(dāng)前的 catalog 和當(dāng)前數(shù)據(jù)庫。用戶也可以通過 Table API 或 SQL 切換當(dāng)前的 catalog 和當(dāng)前的數(shù)據(jù)庫。

標(biāo)識(shí)符遵循 SQL 標(biāo)準(zhǔn)，因此使用時(shí)需要用反引號(hào)進(jìn)行轉(zhuǎn)義。

// get a TableEnvironment
val tEnv: TableEnvironment = ...;
tEnv.useCatalog("custom_catalog")
tEnv.useDatabase("custom_database")

val table: Table = ...;

// register the view named 'exampleView' in the catalog named 'custom_catalog'
// in the database named 'custom_database' 
tableEnv.createTemporaryView("exampleView", table)

// register the view named 'exampleView' in the catalog named 'custom_catalog'
// in the database named 'other_database' 
tableEnv.createTemporaryView("other_database.exampleView", table)

// register the view named 'example.View' in the catalog named 'custom_catalog'
// in the database named 'custom_database' 
tableEnv.createTemporaryView("`example.View`", table)

// register the view named 'exampleView' in the catalog named 'other_catalog'
// in the database named 'other_database' 
tableEnv.createTemporaryView("other_catalog.other_database.exampleView", table)

4 表的查詢

利用外部系統(tǒng)的連接器connector，我們可以讀寫數(shù)據(jù)，并在環(huán)境的Catalog中注冊(cè)表。接下來就可以對(duì)表做查詢轉(zhuǎn)換了。

Flink給我們提供了兩種查詢方式：Table API和 SQL。

Table API的調(diào)用

Table API是集成在Scala和Java語言內(nèi)的查詢API。與SQL不同，Table API的查詢不會(huì)用字符串表示，而是在宿主語言中一步一步調(diào)用完成的。

Table API基于代表一張“表”的Table類，并提供一整套操作處理的方法API。這些方法會(huì)返回一個(gè)新的Table對(duì)象，這個(gè)對(duì)象就表示對(duì)輸入表應(yīng)用轉(zhuǎn)換操作的結(jié)果。有些關(guān)系型轉(zhuǎn)換操作，可以由多個(gè)方法調(diào)用組成，構(gòu)成鏈?zhǔn)秸{(diào)用結(jié)構(gòu)。例如table.select(…).filter(…)，其中select（…）表示選擇表中指定的字段，filter(…)表示篩選條件。

代碼中的實(shí)現(xiàn)如下：

// 獲取表環(huán)境
val tableEnv = ...

// 注冊(cè)訂單表

// 掃描注冊(cè)的訂單表
val orders = tableEnv.from("Orders")
// 計(jì)算來自法國的客戶的總收入
val revenue = orders
  .filter($"cCountry" === "FRANCE")
  .groupBy($"cID", $"cName")
  .select($"cID", $"cName", $"revenue".sum AS "revSum")

// 輸出或者轉(zhuǎn)換表
// 執(zhí)行查詢

注意：需要導(dǎo)入的隱式類型轉(zhuǎn)換

org.apache.flink.table.api._
org.apache.flink.api.scala._
org.apache.flink.table.api.bridge.scala._

SQL查詢

Flink的SQL集成，基于的是Apache Calcite，它實(shí)現(xiàn)了SQL標(biāo)準(zhǔn)。在Flink中，用常規(guī)字符串來定義SQL查詢語句。SQL 查詢的結(jié)果，是一個(gè)新的 Table。

代碼實(shí)現(xiàn)如下：

// get a TableEnvironment
val tableEnv = ... // see "Create a TableEnvironment" section

// register Orders table

// compute revenue for all customers from France
val revenue = tableEnv.sqlQuery("""
  |SELECT cID, cName, SUM(revenue) AS revSum
  |FROM Orders
  |WHERE cCountry = 'FRANCE'
  |GROUP BY cID, cName
  """.stripMargin)

// emit or convert Table
// execute query

如下的示例展示了如何指定一個(gè)更新查詢，將查詢的結(jié)果插入到已注冊(cè)的表中。

// get a TableEnvironment
val tableEnv = ... // see "Create a TableEnvironment" section

// register "Orders" table
// register "RevenueFrance" output table

// compute revenue for all customers from France and emit to "RevenueFrance"
tableEnv.executeSql("""
  |INSERT INTO RevenueFrance
  |SELECT cID, cName, SUM(revenue) AS revSum
  |FROM Orders
  |WHERE cCountry = 'FRANCE'
  |GROUP BY cID, cName
  """.stripMargin)

5 將DataStream轉(zhuǎn)換成表

Flink允許我們把Table和DataStream做轉(zhuǎn)換：我們可以基于一個(gè)DataStream，先流式地讀取數(shù)據(jù)源，然后map成樣例類，再把它轉(zhuǎn)成Table。Table的列字段（column fields），就是樣例類里的字段，這樣就不用再麻煩地定義schema了。

代碼表達(dá)

代碼中實(shí)現(xiàn)非常簡(jiǎn)單，直接用tableEnv.fromDataStream()就可以了。默認(rèn)轉(zhuǎn)換后的 Table schema 和 DataStream 中的字段定義一一對(duì)應(yīng)，也可以單獨(dú)指定出來。

這就允許我們更換字段的順序、重命名，或者只選取某些字段出來，相當(dāng)于做了一次map操作（或者Table API的 select操作）。

代碼具體如下：

val inputStream: DataStream[String] = env.readTextFile("sensor.txt")
val dataStream: DataStream[SensorReading] = inputStream
  .map(data => {
    val dataArray = data.split(",")
    SensorReading(dataArray(0), dataArray(1).toLong, dataArray(2).toDouble)
  })

val sensorTable: Table = tableEnv.fromDataStream(dataStream)

val sensorTable2 = tableEnv.fromDataStream(dataStream, 'id, 'timestamp as 'ts)

數(shù)據(jù)類型與Table schema的對(duì)應(yīng)

在上節(jié)的例子中，DataStream 中的數(shù)據(jù)類型，與表的 Schema 之間的對(duì)應(yīng)關(guān)系，是按照樣例類中的字段名來對(duì)應(yīng)的（name-based mapping），所以還可以用as做重命名。

另外一種對(duì)應(yīng)方式是，直接按照字段的位置來對(duì)應(yīng)（position-based mapping），對(duì)應(yīng)的過程中，就可以直接指定新的字段名了。

基于名稱的對(duì)應(yīng)：

val sensorTable = tableEnv
  .fromDataStream(dataStream, $"timestamp" as "ts", $"id" as "myId", "temperature")

基于位置的對(duì)應(yīng)：

val sensorTable = tableEnv
  .fromDataStream(dataStream, $"myId", $"ts")

Flink的DataStream和 DataSet API支持多種類型。

組合類型，比如元組（內(nèi)置Scala和Java元組）、POJO、Scala case類和Flink的Row類型等，允許具有多個(gè)字段的嵌套數(shù)據(jù)結(jié)構(gòu)，這些字段可以在Table的表達(dá)式中訪問。其他類型，則被視為原子類型。

元組類型和原子類型，一般用位置對(duì)應(yīng)會(huì)好一些；如果非要用名稱對(duì)應(yīng)，也是可以的：

元組類型，默認(rèn)的名稱是 "_1 , "_2"；而原子類型，默認(rèn)名稱是 ”f0”。

6 創(chuàng)建臨時(shí)視圖

創(chuàng)建臨時(shí)視圖的第一種方式，就是直接從DataStream轉(zhuǎn)換而來。同樣，可以直接對(duì)應(yīng)字段轉(zhuǎn)換；也可以在轉(zhuǎn)換的時(shí)候，指定相應(yīng)的字段。

代碼如下：

tableEnv.createTemporaryView("sensorView", dataStream)
tableEnv.createTemporaryView("sensorView",
  dataStream, $"id", $"temperature", $"timestamp" as "ts")

另外，當(dāng)然還可以基于Table創(chuàng)建視圖：

tableEnv.createTemporaryView("sensorView", sensorTable)

View和Table的Schema完全相同。事實(shí)上，在Table API中，可以認(rèn)為View和Table是等價(jià)的。

7 輸出表

更新模式（Update Mode）

在流處理過程中，表的處理并不像傳統(tǒng)定義的那樣簡(jiǎn)單。

對(duì)于流式查詢（Streaming Queries），需要聲明如何在（動(dòng)態(tài)）表和外部連接器之間執(zhí)行轉(zhuǎn)換。與外部系統(tǒng)交換的消息類型，由更新模式（update mode）指定。

Flink Table API中的更新模式有以下三種：

1. 追加模式（Append Mode）

在追加模式下，表（動(dòng)態(tài)表）和外部連接器只交換插入（Insert）消息。

1. 撤回模式（Retract Mode）

在撤回模式下，表和外部連接器交換的是：添加（Add）和撤回（Retract）消息。

• 插入（Insert）會(huì)被編碼為添加消息；

• 刪除（Delete）則編碼為撤回消息；

• 更新（Update）則會(huì)編碼為，已更新行（上一行）的撤回消息，和更新行（新行）的添加消息。

在此模式下，不能定義key，這一點(diǎn)跟upsert模式完全不同。

• Upsert（更新插入）模式

在Upsert模式下，動(dòng)態(tài)表和外部連接器交換Upsert和Delete消息。

這個(gè)模式需要一個(gè)唯一的key，通過這個(gè)key可以傳遞更新消息。為了正確應(yīng)用消息，外部連接器需要知道這個(gè)唯一key的屬性。

• 插入（Insert）和更新（Update）都被編碼為Upsert消息；

• 刪除（Delete）編碼為Delete信息。

這種模式和Retract模式的主要區(qū)別在于，Update操作是用單個(gè)消息編碼的，所以效率會(huì)更高。

8 將表轉(zhuǎn)換成DataStream

表可以轉(zhuǎn)換為DataStream或DataSet。這樣，自定義流處理或批處理程序就可以繼續(xù)在 Table API或SQL查詢的結(jié)果上運(yùn)行了。

將表轉(zhuǎn)換為DataStream或DataSet時(shí)，需要指定生成的數(shù)據(jù)類型，即要將表的每一行轉(zhuǎn)換成的數(shù)據(jù)類型。通常，最方便的轉(zhuǎn)換類型就是Row。當(dāng)然，因?yàn)榻Y(jié)果的所有字段類型都是明確的，我們也經(jīng)常會(huì)用元組類型來表示。

表作為流式查詢的結(jié)果，是動(dòng)態(tài)更新的。所以，將這種動(dòng)態(tài)查詢轉(zhuǎn)換成的數(shù)據(jù)流，同樣需要對(duì)表的更新操作進(jìn)行編碼，進(jìn)而有不同的轉(zhuǎn)換模式。

Table API中表到DataStream有兩種模式：

• 追加模式（Append Mode）

用于表只會(huì)被插入（Insert）操作更改的場(chǎng)景。

• 撤回模式（Retract Mode）

用于任何場(chǎng)景。有些類似于更新模式中Retract模式，它只有Insert和Delete兩類操作。

得到的數(shù)據(jù)會(huì)增加一個(gè)Boolean類型的標(biāo)識(shí)位（返回的第一個(gè)字段），用它來表示到底是新增的數(shù)據(jù)（Insert），還是被刪除的數(shù)據(jù)（老數(shù)據(jù)，Delete）。

代碼實(shí)現(xiàn)如下：

val resultStream: DataStream[Row] = tableEnv
  .toAppendStream[Row](resultTable)

val aggResultStream: DataStream[(Boolean, (String, Long))] = tableEnv
  .toRetractStream[(String, Long)](aggResultTable)

resultStream.print("result")
aggResultStream.print("aggResult")

所以，沒有經(jīng)過groupby之類聚合操作，可以直接用toAppendStream來轉(zhuǎn)換；而如果經(jīng)過了聚合，有更新操作，一般就必須用toRetractDstream。

9 Query的解釋和執(zhí)行

Table API提供了一種機(jī)制來解釋（Explain）計(jì)算表的邏輯和優(yōu)化查詢計(jì)劃。這是通過TableEnvironment.explain（table）方法或TableEnvironment.explain（）方法完成的。

explain方法會(huì)返回一個(gè)字符串，描述三個(gè)計(jì)劃：

• 未優(yōu)化的邏輯查詢計(jì)劃

• 優(yōu)化后的邏輯查詢計(jì)劃

• 實(shí)際執(zhí)行計(jì)劃

我們可以在代碼中查看執(zhí)行計(jì)劃：

val explaination: String = tableEnv.explain(resultTable)
println(explaination)

Query的解釋和執(zhí)行過程，老planner和blink planner大體是一致的，又有所不同。整體來講，Query都會(huì)表示成一個(gè)邏輯查詢計(jì)劃，然后分兩步解釋：

1. 優(yōu)化查詢計(jì)劃

2. 解釋成 DataStream 或者 DataSet程序

而Blink版本是批流統(tǒng)一的，所以所有的Query，只會(huì)被解釋成DataStream程序；另外在批處理環(huán)境TableEnvironment下，Blink版本要到tableEnv.execute()執(zhí)行調(diào)用才開始解釋。

Table API和SQL，本質(zhì)上還是基于關(guān)系型表的操作方式；而關(guān)系型表、關(guān)系代數(shù)，以及SQL本身，一般是有界的，更適合批處理的場(chǎng)景。這就導(dǎo)致在進(jìn)行流處理的過程中，理解會(huì)稍微復(fù)雜一些，需要引入一些特殊概念。

1 流處理和關(guān)系代數(shù)（表，及SQL）的區(qū)別

可以看到，其實(shí)關(guān)系代數(shù)（主要就是指關(guān)系型數(shù)據(jù)庫中的表）和SQL，主要就是針對(duì)批處理的，這和流處理有天生的隔閡。

2 動(dòng)態(tài)表

因?yàn)榱魈幚砻鎸?duì)的數(shù)據(jù)，是連續(xù)不斷的，這和我們熟悉的關(guān)系型數(shù)據(jù)庫中保存的“表”完全不同。所以，如果我們把流數(shù)據(jù)轉(zhuǎn)換成Table，然后執(zhí)行類似于table的select操作，結(jié)果就不是一成不變的，而是隨著新數(shù)據(jù)的到來，會(huì)不停更新。

我們可以隨著新數(shù)據(jù)的到來，不停地在之前的基礎(chǔ)上更新結(jié)果。這樣得到的表，在Flink Table API概念里，就叫做“動(dòng)態(tài)表”（Dynamic Tables）。

動(dòng)態(tài)表是Flink對(duì)流數(shù)據(jù)的Table API和SQL支持的核心概念。與表示批處理數(shù)據(jù)的靜態(tài)表不同，動(dòng)態(tài)表是隨時(shí)間變化的。動(dòng)態(tài)表可以像靜態(tài)的批處理表一樣進(jìn)行查詢，查詢一個(gè)動(dòng)態(tài)表會(huì)產(chǎn)生持續(xù)查詢（Continuous Query）。連續(xù)查詢永遠(yuǎn)不會(huì)終止，并會(huì)生成另一個(gè)動(dòng)態(tài)表。查詢（Query）會(huì)不斷更新其動(dòng)態(tài)結(jié)果表，以反映其動(dòng)態(tài)輸入表上的更改。

3 流式持續(xù)查詢的過程

下圖顯示了流、動(dòng)態(tài)表和連續(xù)查詢的關(guān)系：

流式持續(xù)查詢的過程為：

1. 流被轉(zhuǎn)換為動(dòng)態(tài)表

2. 對(duì)動(dòng)態(tài)表計(jì)算連續(xù)查詢，生成新的動(dòng)態(tài)表

3. 生成的動(dòng)態(tài)表被轉(zhuǎn)換回流

3.1 將流轉(zhuǎn)換成表（Table）

為了處理帶有關(guān)系查詢的流，必須先將其轉(zhuǎn)換為表。

從概念上講，流的每個(gè)數(shù)據(jù)記錄，都被解釋為對(duì)結(jié)果表的插入（Insert）修改。因?yàn)榱魇匠掷m(xù)不斷的，而且之前的輸出結(jié)果無法改變。本質(zhì)上，我們其實(shí)是從一個(gè)、只有插入操作的changelog（更新日志）流，來構(gòu)建一個(gè)表。

為了更好地說明動(dòng)態(tài)表和持續(xù)查詢的概念，我們來舉一個(gè)具體的例子。

比如，我們現(xiàn)在的輸入數(shù)據(jù)，就是用戶在網(wǎng)站上的訪問行為，數(shù)據(jù)類型（Schema）如下：

{
  user:  VARCHAR,   // 用戶名
  cTime: TIMESTAMP, // 訪問某個(gè)URL的時(shí)間戳
  url:   VARCHAR    // 用戶訪問的URL
}

下圖顯示了如何將訪問URL事件流，或者叫點(diǎn)擊事件流（左側(cè)）轉(zhuǎn)換為表（右側(cè)）。

隨著插入更多的訪問事件流記錄，生成的表將不斷增長(zhǎng)。

3.2 持續(xù)查詢（Continuous Query）

持續(xù)查詢，會(huì)在動(dòng)態(tài)表上做計(jì)算處理，并作為結(jié)果生成新的動(dòng)態(tài)表。與批處理查詢不同，連續(xù)查詢從不終止，并根據(jù)輸入表上的更新更新其結(jié)果表。

在任何時(shí)間點(diǎn)，連續(xù)查詢的結(jié)果在語義上，等同于在輸入表的快照上，以批處理模式執(zhí)行的同一查詢的結(jié)果。

在下面的示例中，我們展示了對(duì)點(diǎn)擊事件流中的一個(gè)持續(xù)查詢。

這個(gè)Query很簡(jiǎn)單，是一個(gè)分組聚合做count統(tǒng)計(jì)的查詢。它將用戶字段上的clicks表分組，并統(tǒng)計(jì)訪問的url數(shù)。圖中顯示了隨著時(shí)間的推移，當(dāng)clicks表被其他行更新時(shí)如何計(jì)算查詢。

3.3 將動(dòng)態(tài)表轉(zhuǎn)換成流

與常規(guī)的數(shù)據(jù)庫表一樣，動(dòng)態(tài)表可以通過插入（Insert）、更新（Update）和刪除（Delete）更改，進(jìn)行持續(xù)的修改。將動(dòng)態(tài)表轉(zhuǎn)換為流或?qū)⑵鋵懭胪獠肯到y(tǒng)時(shí)，需要對(duì)這些更改進(jìn)行編碼。Flink的Table API和SQL支持三種方式對(duì)動(dòng)態(tài)表的更改進(jìn)行編碼：

• 僅追加（Append-only）流

僅通過插入（Insert）更改，來修改的動(dòng)態(tài)表，可以直接轉(zhuǎn)換為“僅追加”流。這個(gè)流中發(fā)出的數(shù)據(jù)，就是動(dòng)態(tài)表中新增的每一行。

• 撤回（Retract）流

Retract流是包含兩類消息的流，添加（Add）消息和撤回（Retract）消息。

動(dòng)態(tài)表通過將INSERT 編碼為add消息、DELETE 編碼為retract消息、UPDATE編碼為被更改行（前一行）的retract消息和更新后行（新行）的add消息，轉(zhuǎn)換為retract流。

下圖顯示了將動(dòng)態(tài)表轉(zhuǎn)換為Retract流的過程。

• Upsert（更新插入）流

Upsert流包含兩種類型的消息：Upsert消息和delete消息。轉(zhuǎn)換為upsert流的動(dòng)態(tài)表，需要有唯一的鍵（key）。

通過將INSERT和UPDATE更改編碼為upsert消息，將DELETE更改編碼為DELETE消息，就可以將具有唯一鍵（Unique Key）的動(dòng)態(tài)表轉(zhuǎn)換為流。

下圖顯示了將動(dòng)態(tài)表轉(zhuǎn)換為upsert流的過程。

這些概念我們之前都已提到過。需要注意的是，在代碼里將動(dòng)態(tài)表轉(zhuǎn)換為DataStream時(shí)，僅支持Append和Retract流。而向外部系統(tǒng)輸出動(dòng)態(tài)表的TableSink接口，則可以有不同的實(shí)現(xiàn)，比如之前我們講到的ES，就可以有Upsert模式。

4 時(shí)間特性

基于時(shí)間的操作（比如Table API和SQL中窗口操作），需要定義相關(guān)的時(shí)間語義和時(shí)間數(shù)據(jù)來源的信息。所以，Table可以提供一個(gè)邏輯上的時(shí)間字段，用于在表處理程序中，指示時(shí)間和訪問相應(yīng)的時(shí)間戳。

時(shí)間屬性，可以是每個(gè)表schema的一部分。一旦定義了時(shí)間屬性，它就可以作為一個(gè)字段引用，并且可以在基于時(shí)間的操作中使用。

時(shí)間屬性的行為類似于常規(guī)時(shí)間戳，可以訪問，并且進(jìn)行計(jì)算。

4.1 處理時(shí)間

處理時(shí)間語義下，允許表處理程序根據(jù)機(jī)器的本地時(shí)間生成結(jié)果。它是時(shí)間的最簡(jiǎn)單概念。它既不需要提取時(shí)間戳，也不需要生成watermark。

定義處理時(shí)間屬性有三種方法：在DataStream轉(zhuǎn)化時(shí)直接指定；在定義Table Schema時(shí)指定；在創(chuàng)建表的DDL中指定。

1. DataStream轉(zhuǎn)化成Table時(shí)指定

由DataStream轉(zhuǎn)換成表時(shí)，可以在后面指定字段名來定義Schema。在定義Schema期間，可以使用.proctime，定義處理時(shí)間字段。

注意，這個(gè)proctime屬性只能通過附加邏輯字段，來擴(kuò)展物理schema。因此，只能在schema定義的末尾定義它。

代碼如下：

val stream = env.addSource(new SensorSource)
val sensorTable = tableEnv
  .fromDataStream(stream, $"id", $"timestamp", $"temperature", $"pt".proctime())

1. 創(chuàng)建表的DDL中指定

在創(chuàng)建表的DDL中，增加一個(gè)字段并指定成proctime，也可以指定當(dāng)前的時(shí)間字段。

代碼如下：

val sinkDDL: String =
  """
    |create table dataTable (
    |  id varchar(20) not null,
    |  ts bigint,
    |  temperature double,
    |  pt AS PROCTIME()
    |) with (
    |  'connector.type' = 'filesystem',
    |  'connector.path' = 'sensor.txt',
    |  'format.type' = 'csv'
    |)
  """.stripMargin

tableEnv.sqlUpdate(sinkDDL) // 執(zhí)行 DDL

注意：運(yùn)行這段DDL，必須使用Blink Planner。

4.2 事件時(shí)間（Event Time）

事件時(shí)間語義，允許表處理程序根據(jù)每個(gè)記錄中包含的時(shí)間生成結(jié)果。這樣即使在有亂序事件或者延遲事件時(shí)，也可以獲得正確的結(jié)果。

為了處理無序事件，并區(qū)分流中的準(zhǔn)時(shí)和遲到事件；Flink需要從事件數(shù)據(jù)中，提取時(shí)間戳，并用來推進(jìn)事件時(shí)間的進(jìn)展（watermark）。

1. DataStream轉(zhuǎn)化成Table時(shí)指定

在DataStream轉(zhuǎn)換成Table，schema的定義期間，使用.rowtime可以定義事件時(shí)間屬性。注意，必須在轉(zhuǎn)換的數(shù)據(jù)流中分配時(shí)間戳和watermark。

在將數(shù)據(jù)流轉(zhuǎn)換為表時(shí)，有兩種定義時(shí)間屬性的方法。根據(jù)指定的.rowtime字段名是否存在于數(shù)據(jù)流的架構(gòu)中，timestamp字段可以：

• 作為新字段追加到schema

• 替換現(xiàn)有字段

在這兩種情況下，定義的事件時(shí)間戳字段，都將保存DataStream中事件時(shí)間戳的值。

代碼如下：

val stream = env
  .addSource(new SensorSource)
  .assignAscendingTimestamps(r => r.timestamp)
// 將 DataStream轉(zhuǎn)換為 Table，并指定時(shí)間字段
val sensorTable = tableEnv
  .fromDataStream(stream, $"id", $"timestamp".rowtime(), 'temperature)

1. 創(chuàng)建表的DDL中指定

事件時(shí)間屬性，是使用CREATE TABLE DDL中的WARDMARK語句定義的。watermark語句，定義現(xiàn)有事件時(shí)間字段上的watermark生成表達(dá)式，該表達(dá)式將事件時(shí)間字段標(biāo)記為事件時(shí)間屬性。

代碼如下：

val sinkDDL: String =
  """
    |create table dataTable (
    |  id varchar(20) not null,
    |  ts bigint,
    |  temperature double,
    |  rt AS TO_TIMESTAMP( FROM_UNIXTIME(ts) ),
    |  watermark for rt as rt - interval '1' second
    |) with (
    |  'connector.type' = 'filesystem',
    |  'connector.path' = 'file:///D:\\..\\sensor.txt',
    |  'format.type' = 'csv'
    |)
  """.stripMargin

tableEnv.sqlUpdate(sinkDDL) // 執(zhí)行 DDL

這里FROM_UNIXTIME是系統(tǒng)內(nèi)置的時(shí)間函數(shù)，用來將一個(gè)整數(shù)（秒數(shù)）轉(zhuǎn)換成“YYYY-MM-DD hh:mm:ss”格式（默認(rèn)，也可以作為第二個(gè)String參數(shù)傳入）的日期時(shí)間字符串（date time string）；然后再用TO_TIMESTAMP將其轉(zhuǎn)換成Timestamp。

未完待續(xù)。