createCrossSectionalEngine

createCrossSectionalEngine(name, [metrics], dummyTable, [outputTable], keyColumn, [triggeringPattern='perBatch'], [triggeringInterval=1000], [useSystemTime=true], [timeColumn], [lastBatchOnly=false], [contextByColumn], [snapshotDir], [snapshotIntervalInMsgCount], [raftGroup], [outputElapsedMicroseconds=false], [roundTime=true], [keyFilter], [updatedContextGroupsOnly=false])

别名：createCrossSectionalAggregator

创建横截面引擎，返回一个以 keyColumn 作为键值的键值表。

该表记录了每组最新记录的时间戳（可以是数据自身的时间戳或者系统时间，由参数 timeColumn 和 useSystemTime 指定）。每次新数据的到来都会更新该表。若指定了 lastBatchOnly = true，此时引擎维护的键值表只会保留最新时间戳上不同组的数据。

• 如果没有指定 metrics 或outputTable，横截面引擎不会利用表中的数据进行任何运算和输出，仅更新表中记录。
• 如果指定了 metrics 和 outputTable，横截面引擎会首先更新表中记录，然后对表中的最新记录进行运算，并把结果输出到 outputTable 中。

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

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

metrics 以元代码的格式表示计算指标，支持输入元组。有关元代码的更多信息可参考元编程。

• 计算指标可以是系统内置或用户自定义的函数，如 <[sum(qty), avg(price)]>；可以对计算结果使用表达式，如 <[avg(price1)-avg(price2)]>；也可以对多列进行运算，如 <[std(price1-price2)]>。
• metrics 内支持调用具有多个返回值的函数，例如 <func(price) as `col1`col2>（可不指定列名）。
• metrics 中使用的列名大小写不敏感，不要求与输入表的列名大小写保持一致。

dummyTable 一个表对象，和输入的流数据表的 schema 一致，可以含有数据，亦可为空表。从 3.00.2 版本开始 dummyTable 支持包含 array vector 类型列。

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

• 如果没有指定 contextByColumn， 则输出表各列的顺序如下：
  • 第一列为 TIMESTAMP 类型，用于存放发生计算的时间戳（如果指定了 timeColumn 则是对应记录的时间戳）；
  • 计算结果列，可为多列。其数据类型必须与 metrics 返回结果的数据类型一致。
  • 耗时列，存储引擎内每个 batch 的计算耗时（单位：微秒），数据类型为 LONG。仅在指定 outputElapsedMicroseconds= true 时输出。
  • 记录数列。存储引擎内每个 batch 的记录数，数据类型为 INT，仅在指定 outputElapsedMicroseconds= true 时输出。
• 如果指定 contextByColumn， 则输出表各列的顺序如下：
  • 第一列为 TIMESTAMP 类型，用于存放发生计算的时间戳（如果指定了 timeColumn 则是对应记录的时间戳）；
  • 第二列为 contextByColumn 指定的列；
  • 计算结果列，可为多列。其数据类型必须与 metrics 返回结果的数据类型一致。
  • 耗时列，存储引擎内每个 batch 的计算耗时（单位：微秒），数据类型为 LONG。仅在指定 outputElapsedMicroseconds= true 时输出。
  • 记录数列。存储引擎内每个 batch 的记录数，数据类型为 INT，仅在指定 outputElapsedMicroseconds= true 时输出。

keyColumn 字符串标量或向量，指定流数据表中某列的值为横截面引擎的键值。横截面引擎的计算时，仅使用每个键值对应的最新一行记录。

timeColumn 是一个字符串。当 useSystemTime = false 时，指定订阅的流数据表中时间列的名称。仅支持 TIMESTAMP 类型。

lastBatchOnly 可选参数。横截面引擎是否只保留最新时间戳上的数据。当 lastBatchOnly = true 时，只能指定 triggeringPattern = 'keyCount'，此时横截面引擎维护的键值表只保存最新时间戳的键值数据；否则，更新并保留所有时间戳上的键值数据。

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

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

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

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

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

