阿里大數(shù)據(jù)計(jì)算服務(wù)MaxCompute-流式計(jì)算

1、大數(shù)據(jù)計(jì)算服務(wù)MaxCompute流式計(jì)算大數(shù)據(jù)計(jì)算服務(wù)MaxCompute/流式計(jì)算 PAGE 9流式計(jì)算簡(jiǎn)介Stream SQL是 MaxCompute 提供的一種完全托管的分布式數(shù)據(jù)流式處理服務(wù)。該功能底層采用先進(jìn)的分布式增量計(jì)算框架，可以實(shí)現(xiàn)低延遲響應(yīng)，以SQL的形式提供流式計(jì)算服務(wù)，并且完全屏蔽了流式計(jì)算中復(fù)雜的故 障恢復(fù)等技術(shù)細(xì)節(jié)，極大的提高了開(kāi)發(fā)效率。備注：目前Stream SQL已處于維護(hù)狀態(tài)，不再接入新用戶(hù)。后續(xù)Stream SQL會(huì)成為一款阿里云的獨(dú)立產(chǎn)品，相關(guān)信息請(qǐng)關(guān)注阿里云官方通告。功能特點(diǎn)低延時(shí): 從數(shù)據(jù)寫(xiě)入到計(jì)算出結(jié)果秒級(jí)別的延遲;高可靠: 底層的體系架構(gòu)充分考慮了

2、單節(jié)點(diǎn)失效后的故障恢復(fù)等問(wèn)題，可以保證數(shù)據(jù)在處理過(guò)程中的不重不丟。可擴(kuò)展: 在數(shù)據(jù)量增加時(shí)，用戶(hù)可以通過(guò)簡(jiǎn)單的增加Worker節(jié)點(diǎn)數(shù)量的方式進(jìn)行水平擴(kuò)展，可以支持每天PB級(jí)別的數(shù)據(jù)流量。開(kāi)發(fā)方便: 使用標(biāo)準(zhǔn)的SQL描述流式計(jì)算的過(guò)程，隱藏了底層的復(fù)雜技術(shù)架構(gòu)，極大的提升了開(kāi)發(fā)效率。典型應(yīng)用流式數(shù)據(jù)通常會(huì)采集到消息隊(duì)列中，再由流式計(jì)算的引擎從消息隊(duì)列中訂閱數(shù)據(jù)。在 MaxCompute 中將這種模式進(jìn)行了簡(jiǎn)化，可以將表作為流式數(shù)據(jù)載體，這種表在 MaxCompute 中被稱(chēng)為HubTable，用戶(hù)實(shí)時(shí)上傳到HubTable中的數(shù)據(jù)可以被流式計(jì)算引擎訂閱使用，并且同時(shí)會(huì)被寫(xiě)到離線集群中供離線計(jì)算引

3、擎使用。用戶(hù)同 樣可以開(kāi)發(fā)應(yīng)用從HubTable中訂閱數(shù)據(jù)。因此在Stream SQL中，HubTable首先是流計(jì)算引擎的數(shù)據(jù)源，同時(shí)也可以作為流式計(jì)算的輸出。更多有關(guān)HubTable的介紹請(qǐng)參考 Data Hub ServiceStream SQL的源頭數(shù)據(jù)來(lái)自于DataHub Service提供的實(shí)時(shí)上傳服務(wù)，用戶(hù)可以創(chuàng)建一張表作為實(shí)時(shí)上傳的目標(biāo)表(稱(chēng)為HubTable)，數(shù)據(jù)在上傳后會(huì)在短時(shí)間內(nèi)復(fù)制到到離線集群中，并且還可以提供類(lèi)似消息隊(duì)列的發(fā)布/訂閱服務(wù)(Pub/Sub)，數(shù)據(jù)是以一個(gè)個(gè)小的批量模式上傳，延時(shí)很小。Stream SQL就通過(guò)訂閱數(shù)據(jù)的方式進(jìn)行增量的計(jì)算。關(guān)于HubTa

4、ble的說(shuō)明，請(qǐng)參考DataHub。流式計(jì)算的結(jié)果可以直接寫(xiě)入HubTable，通過(guò)外部的訂閱服務(wù)更新到目標(biāo)庫(kù)中。實(shí)現(xiàn)一個(gè)簡(jiǎn)單的流計(jì)算的步驟 如下:CREATE TABLE stream_in (.);- 指定該表可以允許實(shí)時(shí)數(shù)據(jù)上傳ALTER TABLE stream_in SET HUBLIFECYCLE 2;- 創(chuàng)建另一張流式計(jì)算的輸出結(jié)果表CREATE TABLE stream_out (.);ALTER TABLE stream_out SET HUBLIFECYCLE 2;創(chuàng)建一張數(shù)據(jù)上傳的目標(biāo)表, 并且指定該表可以實(shí)時(shí)上傳數(shù)據(jù)。Grant all on table to user

