共计 5347 个字符，预计需要花费 14 分钟才能阅读完成。
在解决实时流数据时，不仅须要依照工夫做纵向聚合计算，还须要对最新的数据做横向比拟和计算，如金融里对所有股票的最新报价求百分位、工业物联网中计算一批设施的温度均值等。DolphinDB database 提供了横截面聚合引擎，能够对流数据中所有分组的最新数据做聚合运算。
横截面引擎的主体分为两局部：横截面数据表和计算引擎。横截面数据是横截面引擎的外部表，保留了所有分组最新的截面数据。计算引擎是一组聚合计算表达式以及触发器，零碎会依照指定的形式触发聚合运算，计算结果会输入到另外一个表中。
1. 根本用法
在 DolphinDB database 中，通过 createCrossSectionalAggregator 创立横截面聚合引擎。它返回一个横截面数据表，保留了所有分组最新的截面数据，往这个表写入数据意味着这些数据进入横截面聚合引擎进行计算。具体用法如下：
createCrossSectionalAggregator(name, [metrics], dummyTable, [outputTable], keyColumn, [triggeringPattern=”perBatch”], [triggeringInterval=1000])
- name 是一个字符串，示意横截面聚合引擎的名称，是横截面聚合引擎的惟一标识。它能够蕴含字母，数字和下划线，但必须以字母结尾。
- metrics 是元代码。它能够是零碎内置或用户自定义的函数，如 <[sum(qty), avg(price)]>，能够对聚合后果应用表达式，如 <[avg(price1)-avg(price2)]>，也能够对计算列进行聚合运算，如 <[std(price1-price2)]>。详情可参考元编程。
- dummyTable 是表对象，它能够不蕴含数据，但它的构造必须与订阅的流数据表雷同。
- outputTable 是表对象，用于保留计算结果。输出表的列数为 metrics 数量 +1，第一列为 TIMESTAMP 类型，用于寄存产生计算的工夫戳,，其余列的数据类型必须与 metrics 返回后果的数据类型统一。
- keyColumn 是一个字符串，指定 dummyTable 的某列为横截面聚合引擎的 key。keyColumn 指定列中的每一个 key 对应表中的惟一一行。
- triggeringPattern 是一个字符串，示意触发计算的形式。它能够是以下取值：
- “perRow”: 每插入一行数据触发一次计算
- “perBatch”: 每插入一次数据触发一次计算
- “interval”: 按肯定的工夫距离触发计算
- triggeringInterval 是一个整数。只有当 triggeringPattern 的取值为 interval 时才失效，示意触发计算的工夫距离。默认值为 1000 毫秒。
2. 示例
上面通过一个例子阐明横截面聚合引擎的利用。在金融交易中，往往须要实时理解所有股票最新的报价均值、最近一次成交量总和以及最近一次交易的交易量。DolphinDB 的横截面聚合引擎联合流数据订阅性能能够不便地实现这些工作。
（1）创立实时交易表
股票的实时交易表 trades，蕴含以下次要字段：
sym：股票代码
time：工夫
price：成交价
qty：成交量
每当交易产生时，实时数据会写入 trades 表。创立 trades 表的脚本如下：
share streamTable(10:0,timesympriceqty,[TIMESTAMP,SYMBOL,DOUBLE,INT]) as trades
（2）创立横截面聚合引擎
tradesCrossAggregator=createCrossSectionalAggregator(“CrossSectionalDemo”, <[avg(price), sum(qty), sum(price*qty)]>, trades, outputTable, sym, perRow)
tradesCrossAggregator 是横截面数据表，它按股票代码分组，每个股票有且仅有一行。当数据进入该表时，会计算每个股票的 avg(price), sum(qty) 和 sum(price*qty)。每插入一条数据触发一次计算。
（3）横截面数据表订阅实时交易表
subscribeTable(,”trades”,”tradesCrossAggregator”,-1,append!{tradesCrossAggregator},true)
通过流数据订阅性能，把实时数据写入横截面数据表。
（4）模仿数据产生
def writeData(n){
timev = 2000.10.08T01:01:01.001 + timestamp(1..n)
symv = take(AB, n)
pricev = take(102.1 33.4 73.6 223,n)
qtyv = take(60 74 82 59, n)
insert into trades values(timev, symv, pricev,qtyv)
}
writeData(4);
查看实时交易表，共有 4 条数据。
select * from trades
time sym price qty
2000.10.08T01:01:01.002 A 102.1 60
2000.10.08T01:01:01.003 B 33.4 74
2000.10.08T01:01:01.004 A 73.6 82
2000.10.08T01:01:01.005 B 223 59
查看横截面数据表，外面保留了 A、B 两只股票最近的两笔交易记录。
select * from tradesCrossAggregator
time sym price qty
2000.10.08T01:01:01.004 A 73.6 82
2000.10.08T01:01:01.005 B 223 59
查看横截面引擎的输出表，因为横截面引擎采纳了 perRow 每行触发计算的频率，所以每往横截面表写入一行数据，聚合引擎都会做一次计算，因而一共有 4 条记录。
select * from outputTable
time avgPrice sumqty Total
2019.07.08T10:04:41.731 102.1 60 6126
2019.07.08T10:04:41.732 67.75 134 8597.6
2019.07.08T10:04:41.732 53.5 156 8506.8
2019.07.08T10:04:41.732 148.3 141 19192.2
通过 getAggregatorStat 函数查看横截面引擎的状态。
getAggregatorStat().CrossSectionalAggregator
name user status lastErrMsg numRows numMetrics metrics triggeringPattern triggeringInterval
CrossSectionalDemo guest OK 2 3 [avg(price), su…perRow 1000
通过 removeAggregator 函数删除横截面引擎。
removeAggregator(“CrossSectionalDemo”)
3. 触发计算的几种形式
横截面引擎一共有三种触发计算的形式：perRow、perBatch 和 interval。下面的例子中采纳的是每插入一行数据触发一次友链交易计算。上面介绍另外两种触发计算的形式。
- perBatch
perBatch 参数示意每追加一批数据就触发一次写入，下例按 perBatch 模式启用横截面引擎，脚本一共生成 12 条记录，分三批写入，输出表中预期有 3 条记录。
share streamTable(10:0,timesympriceqty,[TIMESTAMP,SYMBOL,DOUBLE,INT]) as trades
outputTable = table(1:0, timeavgPricesumqtyTotal, [TIMESTAMP,DOUBLE,INT,DOUBLE])
tradesCrossAggregator=createCrossSectionalAggregator(“CrossSectionalDemo”, <[avg(price), sum(qty), sum(price*qty)]>, trades, outputTable, sym, perBatch)
subscribeTable(,”trades”,”tradesCrossAggregator”,-1,append!{tradesCrossAggregator},true)
def writeData(n){
timev = 2000.10.08T01:01:01.001 + timestamp(1..n)
symv = take(AB, n)
pricev = take(102.1 33.4 73.6 223,n)
qtyv = take(60 74 82 59, n)
insert into trades values(timev, symv, pricev,qtyv)
}
// 写入三批数据，预期会触发三次计算，输入三次聚合后果。
writeData(4);
writeData(4);
writeData(4);
查看横截面数据表。
select * from tradesCrossAggregator
time sym price qty
2000.10.08T01:01:01.002 A 73.6 82
2000.10.08T01:01:01.003 B 33.4 59
查看输出表。插入了三批数据，因而输出表中有 3 条记录。
select * from outputTable
time avgPrice sumqty Total
2019.07.08T10:14:54.446 148.3 141 19192.2
2019.07.08T10:14:54.446 148.3 141 19192.2
2019.07.08T10:14:54.446 148.3 141 19192.2
- interval
当触发计算的形式为 interval 时，须要指定 triggeringInterval，示意每隔 triggeringInterval 毫秒触发一次计算。上面的例子中，分 6 次写入 12 条记录，每次距离 500 毫秒。设置横截面引擎每 1000 毫秒触发一次计算，预期最终输入 3 条记录。
share streamTable(10:0,timesympriceqty,[TIMESTAMP,SYMBOL,DOUBLE,INT]) as trades
outputTable = table(1:0, timeavgPricesumqtyTotal, [TIMESTAMP,DOUBLE,INT,DOUBLE])
tradesCrossAggregator=createCrossSectionalAggregator(“CrossSectionalDemo”, <[avg(price), sum(qty), sum(price*qty)]>, trades, outputTable, sym, interval,1000)
subscribeTable(,”trades”,”tradesCrossAggregator”,-1,append!{tradesCrossAggregator},true)
def writeData(n){
timev = 2000.10.08T01:01:01.001 + timestamp(1..n)
symv = take(AB, n)
pricev = take(102.1 33.4 73.6 223,n)
qtyv = take(60 74 82 59, n)
insert into trades values(timev, symv, pricev,qtyv)
}
a = now()
writeData(2);
sleep(500)
writeData(2);
sleep(500)
writeData(2);
sleep(500)
writeData(2);
sleep(500)
writeData(2);
sleep(500)
writeData(2);
sleep(500)
b = now()
select count(*) from outputTable
3
如果再次执行 select count(*) from outputTable，会发现随着工夫的推移，输出表的记录数会一直增长。这是因为在 interval 模式下，计算是依照事实工夫定时触发，并不依赖于是否有新的数据进来。
4. 横截面数据表的独立应用
从下面的例子中能够看出，横截面表尽管是为聚合计算提供的一个两头数据表，但其实在很多场合还是能独立发挥作用的。比方咱们须要定时刷新某只股票的最新交易价格，依照惯例思路是从实时交易表中按代码筛选股票并拿出最初一条记录，而交易表的数据量是随着工夫快速增长的，如果频繁做这样的查问，无论从零碎的资源耗费还是从查问的效力来看都不是很好的做法。而横截面表永远只保留所有股票的最近一次交易数据，数据量是稳固的，对于这种定时轮询的场景十分适合。
如果要独自应用横截面表，须要在创立横截面引擎时，把 metrics,outputTable 这两个参数设置为空。
tradesCrossAggregator=createCrossSectionalAggregator(“CrossSectionalDemo”, , trades,, sym, perRow)
