主要內容本篇主要從FlinkSQL實現(xiàn)的內核與原理，工作流等的視角帶大家構建一幅FlinkSQL全景圖（以Blink為主介紹），探知背后支撐的“男人們”（組件）。建議收藏，僅此一份。

主要內容：

1. Table API 與 SQL

2. Apache Calcite

3. 元數(shù)據

4. SQL 函數(shù)

5. Flink Planner 與 Blink Planner

6. Blink SQL執(zhí)行過程

7. SQL優(yōu)化器

8. 總結

Table API 與 Table SQLTable API 和 Table SQL 集成在同一套 API 中。這套 API 的核心概念是Table，用作查詢的輸入和輸出。

Apache Flink 具有兩個關系型 API - Table API 和 Table SQL - 用于統(tǒng)一的流和批處理。Table API 是 Scala 和 Java 的語言集成查詢 API，它允許用非常直觀的方式從關系運算符（如選擇、過濾和連接）組成查詢。Flink 的 SQL 支持是基于 Apache Calcite，它實現(xiàn)了 SQL 標準。無論輸入是批處理輸入（DataSet）還是流輸入（DataStream），在任一接口中指定的查詢都具有相同的語義，并指定相同的結果。

Table API 和 SQL 接口與 Flink 的 DataStream 和 DataSet API 緊密集成。你可以很容易地在所有 API 和建立在 API 基礎上的庫之間切換。

Apache CalciteCalcite 是什么

Apache Calcite是一款開源的動態(tài)數(shù)據管理框架，它提供了標準的 SQL 語言、多種查詢優(yōu)化和連接各種數(shù)據源的能力，但不包括數(shù)據存儲、處理數(shù)據的算法和存儲元數(shù)據的存儲庫。

Calcite采用的是業(yè)界大數(shù)據查詢框架的一種通用思路，它的目標是“one size fits all（一種方案適應所有需求場景）”，希望能為不同計算平臺和數(shù)據源提供統(tǒng)一的查詢引擎。

Calcite作為一個強大的SQL計算引擎，在Flink內部的SQL引擎模塊就是基于Calcite。

Calcite 的特點

支持標準SQL語言；

獨立于編程語言和數(shù)據源，可以支持不同的前端和后端；

支持關系代數(shù)、可定制的邏輯規(guī)則和基于成本模型優(yōu)化的查詢引擎；

支持物化視圖（materialized view）的管理（創(chuàng)建、丟棄、持久化和自動識別）；

基于物化視圖的Lattice和Tile機制，以應用于OLAP分析；

支持對流數(shù)據的查詢。

Calcite 的功能

1. SQL 解析

Calcite 的SQL解析是通過JavaCC實現(xiàn)的，使用JavaCC編寫SQL語法描述文件，將SQL解析成未經校驗的AST語法樹。

2. SQL 校驗

無狀態(tài)的校驗：驗證SQL語句是否符合規(guī)范。有狀態(tài)的校驗：通過與元數(shù)據結合驗證SQL的Schema，F(xiàn)ield，F(xiàn)unction是否存在，輸入和輸出是否符合。

3. 查詢優(yōu)化

對RelNode和邏輯計劃樹進行優(yōu)化，得到優(yōu)化后的生成物理執(zhí)行計劃。

4. SQL 生成器

將物理執(zhí)行計劃生成特定平臺的可執(zhí)行程序，比如Flink，Hive，不同規(guī)則的SQL查詢語句。

5. 執(zhí)行

通過各個執(zhí)行平臺在內存中編譯，然后執(zhí)行查詢。

FlinkSQL 結合 Calcite

一條SQL從提交到Calcite解析，優(yōu)化，到最后的Flink執(zhí)行，一般分以下過程：

1. Sql Parser： 將sql語句通過java cc解析成AST（語法樹），在calcite中用SqlNode表示AST；

2. Sql Validator： 結合數(shù)字字典（catalog）去驗證sql語法；

3. 生成Logical Plan： 將sqlNode表示的AST轉換成LogicalPlan， 用relNode表示;

4. 生成 optimized LogicalPlan： 先基于calcite rules 去優(yōu)化logical Plan，基于flink定制的一些優(yōu)化rules去優(yōu)化logical Plan；

5. 生成Flink PhysicalPlan： 這里也是基于flink里頭的rules將，將optimized LogicalPlan轉成成Flink的物理執(zhí)行計劃；

6. 將物理執(zhí)行計劃轉成Flink ExecutionPlan： 就是調用相應的tanslateToPlan方法轉換和利用CodeGen元編程成Flink的各種算子。

