本文由簡悅SimpRead轉碼，原文地址我們在第本文由簡悅SimpRead轉碼，原文地址我們在第02課時中使用FlinkTable&SQL的API實現了最簡單的WordCount程序。在這一課時中，將分別從FlinkTable&SQL的背景和編程模型、常見的API、算子和內置函數等對FlinkTable&SQLFlinkTable&SQL開發。我們在前面的課時中講過Flink的分層模型，Flink自身提供了不同級別的抽象來支持我們開發流式或者批量處理程序，下圖描述了Flink支持的4種不同級別的抽象。TableAPISQLFlinkAPI操作。FlinkSQLFlink實時計算為簡化計算SQL語義的開發語言。我們在第04課時中提到過，Flink在編程模型上提供了DataStream和DataSet兩套API，并沒有做到事實上的批流統一，因為用戶和開發者還是開發了兩套代碼。正是因為FlinkTable&SQL的加入，可以說Flink在某種程度上做到了事實上的批流一體。你之前可能都了解過Hive，在離線計算場景下HiveSQL的解析用到了ApacheCalcite，Flink同樣把SQL的解析、優化和執行教給了Calcite下圖是一張經典的FlinkTable&SQL實現原理圖，可以看到Calcite從圖中可以看到無論是批查詢SQL還是流式查詢SQL，都會經過對應的轉換器Parser轉換成為節點樹SQLNode從圖中可以看到無論是批查詢SQL還是流式查詢SQL，都會經過對應的轉換器Parser轉換成為節點樹SQLNodetree，然后生成邏輯執行計劃LogicalPlan，邏輯執行計劃在經過優化后生成真正可以執行的物理執行計劃，交給DataSet或者DataStream的API去執行。在這里我們不對Calcite一個完整的FlinkTable&SQLJob也是由Source、Transformation、SinkSource部分Kafka、MySQLTransformation部分FlinkTable&SQLSQLSelect、Groupby等，當然在這里也有更為復雜的多流Join、流與維表的Join等；Sink部分是指的結果存儲比如MySQL、HBase或Kakfa與傳統的表SQL查詢相比，FlinkTable&SQL在處理流數據時會時時刻刻處于動態的數據變化中，所動態表的查詢與靜態表一樣，但是，在查詢動態表的時候，SQL會做連續查詢，不會終止。我們舉個簡單的例子，Flink程序接受一個Kafka流作為輸入，Kafka首先，Kafka的消息會被源源不斷的解析成一張不斷增長的動態表，我們在動態表上執行的SQL會不斷FlinkTable&首先，Kafka的消息會被源源不斷的解析成一張不斷增長的動態表，我們在動態表上執行的SQL會不斷FlinkTable&SQL我們在講解FlinkTable&SQL所支持的常用算子前，需要說明一點，Flink自從0.9Table&SQLPleasenotethattheTableAPIandSQLarenotyetfeaturecompleteandarebeingactively[stream,batch]FlinkTable&SQL的開發一直在進行中，并沒有支持所有場景下的計算邏輯。從我個人實踐角度來講，目前FlinkSQL|||queryUNION[ALL]|queryEXCEPT|queryINTERSECTORDERBYorderItem[,orderItemLIMIT{count|ALL}|queryEXCEPT|queryINTERSECTORDERBYorderItem[,orderItemLIMIT{count|ALL}OFFSETstart{ROW|ROWS}FETCH{FIRST|NEXT}[count]expression[ASC|DESCSELECT[ALL|DISTINCT{*|projectItem[,projectItem]*FROMGROUPBY{groupItem[,groupItem}WINDOWwindowNameASwindowSpec[,windowNameAS]*SELECT[ALL|DISTINCT{*|projectItem[,projectItem]*expression[]columnAlias|tableAliastableReference|joinCondition[NATURAL]expression[]columnAlias|tableAliastableReference|joinCondition[NATURAL]ON|USING'('[,column[matchRecognize[[AS]alias['('columnAlias[,columnAlias]*')']TABLE][[catalogName.]schemaName.]LATERALTABLE'('functionName'('expression[,expression]*')'UNNEST'('expressionVALUESexpression[,expression'(''('expression[,expression]*CUBE'('expression[,expression]*ROLLUP'('expression'(''('expression[,expression]*CUBE'('expression[,expression]*ROLLUP'('expression[,expression]*GROUPINGSETS|ORDERBYorderItem[,orderItem]*PARTITIONBYexpression[,expression]*numericOrIntervalExpression|ROWSnumericExpression可以看到FlinkSQL和傳統的SQL一樣，支持了包含查詢、連接、聚合等場景，另外還支持了包括窗SELECT、WHERESQLDataStreamDataSetSELECT*FROMSELECTname，可以看到FlinkSQL和傳統的SQL一樣，支持了包含查詢、連接、聚合等場景，另外還支持了包括窗SELECT、WHERESQLDataStreamDataSetSELECT*FROMSELECTname，ageFROM當然我們也可以在WHERE條件中使用=、<、>、<>、>=、<=，以及AND、ORSELECTname，ageFROMTablewherenameLIKE小明SELECT*FROMTableWHEREage=SELECTname,FROMWHEREnameIN(SELECTnameFROMGROUPBY/GROUPBY用于進行分組操作，DISTINCTHAVING和傳統SQLDISTINCTnameFROMSUM(score)asTotalScoreFROMTableBYSUM(score)asTotalScoreFROMTableBYnameJOIN可以用于把來自兩個表的數據聯合起來形成結果表，目前Flink的Join只支持等值連接。Flink支持的JOIN類型包括：JOIN-INNERLEFTJOIN-LEFTOUTERRIGHTJOINJOIN-INNERLEFTJOIN-LEFTOUTERRIGHTJOIN-RIGHTOUTERFULLJOIN-FULLOUTERSELECTFROMLEFTJOINProductU=SELECTFROMRIGHTJOINProductONU=SELECTFROMFULLOUTERJOINProductONU=根據窗口數據劃分的不同，目前ApacheFlink有如下3滾動窗口滑動窗口，窗口數據有固定大小，并且有生成間隔；會話窗口[TUMBLE_START(timeCol,[TUMBLE_END(timeCol,FROMGROUPBY[gk],[TUMBLE_START(timeCol,[TUMBLE_END(timeCol,FROMGROUPBY[gk],TUMBLE(timeCol,'1'DAY)asFROMOrdersGROUPBYTUMBLE(timeLine,INTERVAL'1'DAY),其中，TUMBLE_STARTTUMBLE_END代表窗口的開始時間和窗口的結束時間，TUMBLE(timeLine,INTERVAL'1'DAY)中的timeLine代表時間字段所在的列，INTERVAL'1'DAY表示時間間隔為一天。其中，TUMBLE_STARTTUMBLE_END代表窗口的開始時間和窗口的結束時間，TUMBLE(timeLine,INTERVAL'1'DAY)中的timeLine代表時間字段所在的列，INTERVAL'1'DAY表示時間間隔為一天。滑動窗口有固定的大小，與滾動窗口不同的是滑動窗口可以通過slide參數控制滑動窗口的創建頻率。滑動窗口的語法與滾動窗口相比，只多了一個slideFROMGROUPBY[gk],HOP(timeCol,slide,例如，我們要每間隔一小時計算一次過去24SELECTproduct,SUM(amount)FROMOrdersGROUPBYHOP(rowtime,SELECTproduct,SUM(amount)FROMOrdersGROUPBYHOP(rowtime,INTERVAL'1'INTERVAL'1'DAY),INTERVAL'1'HOURSESSION_START(timeCol,gap)ASSESSION_END(timeCol,gap)ASFROMGROUPBY[gk],SESSION(timeCol,舉例，我們需要計算每個用戶過去1SELECTuser,SESSION_START(rowtime,INTERVAL'1'HOUR)AS舉例，我們需要計算每個用戶過去1SELECTuser,SESSION_START(rowtime,INTERVAL'1'HOUR)ASSESSION_ROWTIME(rowtime,INTERVAL'1'HOUR)ASsEnd,GROUPBYSESSION(rowtime,INTERVAL'1'HOUR),Flink中還有大量的內置函數，我們可以直接使用，將內置函數分類如下：上面分別介紹了FlinkTable&SQL上面分別介紹了FlinkTable&SQL的原理和支持的算子，我們模擬一個實時的數據流，然后講解SQLJOIN的用法。在上一課時中，我們利用Flink提供的自定義Source功能來實現一個自定義的實時數據源，具體實現privatebooleanisRunning=*重寫run*@param*@throwsItemitem=public*@throwsItemitem=publicvoidcancel()isRunning=ItemintinewArrayList<String>list=Itemitem=newitem.setName(list.get(newreturn我們把實時的商品數據流進行分流，分成item.setName(list.get(newreturn我們把實時的商品數據流進行分流，分成even和odd兩個流進行JOIN，條件是名稱相同，最后，把兩個流的JOIN結果輸出。classStreamingDemopublicstaticvoidmain(String[]args)throwsExceptionEnvironmentSettingsbsSettings=StreamExecutionEnvironmentbsEnv=StreamTableEnvironmentbsTableEnv=StreamTableEnvironment.create(bsEnv,MyStreamingSource()).map(newMapFunction<Item,Item>(){publicItemmap(Itemitem)throwsExceptionreturnDataStream<Item>=source.split(newvalue)List<String>=if%2==0)DataStream<Item>=source.split(newvalue)List<String>=if%2==0)elsereturnDataStream<Item>oddSelect=List<String>output=newif(value.getId()%2==0)}elsereturnreturnbsTableEnv.createTemporaryView("evenTable",evenSelect