如何跑通第一個 SQL 做業

簡介： 本文由阿里巴巴技術專家周凱波（寶牛）分享，主要介紹如何跑通第一個SQL。

1、SQL的基本概念

1.SQL 分類

SQL分爲四類，分別是數據查詢語言（DQL）、數據操縱語言（DML）、數據定義（DDL）語言和數據控制語言（DCL）。今天將介紹前三種語言的使用。

接下來介紹幾個基本概念。

2.SQL 開發

● Scripts，即SQL文本。在SQL文本里面能夠寫上文介紹的前三種語言；
● Schema，即元數據。SQL裏面須要使用的表和函數，是經過Schema進行定義的；
● Artifacts，即UDF Jar包；

3.Catalog

在 Flink SQL裏，Catalog是管理元數據的。Catalog經過Catalog.DB.Table來定位一張表。除了DB和Table，Catalog還能註冊函數，如UDF、UDTF和UDAF。

在Flink Catalog裏，有三種Catalog實現：

● 第一個是GenericInMemoryCatalog，是內存版的Catalog。日常在使用 Flink SQL的時候，默認是內存版的Catalog。當程序運行結束，第二次從新運行的時候，會從新生成一個Catalog實例。
● 第二個是HiveCatalog，Flink裏比較好的支持了HiveCatalog，能夠從Hive HMS裏讀取元數據，同時也能夠往Hive裏註冊表，寫數據到Hive裏面去。
● 第三個Catalog是 VVP平臺裏面開發的Catalog，即VvpCatalog，它實現了Flink Catalog的接口，底層是使用的數據庫。

4.Deployment

Deployment是一個做業的描述，目前有兩種任務類型，JAR和SQL。

Deployment上有升級策略（Upgrade strategy）和恢復策略（Restore strategy）。Upgrade strategy是指Deployment運行後，用戶能夠對Deployment的參數進行修改，這個修改如何影響Deployment的運行就是由不一樣的升級策略決定的；Restore strategy 指啓動 Flink任務時，是否從 Savepoint/Checkpoint進行恢復就是不一樣的恢復策略。

Flink的版本和配置，經常使用的Flink的參數均可以在這裏進行配置。例如：Task Managers 數量，Jobmanager和Taskmanager 的 CPU 和內存等。

Deployment上除了做業描述外，還有指望狀態和實際狀態。指望狀態是指用戶所指望的目標狀態，例如當要將運行中的做業中止時，指望狀態就是Canceled；操做完成的實際運行狀態就是實際狀態。

總的來講，Deployment是一個任務的描述模板。VVP平臺內部的狀態機會根據Deployment的指望狀態和實際狀態來控制做業的實際運行。

5.Job

Deployment啓動時會生成一個Job，這個Job對應一個具體的 Flink Job。同一時間，一個Deployment上只會有一個正在運行的Job。

2、SQL的語法說明

1.語法說明

首先看下圖的語句，分別是建立源表和建立結果表。

下圖是註冊函數。函數的註冊分爲兩步，第一步上傳JAR包，而後在系統上能夠勾選自動註冊；第二種是使用 Flink 語法進行手工註冊。

使用函數有兩種方式，第一是內置函數的使用，以下圖UPPER是 Flink 自帶的函數；第二種是自定義函數，像MyScalarFunc。

在VVP平臺裏，也支持 Flink 裏的Temporary Table，能夠將它理解爲臨時表，只在當前會話週期內有效。在下圖例子中，咱們建立了兩個Temporary Table，讀取datagen_source表中的數據，輸出到blackhole_sink表。

下圖是Temporary View的語法示例。前面兩段是同樣的臨時表；第三條語句是建立了一個tmp_view，它表明從Datagen_source的查詢。在Flink裏面Temporary View能夠理解爲讓SQL的書寫變得更簡單，它不會對數據進行一個持久化，和數據庫裏面View概念是不同的。第四條語句是從 view裏面讀取數據並寫入到sink表裏。

下圖是Statement set的語法示例，這個語法目前在 Flink 1.11版本里尚未，可是在VVP平臺作了一些支持。

如上圖，BEGIN STATEMENT SET和END這兩個語句之間能夠寫多條 insert into語句。上圖的例子是讀取datagen_source 表往兩張sink表同時寫。這個語句提交後會啓動一個完整的Flink Job，裏面會有1個source和兩個sink。