Table API 來提交任務的話，基本流程和運行SQL類似，稍微不同的是：table api parser： flink會把table api表達的計算邏輯也表示成一顆樹，用treeNode去表式;在這棵樹上的每個節(jié)點的計算邏輯用Expression來表示。

簡單說一下SQL優(yōu)化：RBO（基于規(guī)則）

RBO主要是開發(fā)人員在使用SQL的過程中，有些發(fā)現(xiàn)有些通用的規(guī)則，可以顯著提高SQL執(zhí)行的效率，比如最經典的filter下推：

將Filter下推到Join之前執(zhí)行，這樣做的好處是減少了Join的數(shù)量，同時降低了CPU，內存，網絡等方面的開銷，提高效率。

SQL優(yōu)化的發(fā)展，則可以分為兩個階段，即RBO（基于規(guī)則），和CBO（基于代價）

RBO和CBO的區(qū)別大概在于： RBO只為應用提供的rule，而CBO會根據給出的Cost信息，智能應用rule，求出一個Cost最低的執(zhí)行計劃。需要糾正很多人誤區(qū)的一點是，CBO其實也是基于rule的，接觸到RBO和CBO這兩個概念的時候，很容易將他們對立起來。但實際上CBO，可以理解為就是加上Cost的RBO。

元數(shù)據Catalog 提供了元數(shù)據信息，例如數(shù)據庫、表、分區(qū)、視圖以及數(shù)據庫或其他外部系統(tǒng)中存儲的函數(shù)和信息。

數(shù)據處理最關鍵的方面之一是管理元數(shù)據。元數(shù)據可以是臨時的，例如臨時表、或者通過 TableEnvironment 注冊的 UDF。元數(shù)據也可以是持久化的，例如 Hive Metastore 中的元數(shù)據。Catalog 提供了一個統(tǒng)一的API，用于管理元數(shù)據，并使其可以從 Table API 和 SQL 查詢語句中來訪問。

1. 目前支持的類型

（1） GenericInMemoryCatalog

是基于內存實現(xiàn)的 Catalog，所有元數(shù)據只在 session 的生命周期內可用。

（2） JdbcCatalog

JdbcCatalog 使得用戶可以將 Flink 通過 JDBC 協(xié)議連接到關系數(shù)據庫。PostgresCatalog 是當前實現(xiàn)的唯一一種 JDBC Catalog。

（3） HiveCatalog

HiveCatalog 有兩個用途：作為原生 Flink 元數(shù)據的持久化存儲，以及作為讀寫現(xiàn)有 Hive 元數(shù)據的接口。

（4） 用戶自定義 Catalog

Catalog 是可擴展的，用戶可以通過實現(xiàn) Catalog 接口來開發(fā)自定義 Catalog。想要在 SQL CLI 中使用自定義 Catalog，用戶除了需要實現(xiàn)自定義的 Catalog 之外，還需要為這個 Catalog 實現(xiàn)對應的 CatalogFactory 接口。

CatalogFactory 定義了一組屬性，用于 SQL CLI 啟動時配置 Catalog。這組屬性集將傳遞給發(fā)現(xiàn)服務，在該服務中，服務會嘗試將屬性關聯(lián)到 CatalogFactory 并初始化相應的 Catalog 實例。

2. 元數(shù)據分類

catalog定義主要有三種數(shù)據類型接口，也就是常用到的數(shù)據庫，表&視圖，函數(shù)。當然還有最上層的Catalog容器。

（1） 數(shù)據庫

等同于數(shù)據庫中庫的實例，接口定義為CatalogDatabase，定義數(shù)據庫實例的元數(shù)據，一個數(shù)據庫實例中包含表，視圖，函數(shù)等多種對象。

（2） 表&視圖

CatalogTable對應數(shù)據庫中的表，CatalogView隊形數(shù)據庫中的視圖。

表是一種存儲的實體，包換了字段信息，表的分區(qū)，屬性，描述信息。其實說白了字段定義和之前印象的數(shù)據庫很是類似。你可以對比過來。不同的是，拿flink來說，所有的表都是外部數(shù)據源，除了上面所說的，還需要訪問信息，比如IP端口，mater地址，connector連接類等等。

視圖是一個虛擬概念，本質上是一條SQL查詢語句，底層對應一張表或者多張表。包含SQL查詢語句，視圖的字段信息，視圖的屬性等等的信息。

（3） 函數(shù)

