小 T 導(dǎo)讀：TDengine 3.0 引入了全新的流式計(jì)算引擎，既支持時(shí)間驅(qū)動(dòng)的流式計(jì)算，也支持事件驅(qū)動(dòng)的流式計(jì)算。本文將對(duì)新的流式計(jì)算引擎的語法規(guī)則進(jìn)行詳細(xì)介紹，方便開發(fā)者及企業(yè)使用。

TDengine 是一款開源、云原生的時(shí)序數(shù)據(jù)庫（Time Series Database，TSDB），專為物聯(lián)網(wǎng)、工業(yè)互聯(lián)網(wǎng)、金融、IT 運(yùn)維監(jiān)控等場(chǎng)景設(shè)計(jì)并優(yōu)化。近期發(fā)布的 TDengine 3.0，全新的流式計(jì)算引擎是其一大亮點(diǎn)。

TDengine 3.0 的流式計(jì)算引擎提供了實(shí)時(shí)處理寫入的數(shù)據(jù)流能力，使用 SQL 定義實(shí)時(shí)流變換，當(dāng)數(shù)據(jù)被寫入流的源表后，數(shù)據(jù)會(huì)被以定義的方式自動(dòng)處理，并根據(jù)定義的觸發(fā)模式向目的表推送結(jié)果。它提供了替代復(fù)雜流處理系統(tǒng)的輕量級(jí)解決方案，并能夠在高吞吐的數(shù)據(jù)寫入情況下，提供毫秒級(jí)的計(jì)算結(jié)果延遲。

流式計(jì)算可以包含數(shù)據(jù)過濾，標(biāo)量函數(shù)計(jì)算（含 UDF），以及窗口聚合（支持滑動(dòng)窗口、會(huì)話窗口與狀態(tài)窗口），可以以超級(jí)表、子表、普通表為源表，寫入到目的超級(jí)表。在創(chuàng)建流時(shí)，目的超級(jí)表將被自動(dòng)創(chuàng)建，隨后新插入的數(shù)據(jù)會(huì)被流定義的方式處理并寫入其中，通過 partition by 子句，可以以表名或標(biāo)簽劃分 partition，不同的 partition 將寫入到目的超級(jí)表的不同子表。

TDengine 的流式計(jì)算能夠支持分布在多個(gè) vnode 中的超級(jí)表聚合；還能夠處理亂序數(shù)據(jù)的寫入：它提供了 watermark 機(jī)制以度量容忍數(shù)據(jù)亂序的程度，并提供了 ignore expired 配置項(xiàng)以決定亂序數(shù)據(jù)的處理策略——丟棄或者重新計(jì)算。
下面我們就一起看一下 TDengine 中流式計(jì)算相關(guān)的 SQL 語法。

流式計(jì)算的創(chuàng)建、刪除與展示

創(chuàng)建

CREATE STREAM [IF NOT EXISTS] stream_name [stream_options] INTO stb_name AS subquery
stream_options: {
 TRIGGER    [AT_ONCE | WINDOW_CLOSE | MAX_DELAY time]
 WATERMARK   time
}

其中 subquery 是 select 普通查詢語法的子集：

subquery: SELECT select_list
    from_clause
    [WHERE condition]
    [PARTITION BY tag_list]
    [window_clause]

支持會(huì)話窗口、狀態(tài)窗口與滑動(dòng)窗口，其中，會(huì)話窗口與狀態(tài)窗口搭配超級(jí)表時(shí)必須與 partition by tbname 一起使用：

window_clause: {
    SESSION(ts_col, tol_val)
  | STATE_WINDOW(col)
  | INTERVAL(interval_val [, interval_offset]) [SLIDING (sliding_val)]
}

在上述語句中，SESSION 是會(huì)話窗口，tol_val 是時(shí)間間隔的最大范圍。在 tol_val 時(shí)間間隔范圍內(nèi)的數(shù)據(jù)都屬于同一個(gè)窗口，如果有連續(xù)兩條數(shù)據(jù)的時(shí)間超過 tol_val，則自動(dòng)開啟下一個(gè)窗口。窗口的定義與時(shí)序數(shù)據(jù)特色查詢中的定義完全相同，詳見 TDengine 特色查詢。

例如，使用如下語句創(chuàng)建流式計(jì)算，同時(shí)自動(dòng)創(chuàng)建名為 avg_vol 的超級(jí)表，此流計(jì)算以一分鐘為時(shí)間窗口、30 秒為前向增量統(tǒng)計(jì)這些電表的平均電壓，并將來自 meters 表的數(shù)據(jù)的計(jì)算結(jié)果寫入 avg_vol 表，不同 partition 的數(shù)據(jù)會(huì)分別創(chuàng)建子表并寫入不同子表。

CREATE STREAM avg_vol_s INTO avg_vol AS
SELECT _wstartts, count(*), avg(voltage) FROM meters PARTITION BY tbname INTERVAL(1m) SLIDING(30s);

