flink sqltable編程案例海量資源

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