CatalogFunction是函數(shù)元數(shù)據的接口。函數(shù)元數(shù)據包含了所在的類信息和編程語言。

3. 數(shù)據訪問

Flink的Table API和SQL程序可以連接到其他外部系統(tǒng)，用于讀和寫批處理表和流表。source table提供對存儲在外部系統(tǒng)（如數(shù)據庫、消息隊列或文件系統(tǒng)）中的數(shù)據的訪問。sink table 向外部存儲系統(tǒng)發(fā)送表。根據source和sink器的類型，它們支持不同的格式，如CSV、Avro、Parquet或ORC。

（1） TableSchema

Table Source 和 Sink需要具備對外數(shù)據源的描述能力，所以Flink定義了TableSchema對象來定義字段名稱和字段類型，存儲格式等等信息

（2） 時間屬性

支持處理時間和時間時間

（3） Watermark

用來處理亂序的數(shù)據。

4. Table Source & Table Sink

Flink本地支持各種連接器，可以查看往期總結

Filesystem

Elasticsearch

Apache Kafka

Amazon Kinesis Data Streams

JDBC

Apache HBase

Apache Hive

幾個主要Table Source與Sink體系

（1） StreamTableSource

流數(shù)據抽象，區(qū)分了無界數(shù)據與有界數(shù)據。

（2） LookupableTableSource

按照Join條件中的字段進行關聯(lián)。

（3） FilterableTableSource

過濾不符合條件的記錄。

（4） LimitableTableSource

限制記錄條數(shù)。

（5） ProjectableTableSource

過濾不會被使用的字段。

（6） AppendStreamTableSink

追加模式的TableSink 支持追加，不支持更新。

（7） RetractStreamTableSink

支持召回模式的TableSink，召回模式其實就是流上的update。

（8） UpsertStreamTableSink

有則更新，無則插入

SQL 函數(shù)臨時函數(shù)和持久化函數(shù)。臨時函數(shù)始終由用戶創(chuàng)建，它容易改變并且僅在會話的生命周期內有效。持久化函數(shù)不是由系統(tǒng)提供，就是存儲在 Catalog 中，它在會話的整個生命周期內都有效。

內置函數(shù)

Table API和SQL為用戶提供了一組用于數(shù)據轉換的內置函數(shù)。如果您需要的函數(shù)還不受支持，您可以實現(xiàn)用戶定義的函數(shù)

（1） Comparison Functions（比較型函數(shù)）

eg：value1 = value2

（2） Logical Functions（邏輯函數(shù)）

eg： boolean1 OR boolean2

（3） Arithmetic Functions（算術函數(shù)）

eg： numeric1 + numeric2

（4） String Functions（字符串函數(shù)）

UPPER（string）

（5） Temporal Functions（時間函數(shù)）

YEAR（date）

（6） Conditional Functions（有條件的函數(shù)）

IF（condition， true_value， false_value）

（7） Type Conversion Functions（類型轉換函數(shù)）

CAST（value AS type）

（8） Collection Functions（集合函數(shù)）

array ‘［’ INT ‘］’

（9） Value Construction Functions ， Value Access Functions，Grouping Functions，Hash Functions，Auxiliary Functions，Aggregate Functions，Column Functions （不一一列舉）

自定義函數(shù)

（1） 標量函數(shù)（UDF）

標量函數(shù) 將標量值轉換成一個新標量值，也就是對一行數(shù)據中的一個或者多個字段返回一個單值。

（2） 聚合函數(shù)（UDAGG）

自定義聚合函數(shù)（UDAGG）是把一個表（一行或者多行，每行可以有一列或者多列）聚合成一個標量值。

（3） 表值函數(shù)（UDTF）

表值函數(shù) 將標量值轉換成新的行數(shù)據?？梢越邮找粋€或者多個字段作為參數(shù)，輸出多行列數(shù)據。

（4） 表值聚合函數(shù)（UDTAGG）

自定義表值聚合函數(shù)（UDTAGG）可以把一個表（一行或者多行，每行有一列或者多列）聚合成另一張表，結果中可以有多行多列。

（5） 異步表值函數(shù)

異步表值函數(shù) 是異步查詢外部數(shù)據系統(tǒng)的特殊函數(shù)。

Planner 與 Blink PlannerFlink Table/SQL體系中的Planner（即查詢處理器）是溝通Flink與Calcite的橋梁，為Table/SQL API提供完整的解析、優(yōu)化和執(zhí)行環(huán)境。

