flink sqltable編程案例海量資源

本文由簡悅SimpRead轉(zhuǎn)碼，原文地址我們在第本文由簡悅SimpRead轉(zhuǎn)碼，原文地址我們在第02課時中使用FlinkTable&SQL的API實(shí)現(xiàn)了最簡單的WordCount程序。在這一課時中，將分別從FlinkTable&SQL的背景和編程模型、常見的API、算子和內(nèi)置函數(shù)等對FlinkTable&SQLFlinkTable&SQL開發(fā)。我們在前面的課時中講過Flink的分層模型，F(xiàn)link自身提供了不同級別的抽象來支持我們開發(fā)流式或者批量處理程序，下圖描述了Flink支持的4種不同級別的抽象。TableAPISQLFlinkAPI操作。FlinkSQLFlink實(shí)時計算為簡化計算SQL語義的開發(fā)語言。我們在第04課時中提到過，F(xiàn)link在編程模型上提供了DataStream和DataSet兩套API，并沒有做到事實(shí)上的批流統(tǒng)一，因?yàn)橛脩艉烷_發(fā)者還是開發(fā)了兩套代碼。正是因?yàn)镕linkTable&SQL的加入，可以說Flink在某種程度上做到了事實(shí)上的批流一體。你之前可能都了解過Hive，在離線計算場景下HiveSQL的解析用到了ApacheCalcite，F(xiàn)link同樣把SQL的解析、優(yōu)化和執(zhí)行教給了Calcite下圖是一張經(jīng)典的FlinkTable&SQL實(shí)現(xiàn)原理圖，可以看到Calcite從圖中可以看到無論是批查詢SQL還是流式查詢SQL，都會經(jīng)過對應(yīng)的轉(zhuǎn)換器Parser轉(zhuǎn)換成為節(jié)點(diǎn)樹SQLNode從圖中可以看到無論是批查詢SQL還是流式查詢SQL，都會經(jīng)過對應(yīng)的轉(zhuǎn)換器Parser轉(zhuǎn)換成為節(jié)點(diǎn)樹SQLNodetree，然后生成邏輯執(zhí)行計劃LogicalPlan，邏輯執(zhí)行計劃在經(jīng)過優(yōu)化后生成真正可以執(zhí)行的物理執(zhí)行計劃，交給DataSet或者DataStream的API去執(zhí)行。在這里我們不對Calcite一個完整的FlinkTable&SQLJob也是由Source、Transformation、SinkSource部分Kafka、MySQLTransformation部分FlinkTable&SQLSQLSelect、Groupby等，當(dāng)然在這里也有更為復(fù)雜的多流Join、流與維表的Join等；Sink部分是指的結(jié)果存儲比如MySQL、HBase或Kakfa與傳統(tǒng)的表SQL查詢相比，F(xiàn)linkTable&SQL在處理流數(shù)據(jù)時會時時刻刻處于動態(tài)的數(shù)據(jù)變化中，所動態(tài)表的查詢與靜態(tài)表一樣，但是，在查詢動態(tài)表的時候，SQL會做連續(xù)查詢，不會終止。我們舉個簡單的例子，F(xiàn)link程序接受一個Kafka流作為輸入，Kafka首先，Kafka的消息會被源源不斷的解析成一張不斷增長的動態(tài)表，我們在動態(tài)表上執(zhí)行的SQL會不斷FlinkTable&首先，Kafka的消息會被源源不斷的解析成一張不斷增長的動態(tài)表，我們在動態(tài)表上執(zhí)行的SQL會不斷FlinkTable&SQL我們在講解FlinkTable&SQL所支持的常用算子前，需要說明一點(diǎn)，F(xiàn)link自從0.9Table&SQLPleasenotethattheTableAPIandSQLarenotyetfeaturecompleteandarebeingactively[stream,batch]FlinkTable&SQL的開發(fā)一直在進(jìn)行中，并沒有支持所有場景下的計算邏輯。從我個人實(shí)踐角度來講，目前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和傳統(tǒng)的SQL一樣，支持了包含查詢、連接、聚合等場景，另外還支持了包括窗SELECT、WHERESQLDataStreamDataSetSELECT*FROMSELECTname，可以看到FlinkSQL和傳統(tǒng)的SQL一樣，支持了包含查詢、連接、聚合等場景，另外還支持了包括窗SELECT、WHERESQLDataStreamDataSetSELECT*FROMSELECTname，ageFROM當(dāng)然我們也可以在WHERE條件中使用=、<、>、<>、>=、<=，以及AND、ORSELECTname，ageFROMTablewherenameLIKE小明SELECT*FROMTableWHEREage=SELECTname,FROMWHEREnameIN(SELECTnameFROMGROUPBY/GROUPBY用于進(jìn)行分組操作，DISTINCTHAVING和傳統(tǒng)SQLDISTINCTnameFROMSUM(score)asTotalScoreFROMTableBYSUM(score)asTotalScoreFROMTableBYnameJOIN可以用于把來自兩個表的數(shù)據(jù)聯(lián)合起來形成結(jié)果表，目前Flink的Join只支持等值連接。Flink支持的JOIN類型包括：JOIN-INNERLEFTJOIN-LEFTOUTERRIGHTJOINJOIN-INNERLEFTJOIN-LEFTOUTERRIGHTJOIN-RIGHTOUTERFULLJOIN-FULLOUTERSELECTFROMLEFTJOINProductU=SELECTFROMRIGHTJOINProductONU=SELECTFROMFULLOUTERJOINProductONU=根據(jù)窗口數(shù)據(jù)劃分的不同，目前ApacheFlink有如下3滾動窗口滑動窗口，窗口數(shù)據(jù)有固定大小，并且有生成間隔；會話窗口[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代表窗口的開始時間和窗口的結(jié)束時間，TUMBLE(timeLine,INTERVAL'1'DAY)中的timeLine代表時間字段所在的列，INTERVAL'1'DAY表示時間間隔為一天。其中，TUMBLE_STARTTUMBLE_END代表窗口的開始時間和窗口的結(jié)束時間，TUMBLE(timeLine,INTERVAL'1'DAY)中的timeLine代表時間字段所在的列，INTERVAL'1'DAY表示時間間隔為一天?；瑒哟翱谟泄潭ǖ拇笮。c滾動窗口不同的是滑動窗口可以通過slide參數(shù)控制滑動窗口的創(chuàng)建頻率。滑動窗口的語法與滾動窗口相比，只多了一個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中還有大量的內(nèi)置函數(shù)，我們可以直接使用，將內(nèi)置函數(shù)分類如下：上面分別介紹了FlinkTable&SQL上面分別介紹了FlinkTable&SQL的原理和支持的算子，我們模擬一個實(shí)時的數(shù)據(jù)流，然后講解SQLJOIN的用法。在上一課時中，我們利用Flink提供的自定義Source功能來實(shí)現(xiàn)一個自定義的實(shí)時數(shù)據(jù)源，具體實(shí)現(xiàn)privatebooleanisRunning=*重寫run*@param*@throwsItemitem=public*@throwsItemitem=publicvoidcancel()isRunning=ItemintinewArrayList<String>list=Itemitem=newitem.setName(list.get(newreturn我們把實(shí)時的商品數(shù)據(jù)流進(jìn)行分流，分成item.setName(list.get(newreturn我們把實(shí)時的商品數(shù)據(jù)流進(jìn)行分流，分成even和odd兩個流進(jìn)行JOIN，條件是名稱相同，最后，把兩個流的JOIN結(jié)果輸出。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