createTimeSeriesEngine

createTimeSeriesEngine(name, windowSize, step, metrics, dummyTable, outputTable, [timeColumn], [useSystemTime=false], [keyColumn], [garbageSize], [updateTime], [useWindowStartTime], [roundTime=true], [snapshotDir], [snapshotIntervalInMsgCount], [fill='none'], [forceTriggerTime], [raftGroup], [keyPurgeFreqInSec=-1], [closed='left'], [outputElapsedMicroseconds=false], [subWindow], [parallelism=1], [acceptedDelay=0], [outputHandler=NULL], [msgAsTable=false])

别名：createTimeSeriesAggregator

创建流数据时间序列引擎，以实现基于时间的滑动窗口或滚动窗口进行实时计算。

时序引擎的聚合计算的算子分为增量计算和全量计算两种。增量计算算子不会保留历史数据，每次有数据进来就会进行增量计算；而全量计算算子（例如自定义的聚合函数，或未经优化的内置聚合函数，亦或是嵌套了状态函数的函数）会保留窗口内完整的数据，待触发输出时进行全量计算。

时序引擎对以下聚合计算算子进行了优化，实现了增量计算，显著提升了性能：corr, covar, first, last, max, med, min, percentile, quantile, std, var, sum, sum2, sum3, sum4, wavg, wsum, count, firstNot, ifirstNot, lastNot, ilastNot, imax, imin, nunique, prod, sem, mode, searchK, beta, avg。

更多流数据引擎的应用场景说明可以参考 内置流式计算引擎。

返回值：一个表对象，通过向该表对象写入，将数据注入时间序列引擎进行计算。

name 字符串标量，表示时间序列引擎的名称，作为其在一个数据节点/计算节点上的唯一标识。可包含字母，数字和下划线，但必须以字母开头。

windowSize 正整数标量或向量，表示滑动窗口的长度。

dummyTable 一个表对象，和订阅的流数据表的 schema 一致，可以含有数据，亦可为空表。

keyColumn 可选参数，字符串标量或向量，表示分组列名。若设置，则分组进行聚合计算，例如以每支股票为一组进行聚合计算。

updateTime 可选参数，非负整数，单位与 timeColumn 的时间精度一致。用于指定比 step 更短的计算时间间隔。step 必须是 updateTime 的整数倍。要设置 updateTime， useSystemTime 必须设为 false。

如果没有指定 updateTime，一个数据窗口结束前，不会发生对该数据窗口数据的计算。若一个窗口长时间未触发计算，可以指定 updateTime，分多次触发当前窗口数据的计算。计算触发的规则为：

• updateTime 指定为正整数值时：

  • 从当前窗口的左边界开始，每隔 updateTime 时间，若有新的数据到来，则对当前窗口内该数据之前的所有数据进行计算。

  • 如果系统经过 2 * updateTime （至少2秒）后仍有未被处理的数据，则触发对当前窗口内所有数据的计算。

  • 若分组计算，则每组内进行上述操作。

• updateTime 指定为 0 时：在新的数据到来后，立即对当前窗口的最新数据计算并输出。

指定 updateTime 后，建议使用键值内存表作为输出表。因为每次计算均会增加一条记录，输出表若使用普通内存表或流数据表，则会产生大量带有相同时间戳的结果。因为键值流数据表不可更新记录，输出表亦不推荐使用键值流数据表。

useWindowStartTime 可选参数，布尔值，表示输出表中的时间是否为数据窗口起始时间。默认值为 false，表示输出表中的时间为数据窗口起始时间 + windowSize 。若 windowSize 是向量， useWindowStartTime 必须为 false。

roundTime 可选参数，布尔值，表示若数据时间精度为毫秒或者秒且 step > 一分钟，如何对窗口边界值进行规整处理。默认值为 true，表示按照既定的多分钟规则进行规整。若为 false，则按一分钟规则进行窗口规整。

若要开启快照机制（snapshot），必须指定 snapshotDir 与 snapshotIntervalInMsgCount。

snapshotIntervalInMsgCount 可选参数，为整数类型，表示每隔多少条数据保存一次流数据引擎快照。

