中文手冊(cè)t00160x2 3接收并處理Kafka傳來數(shù)據(jù)

Slipstream簡(jiǎn) Slipstream的優(yōu) Slipstream的技術(shù)指 快速 停止 Slipstream基 Input Derived Window 與用編程語言編寫的流應(yīng)用對(duì) DMLSQL背景運(yùn) 結(jié)果輸 Slipstream的兩種模 兩種模型下的 Input Derived Window 模型的選 相關(guān)配 交互方 進(jìn)入命令 單個(gè)SlipstreamServer登 SlipstreamServerHA登 beeline命令 beenline指令選 SlipstreamSQL介 CREATE SHOW DESCRIBE ALTER DROP Dblink管 使用前準(zhǔn) CREATEDATABASE SHOWDATABASE DESCRIBEDATABASE DROPDATABASE Table管 CREATE 內(nèi)表與外表的區(qū) TEXT ORC ESDrive Hyperdrive Holodesk 分區(qū) 分桶 SHOW SHOW SHOWCREATE SHOW DESCRIBE ALTER 重命 修改或添加TABLE 修改或添加 修改外 TRUNCATE DROP TableColumn管 CHANGE ADD|RECE SHOW DESCRIBE ADD TEMPORARY PERMANENT SHOW DESCRIBE Stream管 CREATE Kafka為 Socket為 自定義 復(fù)雜數(shù)據(jù)類型格式的指 自定義用戶數(shù)據(jù)解 CREATESTREAMAS FILTER轉(zhuǎn) WINDOW轉(zhuǎn) SHOW DESCRIBE ALTER 重命名Input 修改InputStream屬 增加InputStream 替換InputStream DROP CREATE DESCRIBE SHOWCURRENT ALTER SHOW DROP CREATE DESCRIBE ALTER SHOW LIST DROP StreamDML注意事 導(dǎo)入數(shù)據(jù)到表 化 表化 流化 列化 DISTRIBUTE GROUP 單列GROUP GROUPBY 多列GROUP 用表達(dá)式GROUP 在GROUPBY子句中過濾: 窗 滑動(dòng)窗 跳動(dòng)窗 時(shí)間切分的其他設(shè) INFINITE SESSION 對(duì)WindowStream的聚合操 流與表的 一般 GlobalLookup 流之間的 輸出方 輸出到 輸出到 自定義輸 查詢 Slipstream權(quán)限管 角色管理語 Slipstream對(duì)象權(quán)限管理語 授予權(quán) 授予全局權(quán) 授予database級(jí)別權(quán) 授予Stream級(jí)別權(quán) 收回權(quán) 收回全局權(quán) 收回Database級(jí)別權(quán) 收回Stream級(jí)別權(quán) 查看權(quán) Slipstream操作所需權(quán)限總 Holodesk Holodesk建表語 普通建 Index和 HolodeskGlobal Hyperdrive 建 建索 創(chuàng)建索 Global Fulltext 刪除索 刪除Global 刪除Fulltext 分區(qū) 創(chuàng)建分區(qū) 查看表的分 添加分 重命名分 刪除分 清空分區(qū)中的數(shù) 流數(shù)據(jù)插入分區(qū) 分桶 建 插入數(shù) 通過EnBucketing寫數(shù) 分區(qū)表函數(shù) 分區(qū)函數(shù)介 基本操 中間指標(biāo)緩 指標(biāo)管理操 開啟流上的 禁用 嵌套SELECT需要加Ad-hoc的 數(shù)據(jù)庫名不允許以數(shù)字起 簡(jiǎn)單函數(shù)示 游標(biāo)示 運(yùn)行時(shí)管 全局管 背壓功能（Back 復(fù)雜處理 CEP使用范 場(chǎng)景1：銀 場(chǎng)景4:行 CEP的使用高可 Server 微批模式的流處理高可用 驅(qū)動(dòng)模式下的高可用 Standalone模 Job相關(guān)配 Zookeeper模 Job相關(guān)配 JobAuto- Slipstream運(yùn) 參數(shù)整 微批和驅(qū)動(dòng)模式共同參 InputStream相 Application和Job相 微批模式下參 Application和Job相 任務(wù)調(diào)度相 驅(qū)動(dòng)模式下參 Application和Job相 流上 告 History 附錄A: 基礎(chǔ)概 Topic和 Consumer和Consumer Kafka相關(guān)的簡(jiǎn)單使 創(chuàng)建 查看 修改 用ConsoleProducer發(fā)送消 用ConsoleConsumer接收消 Kafka安 權(quán)限管理示 創(chuàng)建 發(fā)送數(shù) 消費(fèi)數(shù) 客戶服 本說明書依據(jù)現(xiàn)有信息制作,其內(nèi)容更改,恕不另行通知。星環(huán)信息科技()在編寫該說明書的時(shí)候已盡最大努力保證期內(nèi)容準(zhǔn)確可靠,但星環(huán)信息科技()不對(duì)本說明遺漏、確注釋:Hadoop?和SPARK?是ApacheTM軟件在 和其他國家的商標(biāo)或的商標(biāo)。Java?是 和其他國家的商標(biāo)或的商標(biāo)。In ?2013年-2018年星環(huán)信息科技() ?星環(huán)信息科技() ,并保留對(duì)本說明書及本的最終解釋權(quán)和修改權(quán)。本說明書的版權(quán)歸星環(huán)信息科技()所有。未得到星環(huán)信息科技()的,任何人不得以任何方式或形式對(duì)本說明書內(nèi)的進(jìn)行、摘錄、備份、修改、、翻譯成其他語言、或?qū)⑵淙縏00160x-10-2018-09-Preface|在4.3版本之前，TranswarpDataHub（TDH）上的流處理應(yīng)用必須用Java或者scala開發(fā)，這將流處理應(yīng)用的開發(fā)門檻設(shè)置得非常高。從TDH4.3開始，TranswarpSlipstream允許用戶通過SQL的方式實(shí)現(xiàn)業(yè)務(wù)邏輯，大以往的流處理平臺(tái)有較高的門檻，比如SparkStreaming，Storm，必須對(duì)框架以及流處理本身，甚加迭代框架，使得原本無法利用SparkAPI進(jìn)行優(yōu)化的迭代計(jì)算效率大大提升。99%的ANSISQL2003Slipstream使用了完整的SQL編譯器,支持ANSISQL92和SQL99標(biāo)準(zhǔn),并且支持ANSISQL2003OLAP擴(kuò)展,可以滿足絕大部分現(xiàn)有業(yè)務(wù)對(duì)SQL的要求,方便應(yīng)用平滑遷移。95%的PLSQLOracle以及PLSQLDB2|1.SlipstreamSlipstream完美兼容PLSQLOracle11g以及PLSQLDB2版本，可以滿足大多數(shù)用戶的復(fù)雜業(yè)務(wù)場(chǎng)后是代碼，對(duì)一些比較的執(zhí)行邏輯生成更高效的執(zhí)行代碼或者JavaByteCode，從而保證SQL與SQL2003兼容的Window語法來創(chuàng)建滑動(dòng)窗口或者跳動(dòng)窗口，并在窗口上進(jìn)行各種復(fù)雜的聚合操作。Slipstream與ANSISQL2003比如，用戶可以使用CREATESTREAM創(chuàng)建流；使用DESCRIBESTREAM獲取流的信息；使用ALTERSTREAM修改已經(jīng)定義的流；使用DROPSTREAM刪除流；使用SHOWSTREAMS查看已Slipstream簡(jiǎn)介| 建一個(gè)Kafka執(zhí)行下面指令，運(yùn)行 ./kafka-topics.sh--zookeeper28:2181--create--topicdemo--partitionsCreatedtopic"demo".查看Kafka執(zhí)行下面指令，運(yùn)行 ./kafka-topics.sh--zookeeper28:2181--describe--topicdemoTopic:demoPartitionCount:3 ReplicationFactor:1Configs:Topic:demoPartition: Leader: Replicas:4Isr:Topic:demoPartition: Leader: Replicas:5Isr:Topic:demoPartition: Leader: Replicas:6Isr:建Kafkaproducer執(zhí)行下面指令，運(yùn)行 這里我們指定了使用28節(jié)點(diǎn)為Kafkabroker（詳細(xì)介紹見后文），并且指定了producer發(fā)布4|2beeline-u"jdbc:hive2://localhost:<port>/default"-nhive-p此處的port由TranswarpManager配置頁上的參數(shù)hive.server2.thrift.port配置，默認(rèn)為10010建一個(gè)CREATESTREAMdemo_stream(idINT,letterSTRING)ROWFORMATDELIMITEDFIELDSTERMINATEDBY查看SHOW |demo_stream 我們可以看到剛剛建好的demo_stream創(chuàng)建并觸發(fā)一個(gè)CREATETABLEdemo_table(idINT,letterINSERTINTOdemo_tableSELECT*FROM |_c0+ LIST快速|(zhì)打開瀏覽器，http://<slipstream_server_ip>:<port>，可以在Slipstream的管理界面看到當(dāng)前正在運(yùn)行的StreamJob，其中此處的port由TranswarpManager配置界面上的spark.ui.port參數(shù)配置，默認(rèn)可以看到在“ActiveStages”下有正在運(yùn)行的demo_stream SELECT*FROM+ |id|letter+ + 切換到正在運(yùn)行向demo發(fā)布數(shù)據(jù)的Kafkaproducer./kafka-console-producer.sh--broker-list28:9092--topicdemo 6|2SELECT*FROM+ |id|letter+ | | | | | | | | + SELECTCOUNT(*)FROMdemo_tableGROUPBY |_c0+| | 停止演示到此結(jié)束，用下面指令可以停止streamid為 b88a85的STOPSTREAMJOB 快速|(zhì)Slipstream1.SlipstreamSlipstream有三個(gè)的概念：Stream、StreamJob和Application。概括地說，Stream是數(shù)據(jù)流，StreamJob圖2. InputStream和DerivedStreamStream右側(cè)是運(yùn)行時(shí)概念,觸發(fā)StreamJob才真正開始接收并處理數(shù)據(jù).Stream分為兩種：InputStreamDerivedStream。直接用于接收數(shù)據(jù)源傳來的數(shù)據(jù)稱為InputStream；對(duì)已有Stream進(jìn)行變形得到的新的Stream稱為DerivedStream.8|3.SlipstreamInput直接用于接收數(shù)據(jù)源傳來的Stream稱為InputStream.InputStream定義了如何從數(shù)據(jù)源數(shù)據(jù)Derived之為對(duì)Stream的轉(zhuǎn)化.Slipstream用CVAS(CREATEVIEWASSELECT)來描述對(duì)存在的Stream進(jìn)行轉(zhuǎn)化,但在真正使用過程中,為了避免跟普通表的,還是建議采用CSAS(CREATESTREAMASSELECT)對(duì)已有的Stream進(jìn)行變形.轉(zhuǎn)化得到的新的Stream,我們稱之為DerivedStream.Window有時(shí)候我們想知道一段時(shí)間的統(tǒng)計(jì)結(jié)果,比如10秒內(nèi)某個(gè)的點(diǎn)擊量,1分鐘內(nèi)某個(gè)商品的量,這個(gè)時(shí)候我們就可以用WindowStream來匯聚這一段時(shí)間的數(shù)據(jù).WindowStream是DeriveStream的一種,是對(duì)一個(gè)stream中在一定時(shí)間窗口(window)Stream我們稱之為WindowStream.個(gè)Stream進(jìn)行窗口變形需要兩個(gè)重要的參數(shù)：Length和Slide。Length是窗口的持續(xù)時(shí)間，Slide則是執(zhí)行Stream,DerivedStream是靜態(tài)的,此時(shí)并不執(zhí)行對(duì)數(shù)據(jù)的與轉(zhuǎn)化,僅僅描述了如何對(duì)數(shù)據(jù)源傳來的數(shù)據(jù)進(jìn)行接受和變形的計(jì)劃.若要執(zhí)行這些計(jì)劃,需要相應(yīng)的Action,這就是接下來要介紹的StreamJob.要讓SlipstreamAction操作（比如將結(jié)果插入結(jié)果表，或者執(zhí)行一個(gè)臨時(shí)查詢）圖3.Application是一組業(yè)務(wù)邏輯相關(guān)的StreamJobSlipstream基礎(chǔ)|如圖Application實(shí)現(xiàn)的資源共享和所示,如果app1只允許user1控制,app2只允許user2控制.那微批模式下,Application內(nèi)使用同一個(gè)InputStream的StreamJob共組receiver例如圖Application實(shí)現(xiàn)的資源共享和中，Application1中的Streamjob2和Streamjob3都使用InputStream2，它們共享Application1中為InputStream2啟動(dòng)的Receiver。資源：不同Application中的Streamjob若使用同一個(gè)InputStream，則每個(gè)Application都為這個(gè)InputStream啟動(dòng)一組receiver。例如圖Application實(shí)現(xiàn)的資源共享和中，Application1和Application2中都有Streamjob使用InputStream2，Application1和Application2各為InputStream2啟動(dòng)一組Receiver。4.Application這里我們以SparkStreaming為代表的用編程語言(JavaorScala10|3.Slipstream5.Slipstream編程語言編寫的流應(yīng)用,一個(gè)流應(yīng)用對(duì)應(yīng)于一個(gè)獨(dú)立jvm的driver程序,函數(shù)是main函數(shù)（注：對(duì)于某些以托管方式運(yùn)行的應(yīng)用，函數(shù)可能是其他函數(shù)）。每個(gè)流應(yīng)用一般只會(huì)對(duì)應(yīng)一個(gè)AppCode(也就是實(shí)現(xiàn)業(yè)務(wù)邏輯的代碼)。每一個(gè)AppCode一般會(huì)有多個(gè)轉(zhuǎn)化過程，因此會(huì)對(duì)應(yīng)多層DStream(DiscretizedStream)。對(duì)應(yīng)到Slipstream，SparkStreaming的Driver程序，將會(huì)被Application個(gè)Application可能包含多個(gè)Stream,Stream可以包含多個(gè)DerivedStream.個(gè)APPSlipstream的DerivedStream這個(gè)概念可以跟DStream的lazyevaluation很DMLSQL持續(xù)執(zhí)行，如果按照普通SQL的方式阻塞用戶令行，那么該命令行將會(huì)被持續(xù)阻塞，無法執(zhí)行其令。行的原因，背景運(yùn)行的SQL無法跟console"insertintoresult_tableselect…"。Slipstream基礎(chǔ)|PAGE14PAGE14|4.Slipstream圖6.驅(qū)動(dòng)模型處理原InputStream流水線增加一個(gè)緩沖,每隔一段時(shí)間去取一批零件加工.7相比較微批模式,驅(qū)動(dòng)模式具有以下特點(diǎn)支持會(huì)話窗口(Session復(fù)雜的處理更加完善的高可用性兩種模型下的Input?驅(qū)動(dòng)模圖8.驅(qū)動(dòng)模式下的Input微批模式下InputStream小數(shù)據(jù)塊（Bach）來處理,Batch單位時(shí)間內(nèi)獲取的數(shù)據(jù)都放進(jìn)一個(gè)batch中，這段單位時(shí)間稱為BatchInterval，它的長度（Batch圖9.InputDerived?驅(qū)動(dòng)模Derived圖10.驅(qū)動(dòng)模式下DerivedSlipstream的兩種模型|PAGE14PAGE14|4.Slipstream11.DerivedWindow?驅(qū)動(dòng)模驅(qū)動(dòng)模式下支持秒級(jí)的window圖 驅(qū)動(dòng)模式下的Window上圖中窗口長度(Length)為4s,滑動(dòng)間隔(Slide)為3s.[t,t+4s),[t+3s,t+7s),以此類推.窗口的長度和滑動(dòng)的間隔只要是1s的整數(shù)倍即可.需要注意的是,當(dāng)t=0s,即Stream[0,3s),比窗口長度會(huì)少1秒,這是因?yàn)镾lide是執(zhí)行窗口操作的間隔,Slide時(shí)間到了就會(huì)執(zhí)行窗口計(jì)算.微批模式下,窗口長度(Length)和滑動(dòng)間隔(Slide)的長度都必須是batchduration圖13.Window上圖的窗口變形中，length是batchduration的2倍，slide是batchduration的1倍。如此圖中的第一個(gè)窗口為[t,t+4s),第二個(gè)窗口為[t+2s,t+6s),以此類推.與 的長度等于Slipstream推薦驅(qū)動(dòng)模式,TDH5.0之后默認(rèn)開啟驅(qū)動(dòng)模式ngmr.engine.mode=mapred為微批模式,ngmr.engine.mode=morphling為驅(qū)動(dòng)模式在啟動(dòng)StreamJob前執(zhí)行SETSlipstream的兩種模型|目前Slipstream提供三種交互方式：S,JDBC,ODBC。用戶可以選擇配置是否需要Kerberos驗(yàn)證。這里詳細(xì)說明S交互使用方式,JDBC和ODBC交互方式請(qǐng)參考《Inceptor使用手冊(cè)》中的《InceptorJDBC手冊(cè)》以及《InceptorODBC手冊(cè)》用server_ip|hostname來指代SlipstreamServer所在的節(jié)點(diǎn)名稱或ip。SlipstreamServer所在節(jié)點(diǎn)可以通過管理界面的Slipstream角色頁面查看。請(qǐng)參考transwarp命令行和beeline命令行。SlipstreamServer2使用Beeline作為命令行工具。單個(gè)SlipstreamServer$beeline-u$beeline-u"jdbc:hive2://<server_ip/hostname>:10010/default"-n<username>-p$beeline-u"jdbc:hive2://localhost:10010/default"-nhive-pKerberos執(zhí)行下面指令之前，當(dāng)前用戶必須有一張有效的KerberosTGT(TicketGranting$beeline-u的SlipstreamServer的principal目前5.1中ServerPrincipal中的UserName固定為hive，即ServerPrincipal的值應(yīng)該形SlipstreamServerHA如果您設(shè)置了SlipstreamServerHA，并希望使用完整的HA$beeline-u$beeline-u$beeline-u$beeline-ujdbc:hive2://hadoop1:10010,hadoop2:10010/default–nhive–p暫時(shí)未支持beeline命令行Beeline是SlipstreamServer令行工具beenline指令選項(xiàng)beeline-u<databaseURL>：指定連接的JDBCURL，$beeline-u-n<username$beeline-u"jdbc:hive://localhost:10010/default"-n-p<password>：指定連接使用的，例如$beeline-u"jdbc:hive://localhost:10010/default"-nalice-p$beeline-u"jdbc:hive://localhost:10010/default"-e"SELECT*FROM-f<file>：執(zhí)行一個(gè)--color=[true|false]false--showHeader=[true|falsetrue!quit：退出命令行（“;”結(jié)尾，為特例）addFILE[S]<filepath>addjar[s]<filepath>[<filepath><filepath>路徑中的jar包添加到Slipstream的classpathdeletejar[s]<filepath>[<filepath>：刪除添加過的jarSlipstreamSQLcreatetable…DOCVALUES意為必須選擇TRUE或者數(shù)據(jù)定義語言（DataDefinitionLanguage）CREATEDROPALTER于DATABASE,TABLE,STREAM,APPLICATION,STREAMJOB,FUNCTIONCREATECREATEDATABASE:CREATEDATABASE[IFNOTEXISTS]<database_name>[COMMENT ment>']<database_name>，SlipstreamIFNOTEXISTSCOMMENT例1.例1.創(chuàng)建一個(gè)名為 CREATEDATABASEIFNOTEXISTS SHOWSHOWDATABASES顯示所有的SlipstreamSQL介紹|SHOWDESCRIBEDESCRIBEDATABASEDESCRIBE|DESCDATABASE[EXTENDED]ALTERALTERDATABASEDBPROPERTIESowner語法：修改ALTERDATABASEALTERDATABASE SETDBPROPERTIES('date'='2015-語法：修改數(shù)據(jù)庫ALTERDATABASESETOWNER[USER|ROLE]ALTERDATABASEexchange_ SETOWNERUSERalice;ALTERDATABASEexchange_ SETOWNERUSERalice;DROPDROPDATABASE:

DROPDATABASE[IFEXISTS]<database_name>;<database_name>SlipstreamIFDROPDATABASEDROPDATABASEIFEXISTS 20|6.SlipstreamSQL驅(qū)動(dòng)程序有rxSlipstreamServer將不能使用驅(qū)動(dòng),導(dǎo)致建好的databaselink驅(qū)動(dòng)程序放好之后需重啟CREATEDATABASECREATEDATABASELINK:databaseCREATEDATABASELINK<db_link_name>CONNECTTO<user_name>IDENTIFIEDBY<password>USING<jdbc_connection_string>;db_link_name>處為databaselink創(chuàng)建databaselink如下:在數(shù)據(jù)庫創(chuàng)建database和table.以mysql為例sshroot@localhost"mysql-uroot-e'usedblink;droptableifexistsdbt1;createtabledbt1(d1tinyint,d2smallint);'"其中,<database_type><databasename>在數(shù)據(jù)庫中使用的SlipstreamSQL介紹|2222|6.SlipstreamSQL6.6.SlipstreamSQL介紹|5.Slipstream支持的數(shù)據(jù)庫的JDBCMysqljdbc:mysql://18:3306/testDB2jdbc:db2://4:50000/dbtest01PostgrepSQL在Slipstream中創(chuàng)建databaselink,mysqlCREATEDATABASELINKmysql_bob_118CONNECTTObobIDENTIFIEDBY'123456'這樣就創(chuàng)建一個(gè)到18上的databaselink,用戶名為bob,123456,數(shù)據(jù)庫為SHOWDATABASESHOWDATABASELINKS:databaseSHOWDATABASEDESCRIBEDATABASELINKdatabselinkDESCRIBEDATABASELINKDROPDATABASEDROPDATABASELINK:databaselinkDROPDATABASELINKCREATECREATETABLE:CREATE[EXTERNAL]TABLE[IFNOTEXISTS][(<column_name><data_type>[COMMENT ment>'][,<column_name><data_type>...])][COMMENT'< ment>']③[PARTITIONEDBY(<part_key><data_type>[COMMENT [,<part_key> >...])][CLUSTEREDBY(<col_name>[,<col_name>...])[SORTEDBY(<col_name>[ASC|DESC][,<col_name>[ASC|DESC]...])]INTO<num_buckets>BUCKETS][ROWFORMAT<row_format>][STOREDAS(TEXTFILE|ORC|CSVFILE)]|STOREDBY'<>'[WITHSERDEPROPERTIES(<...>)]][LOCATION'<hdfs_path>']EXTERNAL<database_name>.COMMENT指定分區(qū)鍵，具體使用方式請(qǐng)參考分區(qū)表分桶表，ASC，DESC⑦行格式，在建TEXT表時(shí)使用，具體細(xì)節(jié)請(qǐng)參考TEXT表storagehandler⑩指向HDFS上的一個(gè) 。這個(gè)選項(xiàng)我們推薦只在建外表時(shí)使用，也就是和EXTERNAL選項(xiàng)合用（雖然也 或文件的owner。如果<hdfs_path>指向的 讓<hdfs_path>指向不存在的 現(xiàn)在建表和刪表的過程中,具體如下:createtablecreateexternaldroptabledroptableTEXTPAGE24PAGE24|6.SlipstreamSQL6.6.SlipstreamSQL介紹|CREATE[EXTERNAL]TABLE<table_name>(<column_name><data_type>,<column_name><data_type>,...)[PARTITIONEDBY...]①[CLUSTEREDBY...]②[ROWFORMAT...]③[STOREDASTEXTFILE]④[LOCATION'<hdfs_path>'][TBLPROPERTIES('<property_name>'='<property_value>',...)];分區(qū)選項(xiàng)，TEXT表可以分區(qū)，本章不詳細(xì)討論如何分區(qū)，請(qǐng)參考DDL分區(qū)表分桶選項(xiàng)，TEXT表可以分桶，本章不詳細(xì)討論如何分桶，請(qǐng)參考DDL分桶表⑤指向HDFS上的一個(gè) 。這個(gè)選項(xiàng)我們推薦只在建外表時(shí)使用，也就是和EXTERNAL選項(xiàng)合用（雖然也可以在建內(nèi)表時(shí)使用，但是我們不建議這樣做）。該路徑可以是一個(gè)絕對(duì)路徑，比如/user/alice/employeeURL：hdfs://nameservice1/user/alice/employee操作的用戶必須是這個(gè)路徑指向 或文件的owner。如果<hdfs_path>指向 限，所以星環(huán)科技建議盡量避免讓<hdfs_path>指向不存在的 。表屬性，由鍵值對(duì)表示。CREATETABLEstudentCREATETABLEstudent(idINT,nameORCCREATETABLE<tableName>)STOREDASCREATETABLEstudentCREATETABLEstudent(idINT,nameSTRING)STOREDASCREATETABLE<tableName><id>STRING,)STOREDAS [WITHSHARDNUMBER [REPLICATION 第一列映射為TranswarpSearch中相映射的Index的_id，必須是STRING②簡(jiǎn)寫方式，指定表的格式為ESTranswarpSearch中相映射的Index10，建議每個(gè)shard的數(shù)據(jù)量不超過25G。建表后不可改1。建表后可改，具體語法參考修改副本數(shù)?？蛇x默認(rèn)值為dbName.tableName。esdrive_table(key1esdrive_table(key1STRING,contentSTRING,tbool)STOREDASwithshardnumber10replication1;如此就創(chuàng)建了一張名為esdrive_table的內(nèi)表，其 格式為ES，并且shard（主分片）數(shù)復(fù)雜方式在建EsdriveEsdrive內(nèi)表是在TranswarpSearch中創(chuàng)建一個(gè)全新的Index，Esdrive外表是創(chuàng)建一個(gè)TranswarpSearch中已存在Index的映射，具體建表語法如下：CREATEEXTERNALTABLE<tableName><id>)STOREDBY'io.transwarp.esdrive.ElasticSearchStorageHandler'storagehandlerSTOREDASESTranswarpSearch_idTranswarpSearch中一個(gè)已存在的表<dbName.tableName1>。CREATEEXTERNALesdrive_external_table(CREATEEXTERNALesdrive_external_table(key1STRING,ex0INT,ex3STRING)STOREDBYWITHSERDEPROPERTIES default.esdrive_external_tableTranswarpSearch中的Index_idstring數(shù)據(jù)類型隱式轉(zhuǎn)換是指數(shù)據(jù)在TranswarpSearch中的實(shí)際類型，能否成功映射為Esdrive中的內(nèi)外表中STRINGTranswarpSearch之不可以（否則會(huì)數(shù)據(jù)丟失）數(shù)字類型不能轉(zhuǎn)換為字符類型，反之也不可以（查詢結(jié)果有問題）Esdrive所支持的數(shù)據(jù)類型及其在TranswarpSearchHyperdriveCREATETABLE<tableName><key>)STOREDAS66.SlipstreamSQL介紹|2828|6.SlipstreamSQLCREATETABLE<tableName><key><column1><column2>)STOREDBY'io.transwarp.hyperdrive.HyperdriveStorageHandler'①指定格式。建Hyperdrive表的較復(fù)雜寫法Hyperdrive內(nèi)表的每一列在底層HBaseHolodeskCREATETABLE<holodesk_table_name>)STOREDAS10.HolodeskholodeskEmployee(IDINT,RegionSTRING,SalaryDECIMAL)STOREDASCREATE[EXTERNAL]TABLE)PARTITIONEDBY(<partition_key><data_type>,...)[CLUSTEREDBY...]CREATE[EXTERNAL]TABLE[IFNOT<column1><column2>)[PARTITIONEDBY...]CLUSTEREDBY(<col_name>)[SORTEDBY(<col_name>[ASC|DESC][,<col_name>INTO<num_buckets><num_buckets>分桶鍵時(shí)，不需要指定分桶鍵的數(shù)據(jù)類型。SHOWSHOWTABLES:SHOWTABLES[IN<database_name>][LIKE如果想要查看非當(dāng)前數(shù)據(jù)庫中的表和流，可以使用[IN<database_name>]來指定數(shù)據(jù)庫。使用[LIKESHOWSHOWTBLPROPERTIES:SHOWTBLPROPERTIESSHOWTBLPROPERTIESSHOWTBLPROPERTIESTBLPROPERTIES。如果只要查看某個(gè)屬性，可以在表名后面的括號(hào)中指SHOWCREATESHOWCREATETABLE:SHOWCREATETABLESHOWSHOWPARTITIONS:showpartitions66.SlipstreamSQL介紹|3030|6.SlipstreamSQLDESCRIBEDESCRIBETABLE:DESCRIBEEXTENDEDDESCRIBEFORMATTEDALTERALTERTABLE<table_name>RENAMETO修改或添加TABLEALTERTABLE<table_name>SETTBLPROPERTIES('<property_name>'='<property_value>'...修改或添加ALTERTABLE<table_name>SETSERDEPROPERTIES('<property_name>'='<property_value>'...修改外表ALTERTABLE<table_name>SETLOCATION將外表指向的 改為<new_location>。注意，執(zhí)行該操作的用戶必須是<new_location>的ownerTRUNCATETRUNCATETABLEtable_name[PARTITION(<partition_key>=<partition_value>[,...][PARTITION(<partition_key><partition_value>[,TRUNCATETABLETRUNCATETABLE12.partitionTRUNCATETRUNCATETABLEuser_infoPARTITIONDROPDROPTABLE:DROPTABLE[IFEXISTS]TableColumnCHANGECHANGECOLUMN:ALTERTABLEtable_nameCHANGE[COLUMN]col_old_namecol_new_namecolumn_type ment][AFTERcolumn_name][CASCADE|RESTRICT];息。上面指令可以對(duì)一列的列名、列的數(shù)值類型、列在表中的位置、列的注解或者這些的任意組合做出修CREATETABLECREATETABLEtest_change(aINT,bINT,c--將列a重命名為列a1ALTERTABLEtest_changeCHANGEaa1INT;--將列a1重命名為列 ALTERTABLEtest_changeCHANGEa1a2STRINGAFTERb;--表test_change的新結(jié)構(gòu)為：(bINT,a2STRING,cADD|RECEADDCOLUMNS：在表中加入新的列RECECOLUMNS：將原來的表中列結(jié)構(gòu)替換為新的列結(jié)ALTERTABLEtable_name CECOLUMNS(col_namedata_type ment],...)--CASCADE表示將修改字段元信息的影響擴(kuò)展至與其相關(guān)的分區(qū)；而RESTRICTADDADDCOLUMNS66.SlipstreamSQL介紹|3232|6.SlipstreamSQL14CREATECREATETABLEtest_change(aINT,bINT,c在表test_change中添加一個(gè)新列d類型為ALTERTABLEtest_changeADDCOLUMNS(dRECERECECOLUMNS將所有已有的列刪除然后加入新指定的列。只能對(duì)使用Slipstream自帶SerDe(DynamicSerDe,MetadataTypedColumnsetSerDe,LazySimpleSerDeandColumnarSerDe)用RECECOLUMNS。RECECOLUMNS也可以被用于刪除列15CREATECREATETABLEtest_change(aINT,bINT,c--將test_change中的列由(aINT,bINT,cINT)替換為(aint,bint)起到將列cALTERTABLE CECOLUMNS(aINT,b只有TEXT、CSV、基于定寬文本文件外表這三種表支持對(duì)字段類型SHOWSHOWCOLUMNS:SHOWCOLUMNSFROM|IN<table_name>[FROM|INFROMINDESCRIBEFunction本節(jié)介紹如何使用SlipstreamSQLADD在介紹如何使用functionjar包到Slipstream.ADDJARADDJAR[S]

Jar 通過ADDJAR[S]添加的jar包不能有重名的類JarLIST file | TEMPORARY

CREATETEMPORARYFUNCTION<function_name>AS任意一個(gè)在classpath中的class。用戶也可以通過ADDJAR向classpath加jar包.DROPTEMPORARYFUNCTION[IFEXISTS]DROPTEMPORARYFUNCTION[IFEXISTS]PERMANENTCREATEPERMANENTFUNCTION[<db_name>.]<function_name>AS<class_name>[USINGJAR|FILE|ARCHIVE'<file_uri>'[,JAR|FILE|ARCHIVE'<file_uri>']];重新建臨時(shí)函數(shù)。需要加入環(huán)境的jar包，文件或者可以通過USING指定。第一次使用該函數(shù)時(shí)，ADDJAR/FILESlipstream不在localmode，那么資源的地址也db_namedb_name.function_namefunction_nameDROPPERMANENTFUNCTION[IFEXISTS]SHOW6.6.SlipstreamSQL介紹|PAGE34PAGE34|6.SlipstreamSQLSHOWFUNCTIONS[<regex>]<regex>DESCRIBEDESCRIBEFUNCTION<functionCREATECREATESTREAM用于建一個(gè)InputStream

語法：創(chuàng)建InputCREATESTREAM<stream_name>(<column><data_type>,<column><data_type>,[ROWFORMATDELIMITEDFIELDSTERMINATEDBY'<delimiter>'COLLECTIONITEMSTERMINATED'<delimiter2>'MAPKEYSTERMINATEDBY'<delimiter3>']TBLPROPERTIES(["topic"="<topic_name>",]["source"="kafka",]③["kafka.zookeeper"="<ip:port>",]④["tocol"="<protocol>",]⑦["transwarp.consumer.sasl.jaas.config"="jass_string",]⑦["batchduration"="<int>",]⑧["<key>"="<value>",]⑨t(yī)opic_name>指定inputstream指向的Kafkatopic，默認(rèn)值是database_name.table_namesource>指定數(shù)據(jù)源來源，默認(rèn)為kafka⑤<ip:port>指定集群中的任意一個(gè)Kafkabroker的ip和端，默認(rèn)為9092可選項(xiàng)。從Topic中指定的partitionpartitionskafka.broker.list⑧整數(shù)，設(shè)置batch的長度，單位為毫秒。該屬性只能在時(shí)間模式下使用，且只對(duì)微批模式的流處理適用。TBLPROPERTIESStreamJob啟動(dòng)Kafkaconsumer使用的屬性值。property_key為Kafkaconsumer所支持的某個(gè)配置屬定 "當(dāng)stream作為輸出流時(shí)，可通過此配置設(shè)定StreamJob啟動(dòng)Kafkaproducer時(shí)使用的屬值。property_key為Kafkaproducer所支持的某個(gè)配置屬性名，例如，想配置producer發(fā)送消息的ackSlipstream支持ApacheKafka作為數(shù)據(jù)源，如果您需要了解關(guān)于ApacheKafka的概念和操作，請(qǐng)參考Kafka基礎(chǔ)。需要注意的是，從4.8開始，Slipstream只支持Kafka0.10.2以上的版本。ApacheKafka作為Slipstream數(shù)據(jù)源時(shí)，發(fā)布給Kafkatopic的消息通過InputStream進(jìn)入Slipstream。一個(gè)InputStream

CREATESTREAMemployee_streamCREATESTREAMemployee_stream(idINT,nameBOOLEAN)ROWFORMATDELIMITEDTERMINATEDBY","TBLPROPERTIES("kafka.zookeeper"="tw- CREATESTREAMdemo(idCREATESTREAMdemo(idINT,letterSTRING)ROWFORMATDELIMITEDFIELDSTERMINATEDBYCREATESTREAMdemo(nCREATESTREAMdemo(nSTRING,vINT)相關(guān)操作請(qǐng)參考Kafka安全手動(dòng)在每個(gè)運(yùn)行Executor的服務(wù)器上的相同 tocol值為SASL_ INTEXT，設(shè)置屬性為Kafkabroker上配置了的相應(yīng)的值，設(shè)置屬性transwarp.consumer.sasl.jaas.config為相應(yīng)的jaas字符串。確保自己對(duì)作為源的KafkatopicCREATESTREAMdemo(idCREATESTREAMdemo(idINT,letterROWFORMATDELIMITEDFIELDSTERMINATEDBY requireduseKeyTab=truestoreKey=truekeyTab=\"/etc/slipstream1/conf/kafka.keytab\"手動(dòng)在每個(gè)運(yùn)行Executor的服務(wù)器上的相同 tocol值為SASL_INTEXT，設(shè)置屬性為Kafkabroker上配置了的相應(yīng)的值，設(shè)置屬性ducer.sasl.jaas.config為相應(yīng)的jaas字符串。確保對(duì)作為自己寫入對(duì)象的KafkatopicCREATESTREAMoutput(idINT,letterSTRING)CREATESTREAMoutput(idINT,letterSTRING)ROWFORMATDELIMITEDFIELDSTERMINATEDBY',' CREATESTREAMsocket_stream(idINT,name BOOLEAN)創(chuàng)建Socket為數(shù)據(jù)源的表，數(shù)據(jù)源是2，端為8888RabbitmqCREATESTREAMrabbitmq_stream(idINT,nameSTRING)ROWFORMATDELIMITEDFIELDSTERMINATEDBY','以上創(chuàng)建了Rabbitmq為數(shù)據(jù)源的stream,rabbitmq所在的服務(wù)器為28,rabbitmq的用戶名密碼為streamuser/psw,從writerqueue隊(duì)列中數(shù)據(jù),開啟autoack(數(shù)據(jù)接收后會(huì)從rabbitmq服務(wù)器端刪3636|6.SlipstreamSQL6.6.SlipstreamSQL介紹|1.rabbitmq相關(guān)TBLPROPERTIES說明數(shù)據(jù)來源,此處為 gth.html）。輸入格式為1mqCREATESTREAM_stream_pull(idSTRING,numINT)ROWFORMATDELIMITEDFIELDSTERMINATEDBY','""mq.consumergroup"="slipstreampull","server"="37:9876","mq.tags"="test","CREATESTREAM_stream_push(idstring,numint)ROWFORMATDELIMITEDFIELDSTERMINATEDBY','""mq.consumergroup"="slipstreampush","server"="37:9876","mq.tags"="test");表2.MQ通用配消費(fèi)的消費(fèi)的tags,例如：TAGA||||nameserver的地 表3.MQPUSH模式配表4.MQPULL模式配在啟動(dòng)streamjob前，應(yīng)該先在mqnameserver上topic，不然會(huì)出從TDH5.1開始,在驅(qū)動(dòng)模式下支持用戶自定義輸入源.添加自定義輸入源步驟如下繼承io.transwarp.slipstream.api.source.CustomDataSourcedefinit(id:Int,params:Properties):non-returnnullifgetdefpoll():Array[Byte]defclose():Unitimportjava.io.IOExceptionimportimportjava.util.concurrent.{TimeUnit,ArrayBlockingQueue}importcom.rabbitmq.._objectConstantvalBLOCKING_QUEUE_SIZE=valHOST="host"valPORT=valUSERNAME="username"valPASSWORD="password"valQUEUE="queue"valDURABLE=valEXCLUSIVE="exclusive"valAUTODELETE="autodelete"valAUTOACK="autoack"valDEFAULT_POLL_TIMEOUT:Int=}valmessageBlockingQueue=newArrayBlockingQueue[Array[Byte]](Constant.BLOCKING_QUEUE_SIZE)varconnection:Connection=_varchannel:Channel=overridedefinit(id:Int,params:Properties):Unit{connection=getConnection(params)channel=getChannel(connection:Connection)configureChannel(channel,params)}overridedefclose():Unit{channel.close()}non-38|6.SlipstreamSQL*returnnullifgetoverridedefpoll():Array[Byte]}defgetConnection(params:Properties):Connection{valfactory:ConnectionFactory=newConnectionFactoryif(params.containsKey("host")){}}{}{}{}}{}}valconnection:Connection=factory.newConnection}defgetChannel(connection:Connection):Channel=connection.createChanneldefconfigureChannel(channel:Channel,params:Properties):Unit={valqueue=valdurable=if{}{}valexclusive=if{}{}valautodelete=if{}{}}defstartConsume(channel:Channel,messageBlockingQueue:ArrayBlockingQueue[Array[Byte]],params:Properties):Unit={valconsumer:Consumer=new{overridedefhandleDelivery(consumerTag:String,envelope:Envelope,properties:AMQP.BasicProperties,body:Array[Byte]){try}catchcasee:InterruptedException=>System.out.println("ERRORwhenputmessageintoblockingqueue",}}}}valqueue=valautoack=if{}{}}}將自定義源打成Jar包添加到SlipstreamServerJar包方式請(qǐng)?jiān)斠夾DDJAR創(chuàng)建6.SlipstreamSQL介紹|PAGE42PAGE42|6.SlipstreamSQL66.SlipstreamSQL介紹|CREATESTREAM<stream)ROWFORMATDELIMITEDFIELDSTERMINATEDBY','"source"="custom",①"slipstream.datasource.classpath"="<classpath>"②["slipstream.<key>"="<value>"]);③slipstream.以第一步的自定義輸入流為例創(chuàng)建CREATECREATESTREAMcustom_stream(idstring,namestring,numint)ROWFORMATDELIMITEDFIELDSTERMINATEDBY','TBLPROPERTIES復(fù)雜類型CREATESTREAMstruct_stream(idINT,s1STRUCT<a:INT,b:STRING>)ROWFORMATDELIMITEDFIELDSTERMINATEDBY','COLLECTIONITEMSTERMINATEDBY'|'復(fù)雜類型CREATESTREAMarray_stream(idINT,a1ARRAY<STRING>)ROWFORMATDELIMITEDFIELDSTERMINATEDBY','COLLECTIONITEMSTERMINATEDBY'|'TBLPROPERTIES(復(fù)雜類型CREATESTREAMmap_stream(idINT,m1MAP<STRING,INT>)ROWFORMATDELIMITEDFIELDSTERMINATEDBY','COLLECTIONITEMSTERMINATEDBY'|'MAPKEYSTERMINATEDBY':'TBLPROPERTIES(復(fù)雜類型CREATESTREAMstudent_stream(idINT,t1TIMESTAMP)1,2015-01-152,2077-11-15io.transwarp.slipstream.api.decoder.ArrayDecoder實(shí)現(xiàn)arrayFromBytes接口來自定義數(shù)據(jù)解21.kafkaimportkafka.utils.VerifiablePropertiesclassExampleDecoder(props:VerifiableProperties=null)extendsArrayDecoder[Array[Byte]](props){overridedefarrayFromBytes(bytes:Array[Byte]):Array[Array[Byte]]{valsep:Byte=';'}}使用自定義數(shù)據(jù)解析構(gòu)建ADDJARCREATESTREAMemployee_stream(idINT,nameSTRING) CREATESTREAMAS使用現(xiàn)有的Stream和Table可以創(chuàng)建DerivedStream。在Slipstream中，DerivedStream使用CREATESTREAMASSELECT創(chuàng)建。CREATESTREAM<stream_name>ASSELECTFILTERCREATESTREAMfemale_name_streamASSELECTnameFROMemployee_stream WINDOWCREATESTREAMemployee_window_streamASSELECT*FROMemployee_streamSTREAMWINDOWw1AS(length'12'secondslide'4'second);6.6.SlipstreamSQL介紹|PAGE46PAGE46|6.SlipstreamSQLSHOW顯示所有的Streams,包括InputStream和DerivedSHOWDESCRIBEDESCRIBE|DESC[EXTENDED|FORMATTED]<stream_name>|<view_name>ALTERALTERSTREAM用于修改InputStream重命名InputALTERSTREAM<old_name>RENAMETO修改InputStreamALTERSTREAM<stream_name>SETTBLPROPERTIES增加InputStreamALTERSTREAM<stream_name>ADDCOLUMNS(nametype[,name1type1替換InputStreamALTERSTREAM<stream_name>RECECOLUMNS(nametype[,name1type1DROP刪除一個(gè)InputStream（也就是以CREATESTREAM創(chuàng)建的DROPSTREAM務(wù)邏輯編寫。而當(dāng)業(yè)務(wù)需要上線時(shí)，可以用一個(gè)prodid，具有嚴(yán)格的權(quán)限控制，并對(duì)應(yīng)一個(gè)Application。CREATECREATEAPPLICATION<app_name>WITHDESCRIBEDESCRIBE[|DESC]APPLICATIONSHOWCURRENTSHOWCURRENT顯示當(dāng)前所在的ApplicationALTERALTERAPPLICATION<app_name>SETSHOWSHOWDROPDROPAPPLICATION<app-StreamJob是觸發(fā)SlipstreamAction，一般具有插入結(jié)果表語義，或者Ad-hoc查詢語義。StreamJob主要StreamJoblevel的參數(shù)配置，以及對(duì)應(yīng)SQL。CREATECREATESTREAMJOB<streamjob_name>as("<query>")創(chuàng)建一個(gè)新的 DESCRIBEDESCRIBE[|DESC]STREAMJOBALTERALTERSTREAMJOB<streamjob_name>SETSHOWSHOW顯示所有持久化的streamjob，也就是通過CREATESTREAMJOB創(chuàng)建的streamjob的streamjob在運(yùn)行時(shí)可以通過LISTSTREAMJOB查看（見[runtime]）；未持久化的streamjob運(yùn)LISTLISTDROPDROPSTREAMJOB<streamjob-刪除指定的StreamJobStreamDMLwindowstreamjoin,支持流與表的支持mapjoin和globallookup流與流的Join要求join的兩個(gè)流必須是windowstream,且window的slide(UPDATEDELETEloaddatalocalinpath"<datapath>"intotable<tablename>;loaddataloaddatalocalinpath"/tmp/fact.txt"intotableSELECTselect_expression,select_expression,...FROMtable_reference[AS]table_alias;table_referenceCREATESTREAMASSELECTstream_alias.column_name1,stream_alias.column_name2,...FROMstream_reference[AS]stream_alias;SELECTselect_expression[AS]col_alias,select_expression[AS]col_alias,...FROMtable_referenceDISTRIBUTEDISTRIBUTEBY可以將流中的數(shù)據(jù)按BY后的字段重新分配到同一個(gè)partitionCREATESTREAM<streamname>asselect[column1,column2,...]from<inputstreamname>DISTRIBUTE<column例23.CREATECREATESTREAMraw_stream(idINT,nameBOOLEAN)ROWFORMATDELIMITEDCREATESTREAMderive_streamASSELECT*fromraw_streamDISTRIBUTE PAGE52PAGE52|6.SlipstreamSQL6.6.SlipstreamSQL介紹|GROUPGROUPBY子句將查詢結(jié)果按列值并組，也就是指定列列值相同的將被并入同組。GROUPBY常常與聚合函數(shù)合使用GROUPBY時(shí)，SELECT語句和GROUPBY本章節(jié)出現(xiàn)的示例只選取SELECTCREATE<streamname>ASSELECTINSERTINTO<table>SELECTSELECTselect_expression,select_expression,...GROUPBYgroupby_expression[,groupby_expression,...]單列GROUP單列GROUPBY就是GROUPBY對(duì)windowstream用GROUPBY對(duì)windowstream用GROUPBYSELECTname,COUNT(*)AScntFROMwindow_stream_demoGROUPBYname;GROUPBYSELECTselect_expression1,select_expression2,...GROUPBYgroupby_expression[,Slipstream中支持groupby1），groupby2（第二列）這類用法，其中數(shù)字1是25.groupby將將單列GroupBy示例中的name換成SELECTname,COUNT(*)AScntFROMwindow_stream_demoGROUPBY1;SELECTCOUNT(*)AScnt,nameFROMwindow_stream_demoGROUPBY2;多列GROUP多列GROUPBY就是GROUPBYSELECTname, ,COUNT(*)SELECTname, ,COUNT(*)AScntFROMwindow_stream_demoGROUPBY 用表達(dá)式GROUP

SELECTSELECTtdh_todate(trans_time,'yyyyMMddHHmmss','MM')ASdate,COUNT(trans_id)AScntFROMGROUPBYtdh_todate(trans_time,'yyyyMMddHHmmss',在GROUPBY子句中過濾:如果過濾條件受帶GROUPBY的查詢結(jié)果影響，那么就不能用WHERE子句來過濾，而要用HAVINGSELECTselect_expression,select_expression,...FROMtable_nameGROUPBYgroupby_expression,group_expressionHAVINGhaving_expressionGROUPBY…SELECTacc_num,MAX(price*amount)ASmax_valueFROMwindow_stream_demoGROUPBYacc_numHAVING在WHERE子句中不能有聚合函數(shù)，因?yàn)镾lipstream在執(zhí)行GROUPBY子句之前就會(huì)執(zhí)行SELECTacc_num,MAX(price*amount)SELECTacc_num,MAX(pri