注意，自 3.00.5 版本起，流计算引擎暂不支持高可用功能，raftGroup 参数将不生效。

outputElapsedMicroseconds 布尔值，表示是否输出每个 batch 中数据从注入引擎到计算输出的总耗时，以及每个 batch 包含的总记录数，默认为 false，不输出。

roundTime 可选参数，布尔值。用于对第一个数据窗口的起始时间进行规整，仅在 triggeringPattern=dataInterval 时有效。系统将根据该参数、triggeringInterval 和时间精度来确定规整尺度（alignmentSize）。窗口规整规则同时序引擎。

keyFilter 可选参数，以元代码形式表示的过滤条件。过滤条件是一个表达式或者函数调用，应用于截面数据中的列（列值为 key），返回一个布尔向量。引擎会从截面数据中过滤出 key 满足条件的数据进行计算。

updatedContextGroupsOnly 可选参数，布尔标量，默认值为 false。决定是否仅计算相较上次输出存在更新的分组数据。

一个表。

例1. 横截面引擎 csEngineDemo1 触发计算的方式为 "perRow"。6 行数据写入流数据表 trades1，结果亦为 6 行。

share streamTable(10:0,`time`sym`price`volume,[TIMESTAMP,SYMBOL,DOUBLE,INT]) as trades1
share table(1:0, `time`avgPrice`volume`dollarVolume`count, [TIMESTAMP,DOUBLE,INT,DOUBLE,INT]) as outputTable
csEngine1=createCrossSectionalEngine(name="csEngineDemo1", metrics=<[avg(price), sum(volume), sum(price*volume), count(price)]>, dummyTable=trades1, outputTable=outputTable, keyColumn=`sym, triggeringPattern="perRow", useSystemTime=false, timeColumn=`time)
subscribeTable(tableName="trades1", actionName="tradesStats", offset=-1, handler=append!{csEngine1}, msgAsTable=true)
insert into trades1 values(2020.08.12T09:30:00.000 + 123 234 456 678 890 901, `A`B`A`B`B`A, 10 20 10.1 20.1 20.2 10.2, 20 10 20 30 40 20);

select * from trades1;

select * from outputTable;

例2. 横截面引擎 csEngineDemo2 触发计算的方式为 "perBatch"。数据分为 2 个批次写入流数据表 trades2，结果为 2 行。

share streamTable(10:0,`time`sym`price`volume,[TIMESTAMP,SYMBOL,DOUBLE,INT]) as trades2
share table(1:0, `time`avgPrice`volume`dollarVolume`count, [TIMESTAMP,DOUBLE,INT,DOUBLE,INT]) as outputTable
csEngine2=createCrossSectionalEngine(name="csEngineDemo2", metrics=<[avg(price), sum(volume), sum(price*volume), count(price)]>, dummyTable=trades2, outputTable=outputTable, keyColumn=`sym, triggeringPattern="perBatch", useSystemTime=false, timeColumn=`time)
subscribeTable(tableName="trades2", actionName="tradesStats", offset=-1, handler=append!{csEngine2}, msgAsTable=true)
insert into trades2 values(2020.08.12T09:30:00.000 + 123 234 456, `A`B`A, 10 20 10.1, 20 10 20);
sleep(1)
insert into trades2 values(2020.08.12T09:30:00.000 + 678 890 901, `B`B`A, 20.1 20.2 10.2, 30 40 20);

select * from outputTable;

例3. 横截面引擎 csEngineDemo4 触发计算的方式为 "keyCount"。设置 lastBatchOnly = true，只有最新时间戳上的数据会参与计算。因为 metrics 里聚合函数和非聚合函数混合使用，所以，输出记录行数等于输入记录行数，聚合结果重复输出多行。