raftGroup 可选参数，表示流数据高可用订阅端 raft 组的 ID (大于1的整数，由流数据高可用相关的配置项 streamingRaftGroups 指定)。设置该参数表示开启计算引擎高可用。在 leader 节点创建流数据引擎后，会同步在 follower 节点创建该引擎。每次保存的 snapshot 也会同步到 follower。当 raft 组的 leader 节点宕机时，会自动切换新 leader 节点重新订阅流数据表。请注意，若要指定 raftGroup，必须同时指定 snapshotDir。

outputElapsedMicroseconds 布尔值，表示是否输出每个窗口从触发计算到计算完成输出结果的耗时（若指定了 keyColumn 则包含数据分组的耗时），默认为 false。指定参数 outputElapsedMicroseconds 后，在定义 outputTable 时需要在时间列和分组列后增加一个 LONG 类型的列，详见 outputTable 参数说明。

subWindow 整型或者 DURATION 数据对。在滑动窗口内指定子窗口，仅计算子窗口内的数据。子窗口边界的开闭情况由参数 closed 决定。子窗口结束后收到第一条数据触发对子窗口内数据的计算（参考例4）。当 subWindow 为整型数据对时，其单位与 timeColumn 的时间精度一致。若指定 subWindow，则：

• windowSize 和 step 必须相等。

• 不可设置 updateTime>0和 useSystemTime=true。

parallelism 为不超过 63 的正整数，可选参数，表示并行计算的工作线程数，默认值为 1。在计算量较大时，合理地调整该参数能够有效利用计算资源，降低计算耗时。建议小于机器核数，推荐值为 4 到 8 。

outputHandler 一元函数。设置此参数时，引擎计算结束后，不再将计算结果写到输出表，而是会调用此函数处理计算结果。默认值为 NULL，表示仍将结果写到输出表。

msgAsTable 布尔标量，表示在设置了参数 outputHandler 时，将引擎的计算结果以表的结构调用函数。默认值为 false，此时将计算结果的每一列作为元素组成元组。

例1. 时间序列引擎 engine1 订阅流数据表 trades，实时计算表 trades 中过去1分钟内每只股票交易量之和。

share streamTable(1000:0, ["time","sym","volume"], [TIMESTAMP, SYMBOL, INT]) as trades
share table(10000:0, ["time","sym","sumVolume"], [TIMESTAMP, SYMBOL, INT]) as output1
engine1 = createTimeSeriesEngine(name="engine1", windowSize=60000, step=60000, metrics=<[sum(volume)]>, dummyTable=trades, outputTable=output1, timeColumn="time", useSystemTime=false, keyColumn="sym", garbageSize=50, useWindowStartTime=false)
subscribeTable(tableName="trades", actionName="engine1", offset=0, handler=append!{engine1}, msgAsTable=true);

insert into trades values(2018.10.08T01:01:01.785,`A,10)
insert into trades values(2018.10.08T01:01:02.125,`B,26)
insert into trades values(2018.10.08T01:01:10.263,`B,14)
insert into trades values(2018.10.08T01:01:12.457,`A,28)
insert into trades values(2018.10.08T01:02:10.789,`A,15)
insert into trades values(2018.10.08T01:02:12.005,`B,9)
insert into trades values(2018.10.08T01:02:30.021,`A,10)
insert into trades values(2018.10.08T01:04:02.236,`A,29)
insert into trades values(2018.10.08T01:04:04.412,`B,32)
insert into trades values(2018.10.08T01:04:05.152,`B,23)

sleep(10)

select * from output1;

下面详细解释时间序列引擎的计算过程。为简便起见，以下提到时间时，省略相同的日期部分，只列出（小时:分钟:秒.毫秒）部分。

首先，时间序列引擎对第一个数据窗口的起始时间进行规整。第一个数据窗口的时间范围是 01:01:00.000 到 01:02:00.000，只包含左边界，不包含右边界。当(01:02:10.789,A,15)到达时，触发第一个窗口 A 组计算；当(01:02:12.005,B,9)到达时，触发第一个窗口 B 组计算。

第二个数据窗口的时间范围是 01:02:00.000 到 01:03:00.000。当(01:04:02.236,A,29)到达时，触发第二个窗口 A 组计算；当(01:04:04.412,B,32)到达时，触发第二个窗口 B 组计算。