刪除

DROP STREAM [IF NOT EXISTS] stream_name;

僅刪除流式計(jì)算任務(wù)，由流式計(jì)算寫入的數(shù)據(jù)不會(huì)被刪除。

展示

SHOW STREAMS;

若要展示更詳細(xì)的信息，可以使用：

SELECT * from performance_schema.`perf_streams`;

流式計(jì)算的 partition

我們可以使用 PARTITION BY TBNAME 或 PARTITION BY tag 對(duì)一個(gè)流進(jìn)行多分區(qū)的計(jì)算，每個(gè)分區(qū)的時(shí)間線與時(shí)間窗口是獨(dú)立的，會(huì)各自聚合，并寫入到目的表中的不同子表。如果不帶 PARTITION BY 選項(xiàng)，那所有的數(shù)據(jù)將寫入到一張子表。

流式計(jì)算創(chuàng)建的超級(jí)表有唯一的 tag 列 groupId，每個(gè) partition 會(huì)被分配唯一 groupId。與 schemaless 寫入一致，我們通過 MD5 計(jì)算子表名，并自動(dòng)創(chuàng)建它。

流式計(jì)算的觸發(fā)模式

在創(chuàng)建流時(shí)，可以通過 TRIGGER 指令指定流式計(jì)算的觸發(fā)模式。

對(duì)于非窗口計(jì)算，流式計(jì)算的觸發(fā)是實(shí)時(shí)的；對(duì)于窗口計(jì)算，目前提供如下 3 種觸發(fā)模式：

1. AT_ONCE：寫入立即觸發(fā)
2. WINDOW_CLOSE：窗口關(guān)閉時(shí)觸發(fā)（窗口關(guān)閉由事件時(shí)間決定，可配合 watermark 使用）
3. MAX_DELAY time：若窗口關(guān)閉，則觸發(fā)計(jì)算。若窗口未關(guān)閉，且未關(guān)閉時(shí)長超過 max delay 指定的時(shí)間，則觸發(fā)計(jì)算。

由于窗口關(guān)閉是由事件時(shí)間所決定的，如果因事件流中斷、或持續(xù)延遲導(dǎo)致事件時(shí)間無法更新，可能無法得到最新的計(jì)算結(jié)果。因此，流式計(jì)算提供了以事件時(shí)間結(jié)合處理時(shí)間計(jì)算的 MAX_DELAY 觸發(fā)模式，MAX_DELAY 模式在窗口關(guān)閉時(shí)會(huì)立即觸發(fā)計(jì)算。此外，當(dāng)數(shù)據(jù)寫入后，計(jì)算觸發(fā)的時(shí)間超過 max delay 指定的時(shí)間，則立即觸發(fā)計(jì)算。

流式計(jì)算的窗口關(guān)閉

流式計(jì)算以事件時(shí)間（插入記錄中的時(shí)間戳主鍵）為基準(zhǔn)計(jì)算窗口關(guān)閉，而非以 TDengine 服務(wù)器的時(shí)間，這樣可以避免客戶端與服務(wù)器時(shí)間不一致帶來的問題，有效解決亂序數(shù)據(jù)寫入等難題。同時(shí)，流式計(jì)算還提供了 watermark 來定義容忍的亂序程度。
在創(chuàng)建流時(shí)，我們可以在 stream_option 中指定 watermark，它定義了數(shù)據(jù)亂序的容忍上界。流式計(jì)算通過 watermark 來度量對(duì)亂序數(shù)據(jù)的容忍程度，watermark 默認(rèn)為 0。

T = 最新事件時(shí)間 – watermark

每次寫入的數(shù)據(jù)都會(huì)以上述公式更新窗口關(guān)閉時(shí)間，并將窗口結(jié)束時(shí)間 < T 的所有打開的窗口關(guān)閉，若觸發(fā)模式為 WINDOW_CLOSE 或 MAX_DELAY，則推送窗口聚合結(jié)果。

在上圖中，縱軸表示不同時(shí)刻，對(duì)于不同時(shí)刻，我們畫出其對(duì)應(yīng)的 TDengine 收到的數(shù)據(jù)，即為橫軸。已知橫軸上的數(shù)據(jù)點(diǎn)表示已經(jīng)收到的數(shù)據(jù)，其中藍(lán)色的點(diǎn)表示事件時(shí)間（即數(shù)據(jù)中的時(shí)間戳主鍵）最后的數(shù)據(jù)，該數(shù)據(jù)點(diǎn)減去定義的 watermark 時(shí)間，就得到亂序容忍的上界 T。所有結(jié)束時(shí)間小于 T 的窗口都將被關(guān)閉（圖中以灰色方框標(biāo)記）。

