流计算时延统计与性能优化

• 普通共享流数据表

创建时指定足够大的 capacity 参数。比如对于股票行情数据可以根据历史数据预估好一天的数据总量，将 capacity 设置为略大于该数值。

capacity = 1000000
share(streamTable(capacity:0, `sym`time`price, [STRING,DATETIME,DOUBLE]), `tickStream)

streamTable 函数的 capacity 参数是正整数，表示建表时系统为该表分配的内存（以记录数为单位）。当记录数首次超过 capacity 时，系统会首先分配 capacity 的1.2倍的新的内存空间，然后复制数据到新的内存空间，最后释放原来的内存，再写入新的数据。当记录数继续增加并超过被分配的内存空间时，会再次进行类似的扩容操作。

从时延统计上看，由于扩容涉及到内存分配和复制等较为耗时的操作，因此若某一批数据写入时触发了扩容则会出现一次写入时延的峰值。可以想象，如果 capacity 仅仅设置为 1 ，在持续写入的过程中则会发生多次动态扩容，每次的峰值耗时将随总数据量增加而逐渐增高，因为需要拷贝的数据越来越多。

• 持久化流数据表

创建时指定合理的 cacheSize 和 capacity 参数。在内存充足的情况下，对于股票行情数据可以根据历史数据预估好一天的数据总量，将 cacheSize 和 capacity 设置为略大于该数值。

cacheSize = 1000000
enableTableShareAndPersistence(table=streamTable(cacheSize:0, `sym`time`price, [STRING,DATETIME,DOUBLE]), tableName="tickStream", asynWrite=true, compress=true, cacheSize=cacheSize, retentionMinutes=1440, flushMode=0, preCache=10000)

enableTableShareAndPersistence 函数的 cacheSize 参数是长整型数据类型（long）的整数，表示该表在内存中最多保留多少行，当内存中的行数超过 cacheSize 且确认目前的所有数据都已经保存到磁盘后，系统会申请新的内存空间，将内存中后 50% 的数据拷贝到新的内存空间，并释放掉原来的内存。

从时延统计上看，由于清理内存中的过期数据涉及到内存分配、复制等较为耗时的操作，因此若某一批数据写入时触发了清理则会出现一次写入的时延峰值。若内存充足，则可以考虑设置足够大的 cacheSize 将全部数据都保留在内存里，不触发清理操作。若需要进行内存控制，则设置合理 cacheSize 以平衡峰值出现的频率和单次峰值的大小，因为 cacheSize 的大小决定了拷贝的数据量进而会影响单次的耗时。

假设我们接收实时逐笔成交数据，并对每条成交记录逐条响应，累计最新的当日成交总量，如果每一次计算都是用截止当前的全量成交数据，则会响应性能不佳，而通过增量的流式实现则可以大大提升性能。具体的，最新的累计成交总量可以在上一次计算出的成交总量的基础上加上最新的一条成交记录的成交量得到，可以看到，这里的增量计算需要用到历史状态（上一次计算出的成交总量），我们称之为有状态计算。

DolphinDB 在响应式状态引擎、时序聚合引擎、窗口连接引擎中，提供了大量的内置状态函数，在 metrics 中使用内置状态函数即可以实现上述的增量有状态计算，而历史状态由引擎内部自动维护。比如，在响应式状态引擎中使用 cumsum 函数可以实现增量的累加。

各个引擎已经支持的状态函数请参考用户手册 createTimeSeriesEngine、 createReactiveStateEngine、createWindowJoinEngine，建议优先选择内置状态函数以增量算法实现实时指标。

即时编译 (英文: Just-in-time compilation, 缩写: JIT)，又译及时编译或实时编译，是动态编译的一种形式，可提高程序运行效率。DolphinDB 的编程语言是解释执行，运行程序时首先对程序进行语法分析生成语法树，然后递归执行。在不能使用向量化的情况下，解释成本会比较高。这是由于 DolphinDB 底层由 C++ 实现，脚本中的一次函数调用会转化为多次 C++ 内的虚拟函数调用。JIT 具体介绍请参考 DolphinDB JIT 教程 。

流计算任务由于不断地被触发计算，因此会反复地调用函数，以 1.3 小节响应式状态引擎的计算为例，仅仅注入 10 条数据便会触发 5 次对 metrics 中函数的调用。解释耗时也会反应到整体的计算耗时中，尤其对于响应式状态引擎，在某些场景下总计算耗时中可能大部分为解释耗时，因此，建议通过实现 JIT 版本的函数来减少解释成本 。具体实践请参考 金融因子流式实现教程：即时编译 (JIT) ，文中使用 JIT 优化了移动平均买卖压力指标的性能。

DolphinDB 中的数组向量 (array vector) 是一种特殊的向量，用于存储可变长度的二维数组。这种存储方式可以显著简化某些常用的查询与计算。array vector 具体介绍请参考 DolphinDB array vector 教程。

十档量价是行情快照中最重要的信息，实现高频因子时为了让十档量价数据能够方便地实现向量化计算，在 DolphinDB 中可以通过 fixedLengthArrayVector 函数组合十档数据。针对这个特点，建议直接使用数组向量 (array vector) 来存储 level 2 行情快照中的十档量价字段，以省去函数实现过程中组装十档数据的步骤，来降低计算延时。具体实践请参考请参考 金融因子流式实现教程：数组向量 (array vector) ，文中使用 array vector 优化了移动买卖压力指标的性能。

并不是所有的计算都需要通过流计算引擎进行，一些简单的无状态计算建议在自定义函数中实现。比如，下例希望在引擎中仅对 9:30 之后的数据进行计算，则可以在 handler 函数中首先对输入数据进行过滤再注入计算引擎。

def filterTime(msg){
	tmp = select * from msg where time > 09:30:00.000
	getStreamEngine("reactiveDemo").append!(tmp) 
}
subscribeTable(tableName="tickStream", actionName="demo", offset=-1, handler=filterTime, msgAsTable=true)

• 通过多加一层响应式状态引擎也可以实现同样的过滤，但是由于流计算引擎内部计算时进行分组以及维护状态等开销，在这种无状态计算的场景下，自定义函数是比流计算引擎更高效的实现方式。
• 注意，handler 函数每次处理的是增量的数据，即以上脚本中的 msg 变量是订阅端收到的新增的数据，而不是最新的 tickStream 表的全量快照。

使用 subscribeTable 时设置适当的 batchSize 和 throttle 参数，可以达到微批处理的效果，以此来提升吞吐避免阻塞，进而降低时延。这是基于单次处理的耗时并不是随着单次处理的数据量而线性增长的，最典型的场景是实时写入分布式数据库，一次写入 1000 条数据和写入 10 条数据的耗时可能相差无几，那么发挥数据批处理的性能优势则可以降低整体的时延。

batchSize 和 throttle 参数均表示了触发消费的条件，这两个条件之间是或的关系。订阅队列中的数据积累到 batchSize 设置的量时会触发消费，但是考虑到实际场景中，流量在不同时刻的波动，某些时候可能会长时间达不到 batchSize 而不能被消费，因此需要设置 throttle 参数，通过等待达到一定的时长来触发消费，消费会对队列中未处理的全部数据进行处理。 batchSize 和 throttle 参数建议综合考虑输入数据的流量和数据处理的速度进行调整。

此外，请注意以下两点：

• 修改系统配置项（dolphindb.cfg 或者 cluster.cfg）中 subThrottle=1 后，throttle 才可以被设置为小于1秒。若不修改，则即使在 subscribeTable 函数中指定 throttle 为 0.001，实际值仍为 1 秒。
• 建议在 subscribeTable 时一定要指定 batchSize 和 throttle 参数。若不指定，则每次处理的数据量为到达订阅队列的一块数据块的大小，而不是队列里的全部数据。举个例子，假设此时订阅队列里的仍未处理的数据有 10 条，它们分别是以 2 条、3 条、5 条的大小分三批到达订阅队列的，那么消息处理线程会依次分三批处理这 10 条数据，而不是批处理一次。因此，若输入数据频繁插入，但是每次插入的记录数较少，则未设置 batchSize 和 throttle 参数的订阅可能会表现较差。

使用 subscribeTable 时设置适当的 hash 参数，以提升并行度来降低时延。执行 subscribeTable 函数相当于为其 handler 参数对应的计算函数分配了一个固定的消息处理线程，合理的分配消息处理线程可以尽可能充分里利用 CPU 资源。大致分为以下两类优化建议：

• 将多个不同的流计算任务指定到不同的处理线程
  • 尽量将不同的任务指定到不同的消息处理线程。假设系统中有 3 个流计算任务，若在提交subscribeTable 函数时均指定同一个 hash 值，则它们会被分配到同一个处理线程中，当同时都有新的消息需要处理时会通过轮转的方式执行计算，这种对计算资源的竞争会影响每一个任务的响应时延。
  • 若任务数过多，超过了消息处理线程数量的上限，则建议至少将复杂任务分配到不同的线程，复杂任务指需要长时间占用线程进行计算的任务。
• 将某一个流计算任务拆分到多个线程上并行处理
  • 若输入数据流量过大，一个线程处理不过来，则建议将数据分发到多个线程上并行处理。实现脚本请参考 DolphinDB 流计算教程：并行处理 ，注意，若多个线程均将结果写到同一张共享内存表，那么同时写入时会等待写锁而带来一定的耗时，为了避免这部分等待可以将结果也写入到不同的数据表。
  • 使用 dispatchStreamEngine 进行流数据分发和并行计算，是替代 subscribeTable 的更轻量级的数据分发方式。

DolphinDB 中实现复杂计算指标的思路是把指标分解成多个阶段，每一个阶段由一个计算引擎来完成。如果两个引擎之间由一张中间表来承接中间的计算结果，并且由中间表的发布订阅来串联两个引擎，则会存在一定的内存和耗时的开销。因此，DolphinDB 内置的流计算引擎均实现了数据表（table）的接口，允许将后一个引擎作为前一个引擎的输出，称为流计算引擎级联，与通过多个流数据表与多次订阅串联引擎相比，有更好的性能表现。实现脚本请参考 DolphinDB 流计算教程：流水线处理

前文介绍了将某一个流计算任务拆分到多个线程上并行处理的优化建议，但是通过 subscribeTable 进行发布端过滤并提交多个并行任务的方式，可能会因为订阅的客户端太多而导致发布瓶颈，使得并发处理带来的优化受发布瓶颈的影响而减弱，因为在同一个节点上只有一个发布线程。在 1.30.22、2.00.9 以及之后的版本中，新增了流数据分发引擎（createStreamDispatchEngine），支持将输入的数据分发到不同的线程，并且在该线程中完成将增量的输入数据注入到对应的输出表的操作，这里的输出表可以指定为流计算引擎。

以下脚本中进行了一次发布订阅，在订阅端的消息处理线程中将数据写入到一个分发引擎中，由该分发引擎将数据按 sym 字段哈希分组后分发到 3 个不同的线程进行实际的计算，计算逻辑由响应式状态引擎定义。

//  create engine
share streamTable(1:0, `sym`price, [STRING,DOUBLE]) as tickStream
share streamTable(1000:0, `sym`factor1, [STRING,DOUBLE]) as resultStream
rseArr = array(ANY, 0)
for(i in 0..2){
	rse = createReactiveStateEngine(name="reactiveDemo"+string(i), metrics =<cumavg(price)>, dummyTable=tickStream, outputTable=resultStream, keyColumn="sym")
	rseArr.append!(rse)
}
dispatchEngine=createStreamDispatchEngine(name="dispatchDemo", dummyTable=tickStream, keyColumn=`sym, outputTable=rseArr, mode="buffer", dispatchType="uniform")
//  subscribe
subscribeTable(tableName="tickStream", actionName="dispatch", handler=dispatchEngine, msgAsTable=true)

流数据分发引擎的原理：

• 分发引擎的参数 outputTable 是一个表（或流计算引擎）组成的列表，创建引擎时会创建与之数量相等的多个线程以及多个缓冲队列（或表）。
• 引擎内部的每个线程会负责将缓冲队列中的增量数据注入到对应的输出表（或流计算引擎）。若分发引擎输出到流计算引擎，则此时的计算在分发引擎分配的该线程上进行。
• 分发引擎分配的线程与 subscribeTable 函数分配的消息处理线程是两个独立的概念，即分发引擎分配的线程不属于由系统配置参数 subExecutor 规定的订阅处理线程。

流数据分发引擎的使用注意事项：

• 建议参数 mode 使用默认设置 ”buffer”，在 ”buffer”模式下，总是把缓存表里未处理的全部数据进行批量计算，符合 2.3.1 小节建议的微批处理方式。
• 在某些情况下，参数 dispatchType 设置为 “hash” 时，可能因为哈希分配不均匀而造成数据倾斜，即某一个线程上处理的数据量过大而成为性能瓶颈。这时，为了优化性能建议使用 “uniform” 分配模式，其会将 key 均匀分配到各个输出表中。
  • “hash” 分配规则的效果等同于使用 hashBucket 函数进行分桶，总的桶数确定后，任意一个 key 值经过哈希取模后总是属于固定的一个桶。
  • “uniform” 分配规则是动态分配的，某一个 key 属于哪一个桶受到达引擎的先后顺序影响，每收到一个新的 key 都依次分配其对应的桶。假设一共分 2 个桶，key 值到来的顺序为 A、B、C，则 A 和 C 属于桶1，B 属于桶2。
• 建议分发引擎的线程数不要超过可用的逻辑 CPU 核数，否则频繁的线程切换将会带来一定的开销。