由于 01:05:00.000 及之后没有数据，因此没有对 01:04:00.000 到 01:05:00.000 之间的数据进行计算。

输出表 output1 保存了时间序列引擎的计算结果。由于 useWindowStartTime 为 false，因此输出表 output1 中的时间为窗口的结束时间。若将 useWindowStartTime 设为 true，则输出表中的时间为窗口的起始时间。例如：

output2 = table(10000:0, ["time","sym","sumVolume"], [TIMESTAMP, SYMBOL, INT])
engine2 = createTimeSeriesEngine(name="engine2", windowSize=60000, step=60000, metrics=<[sum(volume)]>, dummyTable=trades, outputTable=output2, timeColumn="time", useSystemTime=false, keyColumn="sym", garbageSize=50, useWindowStartTime=true)
subscribeTable(tableName="trades", actionName="engine2", offset=0, handler=append!{engine2}, msgAsTable=true)

sleep(10)
select * from output2;

下例中指定 updateTime 为 1000（毫秒）：

share keyedTable(["time","sym"],10000:0, ["time","sym","sumVolume"], [TIMESTAMP, SYMBOL, INT]) as output3
engine3 = createTimeSeriesEngine(name="engine3", windowSize=60000, step=60000, metrics=<[sum(volume)]>, dummyTable=trades, outputTable=output3, timeColumn="time", useSystemTime=false, keyColumn="sym", garbageSize=50, updateTime=1000, useWindowStartTime=false)
subscribeTable(tableName="trades", actionName="engine3", offset=0, handler=append!{engine3}, msgAsTable=true)

sleep(2001)
select * from output3;

例2. 下例中指定 updateTime 为 0（毫秒）：

share streamTable(1000:0, ["time","sym","volume"], [TIMESTAMP, SYMBOL, INT]) as trades
share table(1000:0, ["time","sym","sumVolume"], [TIMESTAMP, SYMBOL, INT]) as output4
engine4 = createTimeSeriesEngine(name="engine4", windowSize=60000, step=60000, metrics=<[sum(volume)]>, dummyTable=trades, outputTable=output4, timeColumn="time", useSystemTime=false, keyColumn="sym", garbageSize=50, updateTime=0, useWindowStartTime=false)
subscribeTable(tableName="trades", actionName="engine4", offset=0, handler=append!{engine4}, msgAsTable=true);

insert into trades values(2018.10.08T01:01:01.785,`A,10)
insert into trades values(2018.10.08T01:01:02.125,`B,26)
insert into trades values(2018.10.08T01:01:10.263,`B,14)
insert into trades values(2018.10.08T01:01:12.457,`A,28)

sleep(100)

insert into trades values(2018.10.08T01:01:12.557,`B,21)
insert into trades values(2018.10.08T01:01:12.557,`A,28)


select * from output4;

updateTime=0 时，每收到一批数据就触发一次计算输出：

• 01:01:12.457 时，触发了第一批数据中 A、B 两组数据的计算输出。

• 01:01:12.557 时，再次触发第二批数据中 A、B 两组数据的计算输出。

最终得到：

下面以最后一个数据窗口为例，介绍时间序列引擎指定 updateTime 时如何进行计算。假设 time 列时间亦为数据进入时间序列引擎的时刻。