share streamTable(10:0,`time`sym`price`volume,[TIMESTAMP,SYMBOL,DOUBLE,INT]) as trades1
share table(1:0, `time`factor1`factor2, [TIMESTAMP, DOUBLE,INT]) as outputTable
agg=createCrossSectionalAggregator(name="csEngineDemo4", metrics=<[price+ 0.1, sum(volume)]>, dummyTable=trades1, outputTable=outputTable, keyColumn=`sym, triggeringPattern="keyCount", triggeringInterval=5, useSystemTime=false, timeColumn=`time,lastBatchOnly=true)
subscribeTable(tableName=`trades1, actionName="csEngineDemo4", msgAsTable=true, handler=append!{agg})
num=10
time=array(TIMESTAMP)
time=take(2018.01.01T09:30:00.000,num)
sym=take("A"+string(1..10),num)
price=1..num
volume=1..num
tmp=table(time, sym, price, volume)
trades1.append!(tmp)

// 第二次输入 5 条数据，lastBatchOnly=true，所以只有最新的 5 条数据参与计算。
num=5
time = array(TIMESTAMP)
time=take(2018.01.01T09:30:01.000,num)
sym=take("A"+string(1..10),num)
price=6..10
volume=6..10
tmp=table(time, sym, price, volume)
trades1.append!(tmp)

例4. 横截面引擎 csEngineDemo3 触发计算的方式为 "interval"，每 500 毫秒触发一次计算。向流数据表中写入6行数据，第一次批量写入，间隔 500 毫秒后再次写入。请注意，这里没有指定 useSystemTime 参数为 false，会返回计算发生的时刻。

share streamTable(10:0,`time`sym`price`volume,[TIMESTAMP,SYMBOL,DOUBLE,INT]) as trades3
share table(1:0, `time`avgPrice`volume`dollarVolume`count, [TIMESTAMP,DOUBLE,INT,DOUBLE,INT]) as outputTable
csEngine3=createCrossSectionalEngine(name="csEngineDemo3", metrics=<[avg(price), sum(volume), sum(price*volume), count(price)]>, dummyTable=trades3, outputTable=outputTable, keyColumn=`sym, triggeringPattern="interval", triggeringInterval=500)
subscribeTable(tableName="trades3", actionName="tradesStats", offset=-1, handler=append!{csEngine3}, msgAsTable=true);

insert into trades3 values(2020.08.12T09:30:00.000, `A, 10, 20)
insert into trades3 values(2020.08.12T09:30:00.000 + 500, `B, 20, 10)
insert into trades3 values(2020.08.12T09:30:00.000 + 1000, `A, 10.1, 20)
insert into trades3 values(2020.08.12T09:30:00.000 + 2000, `B, 20.1, 30)
sleep(500)
insert into trades3 values(2020.08.12T09:30:00.000 + 2500, `B, 20.2, 40)
insert into trades3 values(2020.08.12T09:30:00.000 + 3000, `A, 10.2, 20);
sleep(500)

select * from outputTable;

triggeringPattern = "interval" 时，系统每隔 500ms 触发一次计算，输出一条结果。在计算被触发时，即使同一个 key 有多条记录未被计算，系统也只取最新时间戳上的数据进行计算。