2.SQL的應用範圍

Create Table，它註冊的表會寫入系統Catalog裏，在VVP平臺上面會寫到VvpCatalog中，並進行持久化。好處是適合多個query共享元數據。

Create Temporary Table，臨時表。它會寫到內存版的Catalog裏，不會持久化。因此它適合不須要共享元數據的場景，只給當前query使用。

Create Temporary View，主要目的是簡化SQL語句。若是不經過Create Temporary View，對於邏輯複雜的SQL寫起來會至關複雜，可讀性也不好。

Statement Set，適合須要輸出到多個下游的場景。

3、SQL 實戰

接下來向你們展現銷量統計的實例。以下圖所示，需求是統計每小時成交量。

咱們首先建立兩張表，一個是源表，一個是結果表。下圖是建立源表的語句，數據源來自kafka，而後定義watermark是5秒鐘。

下圖是結果表，也是一個kafka表。

下圖是查詢語句，從源表讀取數據後，會經過tumble window窗口聚合對數據作一個統計，這樣就求出了每小時的成交量。

1.實戰演示

打開VVP的界面，左側有SQL編輯器，在這個編輯器左邊有三欄，第一欄是Scripts，寫SQL文本的地方；第二欄是Schemas，用來定義元數據；第三欄是Artifacts， 用來註冊UDF。

首先定義一張datagen_source的表。點擊右上角的驗證按鈕，驗證經過後點擊旁邊的運行。點擊運行以後，能夠在下面看到運行的結果，運行成功後點擊左側的Schemas，能夠找到剛剛建立的datagen_source表。

而後再建立一張sink表，connector類型是blackhole。而後驗證並運行。

這樣兩張表都已經註冊到Catalog裏面去了，默認的Catalog名字是VVP，Database名字是Default。

接下來就能夠寫SQL語句。好比說寫一條INSERT INTO語句，寫完以後點驗證並運行。在運行這條INSERT INTO語句時，系統會提示是否要建立一個SQL做業，點擊確認，補充名稱等信息，SQL做業就建立好了。

點擊啓動這個做業，啓動過程當中能夠在頁面上看到關於這個做業的不少信息和配置。

2.UDF實戰展現

UDF開發完成後會打一個JAR包，而後點SQL編輯器左側的Artifacts，而後點「+」號，將JAR 包上傳上來。

上傳完畢，完善JAR 包名稱等信息後，點確認完成。JAR包上傳過程當中，VVP系統會對JAR 包進行解析。解析以後系統會提示是否註冊，勾選須要註冊的內容，點擊建立Function。

而後以下圖，上面就是已註冊的Function，下面是可用Function，能夠選擇繼續註冊或關掉窗口。

若是不須要這個函數了，能夠在頁面左側找到已經註冊的Function，點擊右側尾部圖標，選擇Drop Function。若想從新註冊，有兩種方法，第一能夠點擊管理Function；第二經過 Flink的註冊函數手動註冊。
用註冊好的Function建立SQL做業。

在建立頁面下拉能夠看到不少高級配置，只修改本身須要的配置便可。

3.Temporary table的使用

上面的例子是註冊在Catalog裏的，若是不想每次都在Catalog裏面進行註冊，那就能夠直接使用 Temporary table。

以下圖將table的建立和INSERT INTO所有寫在一塊兒，這樣就能夠直接建立一個新的SQL做業，而不用提早在Catalog裏註冊了。

4.Temporary View

將前面Temporary View例子頁面中的語句複製到VVP平臺的SQL編輯器中，直接點擊運行就能夠建立一個做業。

5.Statement Set

將前面Statement Set例子頁面中的語句複製到編輯器中，直接點擊運行就能夠建立一個做業。啓動後，能夠經過下圖看到運行狀況，這個任務從一個源表中讀取數據輸出到了兩個不一樣的sink表中。

6.查詢實戰

將前面SQL 實戰中建立源表、結果表和查詢頁面的語句分別複製粘貼到VVP平臺的SQL編輯器並啓動運行。從下圖能夠看到這個讀寫kafka的任務運行起來了。

做者：周凱波（寶牛），阿里巴巴技術專家
原文連接 本文爲阿里雲原創內容，未經容許不得轉載