1. 在 01:04:04.236 时，A 分组的第一条记录到达后已经过 2000 毫秒，触发一次 A 组计算，输出表增加一条记录(01:05:00.000, `A, 29)。

2. 在 01:04:05.152 时的 B 组记录为 01:04:04.412 所在小窗口[01:04:04.000, 01:04:05.000)之后第一条记录，触发一次 B 组计算，输出表增加一条记录(01:05:00,"B",32)。

3. 2 秒（2*updateTime个时间单位）后，在 01:04:07.152 时，由于 01:04:05.152 时的 B 组记录仍未参与计算，触发一次B组计算，输出一条记录(01:05:00,"B",55)。由于输出表的主键为 time 和 sym，并且输出表中已有(01:05:00,"B",32)这条记录，因此将该记录更新为(01:05:00,"B",55)。

下例中，共享流数据表 "pubT" 包含两个时间列，类型分别时 DATE 和 SECOND，创建时间序列引擎时，通过设置 timeColumn 来将原来流数据表的两个时间列整合为输出表 streamMinuteBar_1min 中的一个类型为 DATETIME 的时间列。

colNames=["symbol","date","minute","price","type","volume"]
colTypes=[SYMBOL, DATE, SECOND, DOUBLE, STRING, INT]
pubTable = streamTable(10000:0,colNames,colTypes)
share pubTable as pubT

colNames = ["time","symbol","open","max","min","close","volume","amount","ret","vwap"]
colTypes = [DATETIME, SYMBOL, DOUBLE, DOUBLE, DOUBLE, DOUBLE, INT, DOUBLE, DOUBLE, DOUBLE]
share streamTable(10000:0,colNames, colTypes) as streamMinuteBar_1min

tsAggrOHLC = createTimeSeriesEngine(name="subT", windowSize=60, step=60, metrics=<[first(price) as open ,max(price) as max,min(price) as min ,last(price) as close ,sum(volume) as volume ,wsum(volume, price) as amount ,(last(price)-first(price))/first(price)) as ret, (wsum(volume, price)/sum(volume)) as vwap]>, dummyTable=pubTable, outputTable=streamMinuteBar_1min, timeColumn=["date","minute"], useSystemTime=false, keyColumn="symbol", fill="none")
subscribeTable(tableName="pubT", actionName="subT", offset=-1, handler=append!{tsAggrOHLC}, msgAsTable=true)

insert into pubT values("000001", 2021.04.05, 09:25:01, 1, 'B', 1)
insert into pubT values("000001", 2021.04.05, 09:30:05, 2, 'B', 1)
insert into pubT values("000001", 2021.04.05, 09:31:06, 3, 'B', 1)
insert into pubT values("000001", 2021.04.05, 09:35:05, 4, 'S', 4)
insert into pubT values("000001", 2021.04.05, 09:40:05, 5, 'S', 5)
insert into pubT values("000001", 2021.04.06, 09:25:05, 6, 'S', 6)

pubT

select * from streamMinuteBar_1min

share streamTable(1000:0, ["time","sym","qty"], [DATETIME, SYMBOL, INT]) as trades
share table(10000:0, ["time","sym","sumQty"], [DATETIME, SYMBOL, INT]) as output5

engine = createTimeSeriesEngine(name="engine", windowSize=6, step=6, metrics=<sum(qty)>, dummyTable=trades, outputTable=output5, timeColumn="time",keyColumn="sym", forceTriggerTime=7,fill=1000)
subscribeTable(tableName="trades", actionName="engine", offset=0, handler=append!{engine}, msgAsTable=true)
sleep(1000)
insert into engine values(2018.08.01T14:05:43,`A,1)
insert into engine values(2018.08.01T14:05:43,`C,3)
sleep(10)
insert into engine values(2018.08.01T14:05:44,`B,1)
sleep(80)
insert into engine values(2018.08.01T14:05:52,`B,3)
sleep(20)
insert into engine values(2018.08.01T14:05:54,`A,3)
sleep(10)
insert into engine values(2018.08.01T14:05:55,`A,5)
sleep(20)
insert into engine values(2018.08.01T14:05:57,`B,5)
sleep(50)
insert into engine values(2018.08.01T14:06:12,`A,1)
sleep(50)
select * from output5 order by sym

例3. 下例计算每个分组窗口的第一个和最后一个 volume，组合后输出为一个数组向量。

//通过 defg 自定义一个函数，引擎在调用该函数时会将返回值转换为数组向量
defg toVector(x){
	return x	
}
share streamTable(1000:0, ["time","sym","volume","price"], [TIMESTAMP, SYMBOL, DOUBLE,DOUBLE]) as trades
//定义输出表，对应 toVector 指标输出的列需要定义为数组向量类型，此例中将其定义为 DOUBLE[]
share table(10000:0, ["time","sym","sumVolume","avg"], [TIMESTAMP,STRING,DOUBLE[],DOUBLE]) as output6
//metrics 中调用了 toVector 函数，将1分钟窗口内的第一个 volume 和最后一个 volume 组合到一个向量中。
engine1 = createTimeSeriesEngine(name="engine1", windowSize=60000, step=60000, metrics=<[toVector([first(volume),last(volume)]),avg(volume+price)]>, dummyTable=trades, outputTable=output6, timeColumn="time", keyColumn="sym" , useSystemTime=false, garbageSize=50, useWindowStartTime=false)

times = sort(2023.10.08T00:00:00.000 + rand(1..(1+3000*200), 30))
syms = rand("A"+string(1..10), 30)
volumes = rand(rand(100.0, 10) join 2.3 NULL NULL NULL NULL, 30)
prices = rand(rand(100.0, 10) join 2.3 NULL NULL NULL NULL, 30)
t=table(times as time, syms as sym, volumes as volume,prices as price)

engine1.append!(t)

select * from output6 where time between 2023.10.08T00:01:00.000 and 2023.10.08T00:05:00.000 order by time,sym 

例4. 本例将说明 subWindow 参数的作用。

share streamTable(1000:0, ["time","sym","volume"], [TIMESTAMP, SYMBOL, INT]) as trades
share table(10000:0, ["time","sym","sumVolume"], [TIMESTAMP, SYMBOL, INT]) as output7
//在长度为 1 分钟的窗口内指定一个子窗口，未指定 closed 时，子窗口左闭右开。这里子窗口为每分钟的[0s, 10s)
engine4 = createTimeSeriesEngine(name="engine4", windowSize=60000, step=60000, metrics=<[sum(volume)]>, dummyTable=trades, outputTable=output7, timeColumn="time", useSystemTime=false, keyColumn="sym", garbageSize=50, useWindowStartTime=true, subWindow=0s:10s)

subscribeTable(tableName="trades", actionName="engine4", offset=0, handler=append!{engine4}, msgAsTable=true);

insert into trades values(2018.10.08T01:01:01.785,`A,10)
insert into trades values(2018.10.08T01:01:02.125,`B,26)
insert into trades values(2018.10.08T01:01:10.000,`A,14)
insert into trades values(2018.10.08T01:01:12.457,`A,28)

sleep(10)

select * from output7;

2018.10.08T01:01:10.000 时刻的数据触发 [2018.10.08T01:01:00.000, 2018.10.08T01:01:10.000) 区间内 A 组数据对应的 volume 的总和。在子窗口结束后没有再收到 B 组的数据，因此 B 组子窗口内数据未被触发计算。

例5. 在使用数据时间截取数据窗口时（useSystemTime=false），若收到乱序数据，可以设置参数 acceptedDelay。在触发窗口结束的时刻（窗口右边界+ acceptedDelay）之前收到属于该窗口的数据都将参与计算。

share streamTable(1000:0, ["time","sym","volume"], [TIMESTAMP, SYMBOL, INT]) as trades
share table(10000:0, ["time","sym","sumVolume"], [TIMESTAMP, SYMBOL, INT]) as output8

engine5 = createTimeSeriesEngine(name="engine5", windowSize=60000, step=60000, metrics=<[sum(volume)]>, dummyTable=trades, outputTable=output8, timeColumn="time", useSystemTime=false, keyColumn="sym", garbageSize=50, useWindowStartTime=true, acceptedDelay=500)

subscribeTable(tableName="trades", actionName="engine5", offset=0, handler=append!{engine5}, msgAsTable=true);

insert into trades values(2024.10.08T09:30:59.000,`A,10)
insert into trades values(2024.10.08T09:31:00.000,`B,26)
insert into trades values(2024.10.08T09:31:00.300,`A,14)
insert into trades values(2024.10.08T09:31:00.000,`A,28)
insert into trades values(2024.10.08T09:32:00.500,`A,16)
sleep(10)

select * from output8;

若未设置 acceptedDelay，窗口中的乱序数据（例如 2024.10.08T09:31:00.000 时刻的数据）将被丢弃，不参与计算。在本例中，设置了 acceptedDelay=500，因此 2024.10.08T09:32:00.500 时刻的数据将触发在它之前收到的所有属于 [2024.10.08T09:31:00.000, 2024.10.08T09:32:000) 区间的 A 组数据对应的 volume 的总和。