// 定义休市和过滤数据的逻辑
def checkTime(time){
    maxTime = max(time)
    if(between(maxTime, 2024.05.22T13:00:00.000:2024.05.22T13:30:00.000)){
        return time > maxTime - 120*60*1000
    } else{
        return time > maxTime - 30*60*1000
    }
}
// 定义截面引擎
share streamTable(10:0,`time`sym`price`volume,[TIMESTAMP,SYMBOL,DOUBLE,INT]) as trades1
t=keyedTable(`sym,10:0,`time`sym`price`volume,[TIMESTAMP,SYMBOL,DOUBLE,INT])
share table(1:0, `time`factor1`factor2, [TIMESTAMP, DOUBLE, DOUBLE]) as outputTable
// 指定 keyFilter 为一个函数，过滤出1小时内未更新的数据。 
csEngine5=createCrossSectionalEngine(name="csEngineDemo5", metrics=<[avg(price), sum(volume)]>, dummyTable=trades1, outputTable=outputTable, keyColumn=`sym, keyFilter=<checkTime(time)>)
// 模拟数据，A1 从 2024.05.22T09:32:00.000 之后未收到新的数据
time = 2024.05.22T09:30:00.000 join 2024.05.22T09:31:00.000 join 2024.05.22T09:32:00.000 join 2024.05.22T13:30:00.000 join 2024.05.22T13:31:00.000
sym=`A1`A2`A1`A2`A2
price=100.0 100.5 100.3 100.8 100.6
volume=100 110 112 200 120
tmp=table(time, sym, price, volume)
csEngine5.append!(tmp)
// 查看截面引擎中的数据
select * from csEngine5

select * from outputTable

在以上的例子中， createCrossSectionalEngine 的返回结果（以下称为横截面表）是为计算提供的一个中间结果，但横截面表亦可为最终结果。例如若需要定时刷新某只股票的最新交易价格， 按照常规思路是从实时交易表中按代码筛选股票并取出最后一条记录，而交易表的数据量是随着时间快速增长的，如果频繁做这样的查询，无论从系统的资源消耗还是从查询的效能来看都不是最优的做法。 而横截面表永远只保存所有股票的最近一次交易数据，数据量是稳定的，对于这种定时轮询的场景非常合适。

要将横截面表作为最终结果，需要在对 createCrossSectionalEngine 中的 metrics 与 outputTable 这两个参数置空。

tradesCrossEngine=createCrossSectionalEngine(name="CrossSectionalDemo", dummyTable=trades, keyColumn=`sym, triggeringPattern=`perRow)

例6. 按照 volume 分组，引擎每插入一行数据，就触发一次计算。

try {
    dropAggregator(`csEngineDemo)
} catch(ex) {}
try {
    dropStreamTable(`trades1)
} catch(ex) {}
share streamTable(10:0, `time`sym`price`volume, [SECOND, SYMBOL, DOUBLE, INT]) as trades1
outputTable = table(1:0, `time`volume`factor1`factor2, [SECOND, INT, DOUBLE, DOUBLE])
agg1 = createCrossSectionalEngine(
    name="csEngineDemo",
    metrics=<[avg(price), sum(volume)]>,
    dummyTable=trades1,
    outputTable=outputTable,
    keyColumn=`sym,
    triggeringPattern="perRow",
    useSystemTime=false,
    timeColumn=`time,
    outputElapsedMicroseconds=false,
    contextByColumn=`volume,
    updatedContextGroupsOnly=false
)

num = 4
time = take(09:30:00, num)
sym = take(`A1`B1`C1`D1, num)
price = rand(100.0, num)
volume = take(1 1 2 2, num)
tmp = table(time, sym, price, volume)
agg1.append!(tmp)
outputTable

以上创建的引擎根据 volume 分组后得到 1 和 2 两个分组。由于设置了 updatedContextGroupsOnly = false，引擎不会检查各个分组的数据是否有更新。每当插入一行数据时，都会触发所有分组的计算和输出，因此， outputTable 中将输出 6 行数据。

try{dropAggregator(`csEngineDemo)}catch(ex){}
try{dropStreamTable(`trades1)}catch(ex){}

share streamTable(10:0,`time`sym`price`volume,[SECOND,SYMBOL,DOUBLE,INT]) as trades1
outputTable = table(1:0, `time`volume`factor1`factor2, [SECOND, INT, DOUBLE, DOUBLE])
agg1=createCrossSectionalEngine(name="csEngineDemo", metrics=<[avg(price), sum(volume)]>, dummyTable=trades1, outputTable=outputTable, keyColumn=`sym, triggeringPattern="perRow",useSystemTime=false, timeColumn=`time, outputElapsedMicroseconds=false, contextByColumn=`volume, updatedContextGroupsOnly=false)
num=4
time = take(09:30:00, num)
sym= take(`A1`B1`C1`D1, num)
price=rand(100.0, num)
volume=take(1 1 2 2, num)
tmp=table(time, sym, price, volume)
agg1.append!(tmp)
outputTable

设置 updatedContextGroupsOnly = true 后，引擎将仅计算本次插入的分组数据，因此，outputTable 中将输出 4 行数据。