5、 HYPERLINK mailto:odps odps給ODPS賬號(hào)授權(quán)在流式計(jì)算過(guò)程中，Stream SQL需要訪問(wèn)DataHub服務(wù)中得表數(shù)據(jù)。因此，需要用戶(hù)顯示授權(quán)該ODPS賬號(hào)以讀權(quán)限。在后續(xù)的改進(jìn)中，我們將取消此次授權(quán)操作。用戶(hù)可以直接通過(guò)Stream SQL訪問(wèn)DataHub表數(shù)據(jù)，并滿(mǎn)足安全需求。CREATE STREAMJOB testjob AS INSERT INTO table stream_out SELECT COUNT(*) FROM stream_in;END STREAMJOB;創(chuàng)建一個(gè)流計(jì)算的任務(wù)，該任務(wù)引用stream_in表作為數(shù)據(jù)源，并且將結(jié)果寫(xiě)入str

6、eam_out表中。在ODPS客戶(hù)端窗口中執(zhí)上面的流計(jì)算任務(wù)，系統(tǒng)會(huì)建立一個(gè)流計(jì)算的任務(wù)，一旦用戶(hù)通過(guò)stream_in表實(shí)時(shí)的上傳數(shù)據(jù)，就會(huì)觸發(fā)流計(jì)算的過(guò)程，并且將結(jié)果增量的寫(xiě)入到stream_out中。調(diào)用DataHub SDK將數(shù)據(jù)上傳到stream_in表中。select * from stream_out;隨著數(shù)據(jù)的上傳，在ODPS客戶(hù)端窗口中執(zhí)行若干次。會(huì)發(fā)現(xiàn)有一系列逐漸累積增長(zhǎng)的匯總數(shù)據(jù)?；靖拍钤?MaxCompute 中流計(jì)算的邏輯是通過(guò)SQL定義的，稱(chēng)為StreamJob。streamJob通過(guò)SQL引用的HubTable讀取實(shí)時(shí)數(shù)據(jù)，并且將結(jié)果寫(xiě)入結(jié)果表中。此外，還可以維

7、表和臨時(shí)表。詳細(xì)說(shuō)明如下：源表Stream Job中引用的源表是HubTable，用戶(hù)必須將數(shù)據(jù)通過(guò)Data Hub Service實(shí)時(shí)上傳到ODPS中。維表維表中的內(nèi)容在運(yùn)行時(shí)會(huì)由系統(tǒng)加載到內(nèi)部緩沖區(qū)，并且可以和流式數(shù)據(jù)進(jìn)行Join運(yùn)算。維表的定義只在該StreamJob中有效。用戶(hù)利用維表可以從離線的表中加載數(shù)據(jù)。維表中的內(nèi)容只被引用，不能在StreamJob中 更改。在一個(gè)StreamJob中定義的維表數(shù)據(jù)不超過(guò)5個(gè)，內(nèi)存不超過(guò)3G。臨時(shí)表如果計(jì)算的邏輯比較復(fù)雜，用一個(gè)SQL難以描述，可以通過(guò)定義臨時(shí)表的方式來(lái)簡(jiǎn)化開(kāi)發(fā)的過(guò)程。臨時(shí)表的定 義只在該streamJob中有效。結(jié)果表流計(jì)算的結(jié)

8、果可以寫(xiě)入HubTable中。作業(yè)運(yùn)行StreamJob對(duì)象在創(chuàng)建后會(huì)長(zhǎng)期運(yùn)行，直到用戶(hù)將該任務(wù)停止。用戶(hù)可以通過(guò)命令將流計(jì)算作業(yè)暫停、恢復(fù)或 者徹底的刪除。StreamJob操作創(chuàng)建StreamJobCREATE STREAMJOB jobname AS聲明維表 聲明臨時(shí)表 聲明結(jié)果表- 通過(guò)Stream SQL定義計(jì)算邏輯END STREAMJOB;通過(guò)客戶(hù)端創(chuàng)建一個(gè)StreamJob的命令格式如下，在StreamJob中包括維表、臨時(shí)表、結(jié)果表的聲明，以及通 過(guò)Stream SQL定義的計(jì)算邏輯，其中維表、臨時(shí)表及結(jié)果表的聲明都是可選項(xiàng)。說(shuō)明：參數(shù)value格式含義odps.stream

