createThresholdEngine

createThresholdEngine(name, threshold, metrics, dummyTable, outputTable, thresholdColumn, [keyColumn], [timeColumn], [sessionBegin], [sessionEnd], [keyPurgeDaily=false], [forceTriggerSessionEndTime=0], [snapshotDir], [snapshotIntervalInMsgCount], [outputElapsedMicroseconds=false], [outputThreshold=false])

创建阈值引擎，以实现基于累计值阈值触发的聚合计算，每当阈值列（thresholdColumn）的累计值达到特定阈值（n * threshold）时，进行一次聚合计算。此处的 n 表示从 1 开始递增的正整数，用于表示当前累计值达到的是第几个阈值边界，即触发边界依次为 threshold、2 * threshold、3 * threshold，依此类推。

• 若指定 keyColumn 进行分组，则聚合计算将在各分组内独立进行。
• 若指定 sessionBegin 和 sessionEnd，则仅有 [sessionBegin, sessionEnd] 范围内的数据会参与计算。

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

threshold 正整数标量，指定阈值步长。

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

outputTable 计算结果的输出表，可以是内存表或分布式表。在使用createThresholdEngine函数之前，需要将输出表预先设立为一个空表，并指定各列列名以及数据类型。阈值引擎会将计算结果插入该表。

输出表的列顺序如下：

keyColumn 可选参数，字符串标量或向量，表示分组列名。若设置，则分组进行聚合计算。

timeColumn 可选参数，字符串标量或向量，用于指定订阅的流数据表中时间列的名称。

• 注：字符串向量必须是 date 和 time 组成的向量，date 类型为 DATE，time 类型为 TIME, SECOND 或 NANOTIME。此时，输出表第一列的时间类型必须与 concatDateTime(date, time) 的类型一致。

sessionBegin 可选参数，可以是与时间列的数据类型对应的 SECOND、TIME 或 NANOTIME 类型的标量或向量，表示每个时间段的起始时刻。如果 sessionBegin 是一个向量，它必须是递增的。

sessionEnd 可选参数，可以是与时间列的数据类型对应的 SECOND、TIME 或 NANOTIME 类型的标量或向量，表示每个时间段的结束时刻。可在 sessionEnd 中指定 00:00:00 表示的次日的零点（即当日的 24:00:00）。

keyPurgeDaily 可选参数，布尔标量。指定是否在引擎收到第一批包含新日期的数据时，先清空之前保存的所有分组，再对这批新数据进行处理。默认为 false，表示引擎不会清理前一天的分组。

snapshotDir 可选参数，字符串，表示保存引擎快照的文件目录。

• 指定的目录必须存在，否则系统会提示异常。
• 创建流数据引擎时，如果指定了 snapshotDir，会检查该目录下是否存在快照。如果存在，会加载该快照，恢复引擎的状态。
• 多个引擎可以指定同一个目录存储快照，用引擎的名称来区分快照文件。
• 一个引擎的快照可能会使用三个文件名：
  • 临时存储快照信息：文件名为 <engineName>.tmp；
  • 快照生成并刷到磁盘：文件保存为 <engineName>.snapshot；
  • 存在同名快照：旧快照自动重命名为 <engineName>.old。

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

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

outputThreshold 可选参数，布尔标量，表示是否输出阈值窗口值和 thresholdColumn 累计和。

一个表对象，通过向该表对象写入，将数据注入阈值引擎进行计算。

例1 通过时序聚合引擎计算1分钟 K 线，然后通过 createThresholdEngine 将 1 分钟 K 线聚合为 5 分钟。在窗口左闭右开的情况下，可以提前一分钟结束窗口并计算输出，从而比使用 createTimeSeriesEngine 减少延时。

// 准备数据
n = 1000000;
sampleDate = 2019.11.07;
symbols = `600519`000001`600000`601766;
trade = table(take(sampleDate, n) as date, 
	(09:30:00.000 + rand(7200000, n/2)).sort!() join (13:00:00.000 + rand(7200000, n/2)).sort!() as time, 
	rand(symbols, n) as symbol, 
	100+cumsum(rand(0.02, n)-0.01) as price, 
	rand(1000, n) as volume)