Flink Table 的新架構實現(xiàn)了查詢處理器的插件化，項目完整保留原有 Flink Planner （Old Planner），同時又引入了新的 Blink Planner，用戶可以自行選擇使用 Old Planner 還是 Blink Planner。

主要區(qū)別：

Blink做到了真正的流批統(tǒng)一，即將批看做是特殊的流，把處理批的API和處理流的API做成了一樣的。也就是說不管是批數(shù)據還是流數(shù)據，底層統(tǒng)統(tǒng)都是DataStream。所以使用Blink作為table planner的程序，Table和DataSet是不能相互轉換的。

Blink planner是不支持BatchTableSource的，它只支持StreamTableSource。

Blink Planner和Old Planner的FilterableTableSource是不兼容的。Old - Planner會下推PlannerExpression到FilterableTableSource。而Blink planner下推的是Expression。

基于String的鍵值對配置項只能用于Blink Planner

Blink Planner會優(yōu)化多個sink到同一個TableEnvironment和StreamTableEnvironment。而Old Planner會為不同的sink優(yōu)化到自己的DAG中，也就是說有幾個sink就有幾個DAG。

Old Planner 不支持 catalog統(tǒng)計，Blink支持。

Old Planner 不支持版本表（versioned Table）。版本表類似HBASE中版本表的意思，每個key可以記住過去的幾個值。

Blink SQL執(zhí)行過程

SQL執(zhí)行過程分三個階段

（1） 從SQL到 Operation

（2） 從Operation 到 Transformation

（3） 環(huán)境的執(zhí)行階段

從SQL到 Operation

（1） 解析SQL轉換為QueryOperation；

（2） SQL解析為SqlNode；

（3） 校驗SqlNode；

（4） 調用Calcite SQLToRelConvertrt將SqlNode轉化為RelNode邏輯樹；

（5） RelNode轉化為Operation。

Operation 到 Transformation

（1） DQL（數(shù)據查詢語言）轉換，在flink中作為中間運算；

（2） DML（數(shù)據操作語言），DQL轉換。

整個轉換從Operation開始，先轉換為Calcite的邏輯計劃樹，再轉化為Flink的邏輯計劃樹，然后進行優(yōu)化。優(yōu)化后的邏輯樹轉換為Flink的物理執(zhí)行，物理執(zhí)行生成一系列的算子，udf等等，包裝到Transformation中。

環(huán)境的執(zhí)行階段

有了Transformation后正式進入到StreamGraph的過程中，最終交給Flink集群去運行。

SQL優(yōu)化器查詢優(yōu)化器

再次提到兩個優(yōu)化器：RBO（基于規(guī)則的優(yōu)化器） 和 CBO（基于代價的優(yōu)化器）

（1） RBO（基于規(guī)則的優(yōu)化器）會將原有表達式裁剪掉，遍歷一系列規(guī)則（Rule），只要滿足條件就轉換，生成最終的執(zhí)行計劃。一些常見的規(guī)則包括分區(qū)裁剪（Partition Prune）、列裁剪、謂詞下推（Predicate Pushdown）、投影下推（Projection Pushdown）、聚合下推、limit下推、sort下推、常量折疊（Constant Folding）、子查詢內聯(lián)轉join等。

（2） CBO（基于代價的優(yōu)化器）會將原有表達式保留，基于統(tǒng)計信息和代價模型，嘗試探索生成等價關系表達式，最終取代價最小的執(zhí)行計劃。CBO的實現(xiàn)有兩種模型，Volcano模型，Cascades模型。這兩種模型思想很是相似，不同點在于Cascades模型一邊遍歷SQL邏輯樹，一邊優(yōu)化，從而進一步裁剪掉一些執(zhí)行計劃。

目前各大數(shù)據庫和計算引擎傾向于CBO。

總結在目前情況下，在阿里對Flink社區(qū)的貢獻下，F(xiàn)link包含了Flink SQL 和 Blink SQL體系，F(xiàn)link Planner稱之為 Old Planner，Blink Planner稱之為 New Planner。從中可以發(fā)現(xiàn) Blink Planner是未來，F(xiàn)link Planner將會被淘汰。

FlinkSQL依靠 Calcite提供了一套SQL驗證，解析，優(yōu)化等等操作。同時FlinkSQL提供元數(shù)據管理，SQL函數(shù)，數(shù)據源的建設。也自由化地提供了自定義函數(shù)，自定義connector連接，豐富了場景的使用。

FlinkSQL你值得擁有?。?！

編輯：jq