9、job.start.timeyyyyMMddhhmmss新提交的Stream作業(yè)從該時(shí)間點(diǎn)開(kāi)始讀取數(shù)據(jù)，默認(rèn)從當(dāng)天的0點(diǎn)0分開(kāi)始o(jì)dps.streamjob.timeout整型字符串，單位秒一批數(shù)據(jù)的處理超時(shí)時(shí)間。如果一批數(shù)據(jù)的處理時(shí)間超過(guò)timeout即認(rèn)為數(shù)據(jù)處理失敗（即使最終處理成功了），會(huì)觸發(fā)數(shù)據(jù)的重新計(jì)算。Timeout時(shí)間設(shè)置太小導(dǎo)致誤判處理失敗的概率增大，設(shè)置太大會(huì)導(dǎo)致發(fā)現(xiàn)系統(tǒng)故障進(jìn)行failover的時(shí)間變長(zhǎng)，默認(rèn)180秒odps.streamjob.worker.num整型字符串指定作業(yè)使用的worker數(shù)目。默認(rèn)通過(guò)DataHub的shard數(shù)進(jìn)行計(jì)算，公式：min(max

10、(shard num, 10),100)1. 在創(chuàng)建StreamJob之前可以通過(guò)set命令指定運(yùn)行時(shí)的可選參數(shù)：用戶(hù)創(chuàng)建StreamJob時(shí)必須有CreateInstance權(quán)限。對(duì)StreamJob中引用的HubTable必須有讀權(quán)限。如果目標(biāo)表是HubTable，必須有更新數(shù)據(jù)的權(quán)限。備注：有關(guān)權(quán)限介紹請(qǐng)參考ACL安全授權(quán)。CREATE STREAMJOB cal_trans_amt AS INSERT INTO table transsumSELECT SUM(amt)FROM transaction;END STREAMJOB;例如，交易的數(shù)據(jù)通過(guò)transaction表實(shí)時(shí)上傳，可

11、以通過(guò)以下方式創(chuàng)建StreamJob統(tǒng)計(jì)交易的總金額，并且將 結(jié)果寫(xiě)入另一個(gè)Hubtable中。羅列StreamJob命令格式:list StreamJobs;list StreamJobs;Name Status StartTime OwnerMyJob1 Running 2015-04-01 13:00:00 HYPERLINK mailto:ALIYUN%24yunyuan ALIYUN$yunyuan說(shuō)明：列出當(dāng)前項(xiàng)目下所有正在運(yùn)行的StreamJob。示例:顯示StreamJob狀態(tài)status StreamJob jobname;命令格式：說(shuō)明：顯示指定的StreamJob運(yùn)行狀態(tài)

12、。status StreamJob/jobname; Status: Running StartTime:2015-04-01 13:00:00ComponentsName InputRecords OutputRecords FailedRecords Latency(ms) J1_MRM_M1 100 100 0 1Streamjob級(jí)別的數(shù)據(jù)項(xiàng)定義：Status：當(dāng)前StreamJob的狀態(tài)， Running或PausedStartTime：StreamJob開(kāi)始運(yùn)行的時(shí)間Component級(jí)別的數(shù)據(jù)項(xiàng)定義Name：該component的名稱(chēng)InputRecords：該作業(yè)在對(duì)應(yīng)的時(shí)間窗

13、口內(nèi)處理的輸入數(shù)據(jù)行數(shù)。OutputRecords：該作業(yè)在對(duì)應(yīng)的時(shí)間窗口內(nèi)輸出的數(shù)據(jù)行數(shù)。FailedRecords：該作業(yè)在對(duì)應(yīng)的時(shí)間窗口內(nèi)觸發(fā)異常的數(shù)據(jù)行數(shù)。Latency(s)：一條記錄自Component的輸入開(kāi)始直至輸出所經(jīng)歷的平均時(shí)間，包括等待延時(shí)。暫停streamjobpause streamjob jobname;命令格式：說(shuō)明：將指定的streamjob置為暫停狀態(tài)，作業(yè)會(huì)被掛起，不再進(jìn)行計(jì)算。作業(yè)的元數(shù)據(jù)和之前的數(shù)據(jù)狀態(tài)仍 然保存著，可以通過(guò)resume命令恢復(fù)作業(yè)?；謴?fù)streamjobresume streamjob jobname;命令格式：說(shuō)明：將已暫停的stre