// 创建dummyTable, outputTable	
share(streamTable(10:0,`date`time`symbol`price`volumn,[DATE, TIME, SYMBOL, DOUBLE, DOUBLE]), `trades);
share(table(1:0, `timestamp`symbol`open`high`low`close, [TIMESTAMP,SYMBOL,DOUBLE,DOUBLE,DOUBLE,DOUBLE]), `outputTable);
go
// 创建阈值引擎
thresholdEngine = createThresholdEngine(name="demo", threshold=1000000, metrics=<[first(price), max(price), min(price), last(price)]>, dummyTable=trades, outputTable=outputTable, thresholdColumn=`volumn, keyColumn=`symbol, timeColumn=[`date, `time]);

// 插入数据
thresholdEngine.append!(trade);

// 查询计算结果
select * from outputTable

部分结果如下：

本例中 outputTable 的定义说明：

创建引擎时指定了 timeColumn、keyColumn 和 metrics 且未指定 outputThreshold=true 和 outputElapsedMicroseconds=true，因此 outputTable 的列依次对应如下：

例2 通过 sessionBegin 和 sessionEnd 限定只处理交易时段内的数据。

// 准备数据
t = table(
    2024.01.02 2024.01.02 2024.01.02 2024.01.02 2024.01.02 as date,
    09:20:00.000 09:31:00.000 10:15:00.000 11:35:00.000 13:05:00.000 as time,
    `A`A`A`A`A as sym,
    10.1 10.2 10.3 10.4 10.5 as price,
    200 500 600 700 800 as volume
)

// 创建 dummyTable, outputTable
share(streamTable(10:0, `date`time`sym`price`volume, [DATE, TIME, SYMBOL, DOUBLE, DOUBLE]), `trades1)
share(table(10:0, `timestamp`sym`sumVol, [TIMESTAMP, SYMBOL, DOUBLE]), `outputTable1)
go
// 创建阈值引擎
engine = createThresholdEngine(
    name="thresholdDemo1",
    threshold=1000,
    metrics=<[sum(volume)]>,
    dummyTable=trades1,
    outputTable=outputTable1,
    thresholdColumn=`volume,
    keyColumn=`sym,
    timeColumn=[`date, `time],
    sessionBegin=09:30:00.000 13:00:00.000,
    sessionEnd=11:30:00.000 15:00:00.000
)

// 插入数据
engine.append!(t)

// 查询结果
select * from outputTable1

09:20:00.000 和 11:35:00.000 的数据不在 [sessionBegin, sessionEnd] 指定的交易时段内，不参与阈值累计。

例3 通过 keyPurgeDaily 控制跨日后是否清空前一日分组状态

// 创建 dummyTable, outputTable
share(streamTable(10:0, `date`time`sym`price`volume, [DATE, TIME, SYMBOL, DOUBLE, DOUBLE]), `trades)
share(table(10:0, `timestamp`sym`sumVol, [TIMESTAMP, SYMBOL, DOUBLE]), `output1)
share(table(10:0, `timestamp`sym`sumVol, [TIMESTAMP, SYMBOL, DOUBLE]), `output2)
go
// keyPurgeDaily=true：跨日后清空前一日分组状态
engine1 = createThresholdEngine(
    name="thresholdDemo2_true",
    threshold=1000,
    metrics=<[sum(volume)]>,
    dummyTable=trades,
    outputTable=output1,
    thresholdColumn=`volume,
    keyColumn=`sym,
    timeColumn=[`date, `time],
    keyPurgeDaily=true
)

// keyPurgeDaily=false：跨日后保留前一日分组状态
engine2 = createThresholdEngine(
    name="thresholdDemo2_false",
    threshold=1000,
    metrics=<[sum(volume)]>,
    dummyTable=trades,
    outputTable=output2,
    thresholdColumn=`volume,
    keyColumn=`sym,
    timeColumn=[`date, `time],
    keyPurgeDaily=false
)

// 准备数据
t = table(
    2024.01.02 2024.01.03 2024.01.03 as date,
    14:59:00.000 09:31:00.000 09:31:01.000 as time,
    `A`A`A as sym,
    10.1 10.2 10.3 as price,
    600 500 700 as volume
)

// 插入数据
engine1.append!(t)
engine2.append!(t)

// 查询结果
select * from output1

select * from output2

例4 通过 outputElapsedMicroseconds 输出每次触发计算的耗时

// 创建 dummyTable, outputTable
share(streamTable(10:0, `date`time`sym`price`volume, [DATE, TIME, SYMBOL, DOUBLE, DOUBLE]), `trades)
share(table(10:0, `timestamp`sym`elapsed`avgPrice`sumVol, [TIMESTAMP, SYMBOL, LONG, DOUBLE, DOUBLE]), `outputTable)
go
// 创建阈值引擎
engine = createThresholdEngine(
    name="thresholdDemo4",
    threshold=1000,
    metrics=<[avg(price), sum(volume)]>,
    dummyTable=trades,
    outputTable=outputTable,
    thresholdColumn=`volume,
    keyColumn=`sym,
    timeColumn=[`date, `time],
    outputElapsedMicroseconds=true
)

// 准备数据
t = table(
    2024.01.02 2024.01.02 2024.01.02 as date,
    09:31:00.000 09:32:00.000 09:33:00.000 as time,
    `A`A`A as sym,
    10.1 10.2 10.3 as price,
    300 400 500 as volume
)

// 插入数据
engine.append!(t)

// 查询结果
select * from outputTable

例5 通过 outputThreshold 输出阈值窗口值和 thresholdColumn 累计和

// 创建 dummyTable, outputTable
share(streamTable(10:0, `date`time`sym`price`volume, [DATE, TIME, SYMBOL, DOUBLE, DOUBLE]), `trades)
share(table(10:0, `timestamp`sym`thresholdValue`cumThreshold`lastPrice, [TIMESTAMP, SYMBOL, LONG, DOUBLE, DOUBLE]), `outputTable)
go
// 创建阈值引擎
engine = createThresholdEngine(
    name="thresholdDemo5",
    threshold=1000,
    metrics=<[last(price)]>,
    dummyTable=trades,
    outputTable=outputTable,
    thresholdColumn=`volume,
    keyColumn=`sym,
    timeColumn=[`date, `time],
    outputThreshold=true
)

// 准备数据
t = table(
    2024.01.02 2024.01.02 2024.01.02 as date,
    09:31:00.000 09:32:00.000 09:33:00.000 as time,
    `A`A`A as sym,
    10.1 10.2 10.3 as price,
    300 400 500 as volume
)

// 插入数据
engine.append!(t)

// 查询结果
select * from outputTable