在 T2 時(shí)刻，亂序數(shù)據(jù)（黃色的點(diǎn)）到達(dá) TDengine，由于有 watermark 的存在，這些數(shù)據(jù)進(jìn)入的窗口并未被關(guān)閉，因此可以被正確處理。在 T3 時(shí)刻，最新事件到達(dá)，T 向后推移超過了第二個(gè)窗口關(guān)閉的時(shí)間，該窗口被關(guān)閉，亂序數(shù)據(jù)被正確處理。

但要注意，在 window_close 或 max_delay 模式下，窗口關(guān)閉直接影響推送結(jié)果。在 at_once 模式下，窗口關(guān)閉只與內(nèi)存占用有關(guān)。

流式計(jì)算的過期數(shù)據(jù)處理策略

對(duì)于已關(guān)閉的窗口，再次落入該窗口中的數(shù)據(jù)就會(huì)被標(biāo)記為過期數(shù)據(jù)。TDengine 對(duì)于過期數(shù)據(jù)提供兩種處理方式，由 IGNORE EXPIRED 選項(xiàng)指定：

1. 重新計(jì)算，即 IGNORE EXPIRED 0：默認(rèn)配置，從 TSDB 中重新查找對(duì)應(yīng)窗口的所有數(shù)據(jù)并重新計(jì)算得到最新結(jié)果
2. 直接丟棄，即 IGNORE EXPIRED 1：忽略過期數(shù)據(jù)

無論在哪種模式下，watermark 都應(yīng)該被妥善設(shè)置，來得到正確結(jié)果（直接丟棄模式）或避免頻繁觸發(fā)重算帶來的性能開銷（重新計(jì)算模式）。

示例

企業(yè)電表的數(shù)據(jù)經(jīng)常都是成百上千億條的，想要將這些分散、凌亂的數(shù)據(jù)清洗或轉(zhuǎn)換都需要比較長的時(shí)間，很難做到高效性和實(shí)時(shí)性。在如下例子中，通過 TDengine 流計(jì)算可以將電表電壓大于 220V 的數(shù)據(jù)清洗掉，然后以 5 秒為窗口整合并計(jì)算出每個(gè)窗口中電流的最大值，最后將結(jié)果輸出到指定的數(shù)據(jù)表中。

創(chuàng)建 Database 和原始數(shù)據(jù)表

首先準(zhǔn)備數(shù)據(jù)，完成建庫、建一張超級(jí)表和多張子表操作：

DROP DATABASE IF EXISTS power;
CREATE DATABASE power;
USE power;

CREATE STABLE meters (ts timestamp, current float, voltage int, phase float) TAGS (location binary(64), groupId int);

CREATE TABLE d1001 USING meters TAGS ("California.SanFrancisco", 2);
CREATE TABLE d1002 USING meters TAGS ("California.SanFrancisco", 3);
CREATE TABLE d1003 USING meters TAGS ("California.LosAngeles", 2);
CREATE TABLE d1004 USING meters TAGS ("California.LosAngeles", 3);

創(chuàng)建流

create stream current_stream into current_stream_output_stb as select _wstart as start, _wend as end, max(current) as max_current from meters where voltage <= 220 interval (5s);

寫入數(shù)據(jù)

insert into d1001 values("2018-10-03 14:38:05.000", 10.30000, 219, 0.31000);
insert into d1001 values("2018-10-03 14:38:15.000", 12.60000, 218, 0.33000);
insert into d1001 values("2018-10-03 14:38:16.800", 12.30000, 221, 0.31000);
insert into d1002 values("2018-10-03 14:38:16.650", 10.30000, 218, 0.25000);
insert into d1003 values("2018-10-03 14:38:05.500", 11.80000, 221, 0.28000);
insert into d1003 values("2018-10-03 14:38:16.600", 13.40000, 223, 0.29000);
insert into d1004 values("2018-10-03 14:38:05.000", 10.80000, 223, 0.29000);
insert into d1004 values("2018-10-03 14:38:06.500", 11.50000, 221, 0.35000);

查詢以觀察結(jié)果

taos> select start, end, max_current from current_stream_output_stb;
          start          |           end           |     max_current      |
===========================================================================
 2018-10-03 14:38:05.000 | 2018-10-03 14:38:10.000 |             10.30000 |
 2018-10-03 14:38:15.000 | 2018-10-03 14:38:20.000 |             12.60000 |
Query OK, 2 rows in database (0.018762s)

寫在最后

如果大家能夠運(yùn)用好 TDengine 3.0 提供的流計(jì)算引擎，就不需要再部署其他的第三方流處理系統(tǒng)，這樣一來，不僅降低了系統(tǒng)的復(fù)雜度，還大大減少了研發(fā)和運(yùn)維成本。在實(shí)際操作中應(yīng)用 TDengine 流計(jì)算引擎時(shí)，上述的詳細(xì)語法會(huì)帶給你很多幫助，如果還產(chǎn)生了其他更為復(fù)雜的應(yīng)用問題，你也可以進(jìn)入 TDengine 社區(qū)向技術(shù)人員尋求幫助。