14、amjob恢復(fù)運(yùn)行。在恢復(fù)運(yùn)行時(shí)系統(tǒng)會(huì)讀取參數(shù)，通過(guò)這種方式可以調(diào)整在恢復(fù)運(yùn)行后 的流計(jì)算作業(yè)參數(shù)。停止streamjobdelete streamjob jobname;命令格式：說(shuō)明：停止指定的streamjob，作業(yè)的元數(shù)據(jù)和之前運(yùn)行的數(shù)據(jù)和狀態(tài)會(huì)被刪除。在停止前必須先將作業(yè)置為 暫停狀態(tài)以防止誤操作。Stream SQL語(yǔ)法Stream SQL的語(yǔ)法與離線SQL的語(yǔ)法相似，但并不完全相同。很多在離線SQL中支持的功能在Stream SQL中并不支持。數(shù)據(jù)類(lèi)型在Stream SQL中，暫時(shí)僅支持Bigint, Double和String三種類(lèi)型。DDL操作維表維表在流計(jì)算任務(wù)開(kāi)始時(shí)加載一

15、次，在一個(gè)流計(jì)算任務(wù)中，維表的個(gè)數(shù)不超過(guò)3個(gè)，內(nèi)存限制為100M，超出限 制時(shí)拋異常。CREATE DIMENSION TABLE tablename(語(yǔ)法：col_name data_type comment;)LOCATION URI;說(shuō)明：維表支持ODPS維表, ODPS維表只允許一個(gè)分區(qū)。不支持View對(duì)象schema:/authority/path?query#fragmentURI指定維表的來(lái)源，目前支持ODPS表，后續(xù)會(huì)擴(kuò)充支持OTS維表。URI的整體格式：ODPS:/endpoint/project_name/tablename#part_key=part_value/part

16、_key2=part_value2Schema當(dāng)前支持兩種格式：Schema部分說(shuō)明訪問(wèn)的目標(biāo)數(shù)據(jù)類(lèi)型，當(dāng)前僅支持ODPS。Userinfo, host(合并稱(chēng)為authority)部分為可選項(xiàng)，在阿里云內(nèi)網(wǎng)使用時(shí)可以不用寫(xiě)，但是輸出到外部時(shí)，需要用戶(hù)明確指定連接信息。Query參數(shù)可以用來(lái)傳遞其它通用參數(shù)。定義project A中的表user為維表，將分區(qū)ds=20150101指定為維表。CREATE DIM TABLE user(Username string, Age bigint)LOCATION ODPS:/PRJA/user#ds=20150101例:臨時(shí)表語(yǔ)法格式：CREATE

17、TEMP TABLE tablename( col_name data_type comment;);說(shuō)明：在STREAMJOB中定義的Temp Table語(yǔ)句，用于輔助計(jì)算邏輯的描述，不會(huì)產(chǎn)生數(shù)據(jù)的物理存儲(chǔ)。Temp table只在本作業(yè)中有效，名稱(chēng)只在本作業(yè)可見(jiàn)Temp table描述的是streamjob中的流式數(shù)據(jù)，只接受insert into操作。結(jié)果表CREATE RESULT TABLE tablename( Col_name1 data_type comment,Col_name2 data_type commentPRIMARY KEY (col1, col2,)LOCATI

18、ON uri;語(yǔ)法格式：說(shuō)明：結(jié)果表是指最終將數(shù)據(jù)輸出的表，包括HubTable以及外部結(jié)果表兩種（臨時(shí)表并不能將結(jié)果輸出）。結(jié)果表默認(rèn)是ODPS的HubTable，僅支持insert into的方式追加數(shù)據(jù)。對(duì)結(jié)果表的操作類(lèi)型包括insert into和replace into兩種。其中，replace into僅支持定義了primary key的表，replace into一個(gè)未定義primary key的結(jié)果表則拋異常；相反，insert into僅支持未定義primary key的表，如果insert into一個(gè)定義了primary key的結(jié)果表則拋異常。DMLStream SQL支持的語(yǔ)法格式與ODPS離線SQL基本相同，但在部分SQL語(yǔ)句中行為略有不同：Join操作：join中只支持等值連接Stream SQL中支持內(nèi)連接和左連接對(duì)維表的連接，只支持一個(gè)連接鍵Select語(yǔ)句中不支持Limit操作；內(nèi)建函數(shù)Stream S