如何跑通第一个 SQL 作业
作者 | 周凱波（寶牛），阿里巴巴技術專家
本文由阿里巴巴技術專家周凱波（寶牛）分享，主要介紹如何跑通第一個SQL。內容將從以下3個部分介紹：
1、SQL的基本概念
2、SQL的語法說明
3、SQL 實戰
一、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。
二、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，適合需要輸出到多個下游的場景。
三、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的任務運行起來了。
原文鏈接：https://developer.aliyun.com/article/781172?
版權聲明：本文內容由阿里云實名注冊用戶自發貢獻，版權歸原作者所有，阿里云開發者社區不擁有其著作權，亦不承擔相應法律責任。具體規則請查看《阿里云開發者社區用戶服務協議》和《阿里云開發者社區知識產權保護指引》。如果您發現本社區中有涉嫌抄襲的內容，填寫侵權投訴表單進行舉報，一經查實，本社區將立刻刪除涉嫌侵權內容。總結
以上是生活随笔為你收集整理的如何跑通第一个 SQL 作业的全部內容，希望文章能夠幫你解決所遇到的問題。
- 上一篇： Android 热修复之DexPatch
- 下一篇： mPaas-RPC拦截器各种场景下的使用