新用户入门指南（金融篇）

阅读本教程前请确保 DolphinDB 已经下载（下载页面：产品 - DolphinDB ）并部署完成（部署教程：部署 ）。

实验环境说明

本教程的实验环境如下：

• server 版本： 3.00.1.2 2024.07.30 LINUX x86_64
• 部署方式：单节点 single mode

  （为确保后续脚本能正常执行，建议用户和本实验环境保持一致）

社区版单节点部署示例（Linux）

在本页面选择下载的版本：

下载后，解压 zip 包，并进入 server 目录：

unzip DolphinDB_Linux64_V3.00.1.2.zip -d <YOUR_PATH>
cd <YOUR_PATH>/server

server 目录中，dolphindb 为 server 可执行文件，请确保该文件有执行权限。

chmod +x dolphindb

dolphindb.cfg 为数据库配置文件，用户可以通过添加或修改配置项，对数据库的日志、内存、性能、网络等方面进行配置。启动前，请配置 localSite 中端口号为空闲端口（默认是 8848）。

参数配置示例：

• 例1：修改数据存储路径，需要在配置文件中添加 volumes 配置项（以配置多个磁盘为例）：

  volumes=/hdd/hdd1/volumes, /hdd/hdd2/volumes

• 例2：需要开启流数据持久化功能，需要在配置文件中添加 persistenceDir 配置项：

  persistenceDir=/home/DolphinDB/Data/Persistence

配置完毕后，启动 server：

• 前台启动

  ./dolphindb

• 后台启动

  sh startSingle.sh

  HINT：windows 版本后台启动脚本为对应的 vbs 文件。

若 server 无法启动，请检查 dolphindb.log 日志中是否有异常信息，并参考文档 节点启动异常 进排查。

启动成功后，用户可以在 web 端输入 <ip:port>，如 localhost:8848 打开 Web 管理器。

部分系统运维函数和库表创建对用户的权限有一定要求。初始条件下，默认用户为 guest，社区用户可以点击右上角的用户头像，选择 Log in 选项，然后通过 admin 账号进行登录：

登录后，点击左侧的 Shell 模块即可执行简单的代码进行测试。

如果要构建项目，用户可以通过 VSCode 或者 GUI 客户端连接 server 执行脚本。以 VSCode 为例，下载 DolphinDB 拓展后，点击左侧的数据库图标，在 setting.json 中必须修改配置参数 “url” 为 “ws://<连接节点 ip>:<连接节点 port>”，其余参数均为可选配置：

配置完成后，可以在任意用户的工作目录下，创建任意一个以 “.dos” 为尾缀的文件，编写 DolphinDB 脚本。脚本执行有两种方式：（1）点击右上角的执行按钮执行；（2）按快捷键 ctrl+E 执行。

数据说明

本教程提供的数据均为模拟数据，范围包含： Level2 的逐笔数据、Level2 快照数据。

参考：金融 Mock 数据生成模块

本教程通过模块 MockData 模拟数据，用户下载附录的 MockData.dos 文件后，需将该文件置于 <homeDir>/modules/ 路径目录下，其中 <homeDir> 默认是 server 主目录（由配置项 homeDir 指定），用户也可以在线调用getHomeDir()函数获取该路径。

模块安装后，在客户端中通过脚本 use MockData 导入该模块后，即可调用模块内部的函数产生模拟数据。

教程代码说明

1. 在执行代码时，请确保代码中的变量 <YOUR_PATH> 已经替换成您自己的路径（建议使用绝对路径），如 “/home/usr/data”。请注意，如文档中无特殊说明，该路径必须是 DolphinDB server 所在物理机的路径而不是客户端的路径。
2. 部分代码依赖上文创建的库表，测试脚本前请参考教程说明，预先建库建表。

快速配置数据库

DolphinDB 数据库的内存、线程、网络、文件、流数据等等资源都可以通过离线的配置项文件或在线函数进行配置和修改。下面简述两种方式的具体操作：

（1）配置文件：下载 DolphinDB 后，打开安装包中的配置项文件，添加或者修改相关配置信息即可。配置文件路径如下：

表 2-1 不同部署场景下的配置项路径说明

（2）在线函数：可以通过 setDynamicConfig 函数在线动态修改部分配置项的值，但注意在线动态修改的配置重启后会失效，若需要永久化需在配置文件进行修改。

快速开始编写 DolphinScript

DolphinDB 编程对以下用户天然友好：

• 有 SQL 编程经验的用户：DolphinDB SQL 语法基本兼容 SQL-92 标准。兼容性可参考 SQL 标准化 。

  t1 = table(2024.01.01 + (0 31 60 91 121) as date, 1.2 7.8 4.6 5.1 9.5 as value)
  t2 = table(2024.02.01 + (0 15 90) as date, 1..3 as qty)
  SELECT * FROM t1 LEFT JOIN t2 ON t1.date = t2.date

• 有 Python 编程经验的用户：DolphinDB 和 Python 一样都是动态类型语言，即声明变量无需指定变量类型。此外，DolphinDB 在函数定义、函数调用、传参方式上都与传统编程语言类似。

  def f(a, mutable b, c=1){
      if(a < b){
          b*=c
      }else{
          b*=-c
      }
      return b
  }
  f(6, 10)

对有编程经验的用户而言，上手写脚本并不会成为学习瓶颈，真正有难度的是如何编写出高性能且简洁的脚本。尤其在金融数据分析领域，对计算响应延时往往要求很高，高性能是用户关注的核心点。

为了提高脚本性能，用户需要牢牢把握两个核心的思维方式：向量式和函数式，这两个思维是编写出高效简洁脚本的关键。

• 向量式编程，简单来说就是允许程序员以数学向量（即一组数值）作为操作的基本单位，而不是单个的标量数值。在向量式编程中，操作和函数可以直接应用于整个数组或矩阵，而不需要使用显式的循环结构。

  该编程方式主要得益于 DolphinDB 在指令集上的优化，例如二元运算时，将 1 对 1的两两元素运算拓展到 n 对 n 的并发计算；或对单个向量计算时，在算法上通过增量计算优化函数的性能。使用经过向量化优化的函数计算，其性能远远高于传统遍历的方式。

  例. 比较循环遍历和向量化计算两个向量和的性能开销。

  n = 10000
  a = rand(10.0, n)
  b = rand(10.0, n)
  c = array(DOUBLE, n)
  timer for(i in 0:n){
    c[i]=a[i]+b[i]
  } // Time elapsed: 22.021 ms
  timer c=a+b // Time elapsed: 0.069 ms

  DolphinDB 绝大部分函数均支持向量计算，内部通过向量化的方式优化了函数的性能，因此要运用好向量化的思维，就要灵活的掌握各种函数的应用。

• 函数式编程，一种以函数为中心的编程范式。函数式编程强调纯函数，即输入决定输出，不依赖外部状态且无副作用。由于没有副作用，多个线程和进程可以并发执行，同时代码方便调试且能够在不同环境进行复用；由于函数链式组合编写代码，代码也会更简洁、可读性也会更高。

  例. 使用函数式编程的方式实现：计算两个内存表存在不同之处的所有行号。

  t1 = table(`A`B`C as id, 1000 2000 1000 as qty, 50.5 60.5 100.0 as price)
  t2 = table(`A`B`C as id, 1000 2100 1000 as qty, 50.5 60.5 100.0 as price)
  
  byRow(all, byColumn(eq, t1, t2)).not().at()
  // 或使用高级函数模式符号替代 
  eq:V(t1, t2).all:H().not().at()

  利用好函数式编程，可以编写出更简洁的脚本。其编程的关键点在于高阶函数、链式调用、纯函数。

为了更好领悟上述的编程特点和编程思想，下文将通过一个完整连贯的案例，一步步带领大家搭建一个基础的 DolphinDB 项目。

针对不同从业方向人群，本教程推荐的阅读指引如下：

• 数据平台开发与运维人员：主要阅读第 3~5 章，这些章节介绍了建库建表、数据导入、数据清洗和预处理等方面的内容。
  • 阅读参考：数据迁移方法 系统运维 故障排查 数据库
• 数据分析人员 ：主要阅读第 5~7 章，这些章节介绍了 SQL 编程（聚合、时序等）、流计算框架（流订阅、引擎级联计算等）等方面的内容。
  • 阅读参考：函数分类 Python 到 DolphinDB 的函数映射 编程语言 流数据
• 每个章节都提供了常见问题的解决引导和下一步阅读的指南，供用户参考。

对于初学者，入门 DolphinDB 时，可以优先在金融场景案例中搜索是否有和自己场景类似的案例，进行阅读实操。在学习的过程如遇问题，建议优先在用户手册和 ask 社区搜索相关问题，如无法解决可以通过社区、邮件等途径寻求技术支持。

分布式是 DolphinDB 实现海量数据存储、并行计算提速的关键架构。分布式架构对用户开放的“接口”叫做分布式表，在 DolphinDB 中指路径以 “dfs://” 开头的数据库下存储的表。

DolphinDB 的分布式是基于分布式文件系统实现的，是一个逻辑上的概念，而不是指物理上的多机存储。即使是单服务器单节点，也可以创建一个分布式表。分布式表的数据按 1 个或多个维度，在逻辑上分成很多个部分，每一个部分称之为分区。

为了保证数据的高可用性、可扩展性，提升系统的负载均衡，用户可以部署一个分布式集群。

在分布式集群中，控制节点负责集群的监控、元数据管理，多个控制节点（至少 3 个）可以构成一个 raft 组，以实现高可用集群部署。数据存储和计算的工作由集群中的数据节点和计算节点负责，控制节点不参与此类工作。数据节点和计算节点既可以部署在同一台物理机，也可以部署在不同的物理机，每台物理机通过一个代理节点对其上部署的数据节点和计算节点进行管理。

数据节点负责数据存储，同时也支持内存计算，在分布式集群中，分区可以按照设定包含固定数量的副本，分散地存储在数据节点上；计算节点只支持计算，不支持存储，它主要用于一些复杂且开销较大的计算任务。

以一个 3 个数据节点的集群为例，数据按天值分区同时按股票代码哈希分区，双副本存储。集群中的数据存储架构如下图所示（最小的蓝色方块表示分区的一个副本）：

分布式存储必然涉及到分布式的存储/查询/计算，因此 DolphinDB 数据库实现了事务对分布式的支持。DolphinDB 的分布式实现以 map-reduce 框架为核心。当用户发起分布式查询时，查询语句会下沉（map）到各个分区实现并行计算，最后汇总（merge）计算结果到查询发起节点，如有必要，还可以基于汇总结果做进一步计算（reduce）。

数据分区主要发挥以下作用：

• 缩小查询范围：如果查询条件包含分区列，则可以通过剪枝操作，快速过滤掉不必要的分区查询，避免全表扫描带来的开销。
• 提高计算性能：将数据分散存储在不同的节点上，系统可以并行处理计算请求，提高数据处理速度。
• 提高可靠性和容错能力：通过数据复制和冗余存储，分布式系统可以在某些节点失败时继续运行，不会丢失数据，从而提高系统的可靠性。节点故障时，也可以从分区副本中恢复数据。
• 增强可扩展性：分布式存储系统可以添加更多的节点来存储数据分区，从而实现水平扩展，以应对数据量的增长。

分区粒度（即每个分区包含的数据量大小）会对查询性能造成影响，这里总结为：

• 分区粒度过大：单分区数据量大，查询时 IO 开销高。受到 license 内存限制，可能导致内存溢出、内存磁盘频繁切换等问题。
• 分区粒度过小：单分区数据量小，查询涉及分区多，查询子任务数量大，将造成节点间通讯成本增加。此外，由于控制节点存储分区元数据，分区过多，控制节点元数据信息爆炸，可能导致控制节点内存不足。最后，由于涉及的分区多，导致系统读写频繁，从而造成很多低效的磁盘访问（小文件读写)，造成系统负荷过重。

用户建表时需要指定分区字段，系统将依据分区字段的值按照分区模式划分分区，分区模式由用户在建库时指定。下面结合图例对不同分区模式的机制进行说明，假设原始数据为：

• VALUE 分区：用户需要指定分区字段的部分值（以列表形式提供，如以日期分区，可以提供 2022.01.01..2022.01.31），数据依据分区字段值进行分区，一个值对应一个分区。

  注意：官网下载的安装包中的配置项默认设置了 newValuePartitionPolicy=add，即支持写入不在值分区内的数据时，值分区也可以自动拓展。因此，用户在配置 VALUE 分区的模式时，可以只指定少量字段值，以避免元数据过多（例如用户设置了 2000.01.01..2030.01.01，会造成分区元数据很大）。

  
• HASH 分区：用户指定固定的 HASH 分区数，数据依据分区字段值调用哈希函数 hashBucket 映射后的结果进行分区，每个哈希值对应一个分区。
• RANGE 分区：用户指定分区范围，数据依据分区字段值划分到对应的范围中，一个范围对应一个分区。范围分区不支持自动拓展，用户需要手动调用函数 addRangePartitions。
• LIST 分区：用户指定以列表形式指定哪些值属于哪些。实践中，金融场景客户很少使用这种分区方式。
• COMPO 分区：将多个字段的不同的分区方案进行组合。

金融领域通常将时间字段作为分区字段，指定分区模式时，最常用的是 VALUE 和 RANGE 分区，如日 K 线数据按照年划分 RANGE 分区，分钟 K 线按照日划分 VALUE 分区等。但是部分场景下，仅依靠一层分区，仍无法控制分区在合理的范围，此时可以通过 COMPO 组合分区进行更细粒度的划分，例如逐笔数据按照日划分 VALUE 分区，同时按照股票代码 HASH 分区。 HASH 分区是灵活调整分区粒度的一个重要的方式。

调整分区粒度的方法：

• 降低分区粒度：（1）采用 COMPO 分区；（2）增加分区个数；（3）将 RANGE 分区改为 VALUE 分区。
• 增加分区粒度：（1）采用 RANGE 分区取代 VALUE 分区；（2）HASH 分区减小哈希映射数。

合理的分区大小推荐：

• OLAP 引擎：100MB - 300MB （压缩前）
• TSDB/PKEY 引擎：400MB - 1GB（压缩前）

选择合适的存储引擎：

分布式数据可选的存储引擎有 TSDB, OLAP 和 PKEY，其中 TSDB 和 OLAP 是较为基础的两个存储引擎。TSDB 引擎相较于 OLAP 引擎而言，底层设计更加复杂，详情可参考 TSDB 存储引擎简介 。从用户层面来说，最显著的功能特点就是 TSDB 除了分区机制外，还额外增加一层字段索引机制，因而更适用于大数据点查的场景。对于较小的数据量可以继续用 OLAP 存储甚至存储为无分区的维度表。如果对数据有主键约束的需求可以使用 PKEY 引擎进行存储。

常见的金融数据分区方案参见：存储金融数据的分区方案最佳实践。此处摘选了一部分股票的分区方案供参考：

表 3-1 不同种类股票数据的分区方案示例

DolphinDB 可以通过函数建库建表，也可以使用 SQL 进行建库建表，本教程以 SQL 编程进行示例，相关参考文档 create。

分区

1. 创建数据库时如何按年进行分区？

   DolphinDB 不支持年这一数据类型，因此无法按年进行值分区。按年分区需使用 RANGE 分区进行，可参考下述脚本：

   create database "dfs://yearDB" partitioned by RANGE(date(2010.01M+12*1..10)), engine='TSDB'

2. 建库建表完成后，是否支持修改分区方案？

   由于数据存储是依赖于分区的，因此不支持直接修改分区方案。最佳做法是重新创建一个新分区方案的库表，然后将数据迁移过去，最后删除原来的库表。

   可以借助 mr + repartitionDS函数进行数据迁移（以本教程的快照数据为例）：

   // 创建一个新的分区方案的数据库表
   drop database if exists "dfs://stock_lv2_snapshot_1" 
   create database "dfs://stock_lv2_snapshot_1" partitioned by VALUE(2022.01M..2022.12M), HASH([SYMBOL, 10]), engine='TSDB'
   
   // 省略部分表字段代码
   create table "dfs://stock_lv2_snapshot_1"."snapshot"(
       TradeDate DATE[comment="交易日期", compress="delta"]   
       TradeTime TIME[comment="交易时间", compress="delta"]
       ...
   )
   partitioned by TradeDate, SecurityID,
   sortColumns=[`SecurityID,`TradeTime],
   keepDuplicates=ALL
   
   // 由于 repartitionDS 只能按照一个字段重新分区，这里选用了不会造成新库表写入分区冲突的 TradeDate 字段
   snapshot = loadTable("dfs://stock_lv2_snapshot", "snapshot")
   ds = repartitionDS(<select * from snapshot>,`TradeDate, VALUE, 2022.01M..2022.12M);
   
   // 定义 map 函数用于写入新的库表
   def writeDS(dst_dbPath, dst_tbName, data){
       dstTb=loadTable(dst_dbPath, dst_tbName)
       dstTb.append!(data)
       flushTSDBCache()
   }
   // mr 迁移数据，如果内存不够大：
   1. 串行迁移，即 parallel = false
   2. ds 数据源拆分成更小的单元，如按日拆分
   mr(ds, writeDS{"dfs://stock_lv2_snapshot_1", "snapshot"}, parallel=false)
   
   // 删除旧的库表
   drop database if exists "dfs://stock_lv2_snapshot" 

   社区版默认配置执行上述代码时，可能会报错：The size of the write table cannot exceed the size of cacheEngine. 或者 Out of Memory. 推荐根据注释方案调整写入粒度，参考脚本如下：

   ds = repartitionDS(<select * from snapshot>,`TradeDate, VALUE, 2022.01.01..2022.12.31);
   mr(ds, writeDS{"dfs://stock_lv2_snapshot_1", "snapshot"}, parallel=false)

数据库&数据表

1. 表字段命名的规范是什么？能否以数字开头？

   创建表时，列名必须由中文或英文字母、数字或下划线 (_)组成，且必须以中文或英文字母开头。

   有以下几种场景可能出现特殊列名，即不满足上述需求的列名：

   • as 函数取别名：table(1 2 3 as "_aa")
   • addColumn 添加列：t.addColumn("_bb", INT)
   • pivot by 行列转换：

     t = table(2022.01.01 + 1..5 as date, `a`a`c`c`b as sym, 1..5 as val)
     select last(val) from t pivot by sym, date
     /* output:
     sym 2022.01.02 2022.01.03 2022.01.04 2022.01.05 2022.01.06
     --- ---------- ---------- ---------- ---------- ----------
     a   1          2                                          
     b                                               5         
     c                         3          4       
     */
     

2. 是否支持修改数据库名、数据表名、字段名?

   不支持修改数据库名，但是可以通过函数 renameTable 和 rename! 修改表名和字段名，但需要注意 TSDB 引擎存储的分布式表不支持对字段进行改名、修改值和删除的操作。

3. 建库时 "dfs://" 路径是什么含义？此类库中的表存储在什么路径下?

   "dfs://" 用于声明数据库的类型是分布式数据库，除此以外，还有 "oltp://" 用于声明在线事务处理数据库。所有 "dfs://" 开头的数据库中的数据都存储在 volumes 参数配置的路径下，可通过 getConfig("volumes") 查看。

   单节点模式下，volumes 的默认目录是 <HomeDir>/storage。集群模式下，默认目录是<HomeDir>/<nodeAlias>/storage。

   

4. 存储时，推荐选择宽表还是窄表？各自的适用的场景是什么？

   • 宽表：截面计算便捷，数据写入速度上优于窄表；增删改基于字段操作，开销较大。

   • 窄表：增删改基于记录操作，比较便捷开销较小；截面计算不便，需要通过 pivot 转换为宽表。

   以中高频因子存储为例，由于增删因子操作较为频番，综合性能考虑窄表优于宽表，具体可参考 中高频多因子库存储最佳实践 中的分析步骤去评估是存储为宽表还是窄表。

5. 存储时，推荐选择分布式表还是维度表？各自的适用的场景是什么？

   首先要理解维度表的概念，维度表实质上是一个持久化且不进行分区的表。前文提到分区的主要目的是将海量数据划分为小块，方便快速定位查找，因此如果数据量较小且需要持久化，则推荐存储为维度表，否则存储为分布式表。

6. 表中是否可以存储字典、多维数据等对象？

   不支持存储字典，如果是类似 json 数据，可以通过 BLOB 超长字符串的格式存储在表中；表中只能存储数组向量、列式元组（仅内存表）等类型强一致的二维对象，不支持超过二维或者类型不一致的二维数组存储。

7. 是否支持多线程并发写入数据表，目前报错：filepath xxx has been owned by transaction.

   DolphinDB 默认不允许对同一分区进行并发写入，否则会造成分区冲突，报上述错误。有一个方案是创建数据库时配置 atomic='CHUNK'，设置后，系统允许用户多线程并发写入同一个分区，但由于不能完全保证事务的原子性，可能出现部分分区写入成功而部分分区写入失败的情况。同时由于采用了重试机制，写入速度可能较慢。

事务

1. 删除数据库是事务操作吗？是否支持回滚？

   是的，如果删除数据库（dropDatabase）或者删除分区（dropParition）失败，系统会自动回滚事务。

2. 删除数据时产生下述报错，如何解决：

   • 报错：deleteSubChunks failed, path "/xxx" does not exist.
   • 报错：deleteSubChunks failed on '/dbName, chunk xxxx is not in COMPLETE state'

   该报错是在数据节点上删除数据时，找不到对应的 chunk 分区信息时产生的。

   下面给出一些此类问题的排查思路：

   1. 首先在控制节点调用函数 getClusterChunksStatus 函数查看对应数据库下的分区状态信息，如果报错给出了 chunk 信息，则重点查看该 chunk 的状态。需要重点关注并记录的信息有 version（chunk 版本），state（chunk 状态） 和 replicaCount（副本数）。
   2. 然后在数据节点调用 pnodeRun(getAllChunks{"/dbName%"}) 统计一下所有 chunk 的副本数是否有缺失，然后针对报错的 chunk 记录一下 version 和 state 信息。
   3. 比较控制节点和数据节点 version 是否一致；检查副本数是否与控制节点记录的元数据一致；确认 chunk 状态是否都为完成状态（COMPLETE）。

   若排查得到分区不一致，推荐将上述信息截图后提交给官方进行排查，有运维经验的用户可以参照教程 分区状态不一致 中的方法尝试解决。

   • 报错：<ChunkInRecovery>deleteSubChunks failed on '/xxx', chunk xxx is in RECOVERING state.若等待分区恢复耗时过久，则可以通过下述脚本强制删除处在恢复状态的分区：

     use ops
     dropRecoveringPartitions(dbPath="dfs://xxx", tableName="zzz")
     // tableName 为可选参数，可不指定

概念学习

数据分区

分布式事务

实操案例

存储金融数据的分区方案最佳实践

中高频多因子库存储最佳实践

DolphinDB 支持多种文件格式导入、从数据库迁移、实时数据订阅等多种数据导入方式。

下表整理列出了数据导入 DolphinDB 数据库的不同方式、对应方法以及相关参考文档的链接：数据迁移方法。

表 4-1 DolphinDB 中的数据导入方法

installPlugin("hdf5")
loadPlugin("hdf5")

installPlugin("parquet")
loadPlugin("parquet")

loadPlugin("mysql")

python datax.py ../job/OracleToDolphinDB.json

installPlugin("kafka")
loadPlugin("kafka")

// 创建消费者
config = dict(STRING, ANY)
config["metadata.broker.list"] = "127.0.0.1:9092"
config["group.id"] = "kafkaDemo"
consumer = kafka::consumer(config)

// 订阅对应主题的数据
kafka::subscribe(consumer, ["demo"])

// 创建消费线程
def kafkaJsonParse1(msg,key,topic){
  return parseJsonTable(msg)
}
conn = kafka::createSubJob(consumer,st,kafkaJsonParse1,"kafka1Structural")

// 订阅流表数据入库
subscribeTable(tableName = `st,actionName = `demo,
                offset = 0,handler = loadTable("dfs://demo","test"),
                msgAsTable=true,batchSize = 1024)

installPlugin("WindTDF")
loadPlugin("WindTDF")
go

// 建立 handle

handle = WindTDF::createHandle(HOST, PORT, USERNAME, PASSWORD, dict(["outputRecvTime", "outputElapsed"], [true, true]));
// 订阅上交所的股票快照
schema = WindTDF::getSchema(`snapshot);
snapshot_sh_s = streamTable(10000:0, schema[`name], schema[`typeString]);
share snapshot_sh_s as snapshot_sh_s1;
WindTDF::subscribe(handle, snapshot_sh_s1, "SH-2-0", "snapshot");

// 订阅上交所的期权快照
schema = WindTDF::getSchema(`futures);
futures_sh_s = streamTable(10000:0, schema[`name], schema[`typeString]);
share futures_sh_s as futures_sh_s1;
WindTDF::subscribe(handle, futures_sh_s1, "SH-1-1", "futures");

// 获取已有的 handle
handle = WindTDF::getHandle();

// 建立连接
WindTDF::connect(handle)

// 查看订阅状态
WindTDF::getStatus()

// 关闭连接
WindTDF::close(handle)

// 取消已有的订阅（非必要）
WindTDF::unsubscribe(handle, "SH-2-0", "snapshot");
WindTDF::unsubscribe(handle, "SH-1-1", "futures");

// 移除handle
WindTDF::delete(handle)

本教程作为入门教程，将以最基础的文本文件导入数据作为示例，介绍导入的详细过程，以及导入中可能碰到的问题。

本章节使用的历史数据通过模拟产生，数据类型为快照数据，存储方式为 csv 文件。注意，导入的 csv 文件需要和 DolphinDB server 位于同一台服务器，且启动 DolphinDB server 的用户需要有该文件的读权限。

很多场景下，数据入库前需要进行预处理操作，包括：

• 类型转换：将时间类型转换成 DolphinDB 可以识别的格式；格式化小数数值等
• 字段处理：合并多个字段、过滤掉不必要的字段、将字段重排序、增加字段等
• 记录处理：对记录进行异常值过滤；跳过、过滤部分记录等

下文将场景结合具体的案例进行说明。

1. 由于内存溢出无法写入/查询？

如果是社区版本，由于 8G 的内存限制，在导入时可能会出现内存溢出的情况。此时，getRecentJob 的 errorMsg 字段可能会出现以下报错：

loadDataByDay: loadTextEx(database(dbName), tbName, ["TradeDate","SecurityID"], fileName, , schema, , , ["SecurityID","TradeTime"]) => Out of memory
loadDataByDay: loadTextEx(database(dbName), tbName, ["TradeDate","SecurityID"], fileName, , schema, , , ["SecurityID","TradeTime"]) => [appendDFSTablet] Failed to save table on data node, with error: Out of memory. RefId:S01077

若发生溢出情况，可以提前结束其他任务：

// 取消单机的后台作业
cancelJob(getRecentJob(242).jobId)

// 取消集群节点上所有后台作业
use ops
cancelJobEx()

为了能弄清楚造成溢出的原因，我们逐步进行排查。本教程仅列出基础的排查方式，详细内容参考 Out of Memory 。

内存占用的对象包括：临时变量、读写缓存、元数据缓存等等。

• 临时变量占用内存量

  获取内存中当前所有临时变量的内存占用（单位：MB）。

  tempVarSize = exec sum(bytes) \ 1024 \ 1024 from objs(true)

  清除不必要的临时变量，可以通过 undef 函数进行操作。

  undef(obj, [objType=VAR]) // 清除某个变量
  undef all // 清除所有变量

• 缓存占用内存量

  获取所有缓存的内存占用（单位：B）。

  getSessionMemoryStat()

对缓存进行清理。每类缓存对应的清理方法如下表所示：

表 4-3 系统缓存类型及其对应的清理方式

• 其他内存占用排查方式

  如果清理缓存和清理临时变量后仍然无法解决问题，那么可以参照下述方法进行检查：

  （1）内存、缓存相关的配置项是否设置合理，下面列举部分可能造成内存大量占用的场景。

  内存相关配置请参照 功能配置 内存、系统内存、Cache Engine 小节。

  表 4-4 TSDB 内存相关配置项

（2）数据库分区或索引等配置不合理导致元数据爆炸。下表列举了几个可能的场景：

表 4-5 导致内存中元数据膨胀的场景

2. 写入时报错列字段不一致，The column [xxx] expects type of DATETIME, but the actual type is DOUBLE.

解决方案：这通常是由于写入数据的字段顺序和表字段的顺序不一致造成的。DolphinDB 不支持表字段根据字段名自动对应，也不支持 insert into 指定字段名进行写入。需要在写入时确保写入数据的类型和字段顺序能够与表定义的顺序完全一致。

3. 写入函数执行没有报错，但是数据没有成功写进去。

解决方案：请检查一下建库时的分区的类型和分区模式。

如是 VALUE 分区，请检查配置项 newValuePartitionPolicy 是否设置为 add，默认下载的 DolphinDB 配置文件中会设置为 add 策略，即 VALUE 分区自动增加；或者调用 addValuePartitions 手动增加分区。

配置文件，如下图所示（以单机为例）：

如是 RANGE 分区，不支持自增，请检查写入数据的范围是否超过了建库时设置的范围，如超过了需要手动调用函数 addRangePartitions 增加分区，然后再写入数据。

• 文本导入：文本数据导入 金融 PoC 用户历史数据导入指导手册之股票 Level-2 逐笔篇
• 数据清洗： 利用 DolphinDB 高效清洗数据
• 行情数据：easyNSQ 实时行情数据接入功能模块 easyTLDataImport 通联历史数据自动化导入功能模块
• 数据迁移和导入插件：插件

在 DolphinDB 中可以通过三种方式写入数据，分别为：

• tableInsert 函数（推荐）：支持任意表对象的写入，会返回写入的记录数。
• append! 函数：支持任意表对象的写入，会直接对原表进行修改。
• insert into 语句：适合于数据量较少的场景，支持内存表写入；从 3.00.1 版本开始，若配置 enableInsertStatementForDFSTable 配置项，则 insert into 也支持分布式表写入。

例：使用三种方式向快照表里追加股票的数据。

use MockData

dbName = "dfs://stock_lv2_snapshot"
tbName = "snapshot"

// 模拟一只股票的数据
date = 2023.01.03
t = stockSnapshot(tradeDate=date, securityNumber=1)

使用 tableInsert 写入：

re2 = loadTable(dbName, tbName).tableInsert(t)
re2 // output: 4802

使用 append! 写入：

re1 = loadTable(dbName, tbName).append!(t)
select count(*) from re1

使用 insert into 语句写入：

getConfig(`enableInsertStatementForDFSTable) // true

insert into snaptb values(2023.01.03,09:30:00.000,'',"000001",'',"OCALL",2,1.045,37,0,10001.799999999999272,
    1.4,1.41,1.631,1.633,0,0,0,0,0,0,0,22300,7.002,0,12100,7.002,0,0,0,0,0,
    array(DOUBLE[]).append!([[6.940000000000001,6.940000000000001,6.950000000000001,6.97,6.980000000000001,6.990000000000001,7.02,7.05,7.08,7.1]]),
    array(DOUBLE[]).append!([[7.09,7.06,7.05,7.04,7.030000000000001,7.01,6.990000000000001,6.980000000000001,6.950000000000001,6.92]]),
    array(INT[]).append!([[32400,32400,33300,36900,36900,36900,39600,41400,41400,44100]]),
    array(INT[]).append!([[48600,48600,47700,46800,42300,41400,40500,40500,34200,31500]]),
    array(INT[]).append!([[20,20,18,18,17,17,16,16,15,13]]),
    array(INT[]).append!([[10,10,12,16,17,17,17,18,18,19]]),
    0,0,0,0,0,0,0,0,1,1000,2454.300000000000181,30,45000,31392.200000000000727,237,237,204,210,236,226,10:57:45.000,1,
    array(LONG[]).append!([[,0,0,0,3600,5200,5200,12400,15600,15600]]),
    array(LONG[]).append!([[,0,100,3500,3600,3900,5900,12400,13500,15600]]))

思考题：

（1）对比上述三种写入数据的方式，对于分布式表而言，哪种写入方式是最方便的？

（2）使用 insert into 写入数据到分布式表，对写入性能有何影响?

在 DolphinDB 中内存对象可以直接通过切片、索引下标取数，且支持应用于函数直接进行计算，不光是向量、矩阵支持这种特性，内存表也同样支持。但是对于分区表、分布式表仅支持使用 SQL 语句进行访问。

表 5-1 不同类型表的访问方式

t[1]
t[0:3]

sum(t)
cumsum(t)

select * from t

例 1：查询某日的所有股票数据。

dbName = "dfs://stock_lv2_snapshot"
tbName = "snapshot"
select * from loadTable(dbName, tbName) where TradeDate = 2023.01.03

例 2：查询某日股票部分字段的数据。

dbName = "dfs://stock_lv2_snapshot"
tbName = "snapshot"
select concatDateTime(TradeDate, TradeTime) as TradeDateTime,
       SecurityID, OfferPrice[0] as offerPx1, 
       BidPrice[0] as bidPx1 
from loadTable(dbName, tbName) 
where TradeDate = 2023.01.03

在 DolphinDB 中，为了应对不同的删除场景，系统支持按行、按分区以及全表删除等多种方式。

DolphinDB 的数据更新和数据删除的机制类似，每次更新时，由于 MVCC 机制，系统会额外产生数据副本存储在磁盘上，由系统定时回收；对于 TSDB 引擎，在配置数据保留策略为保留最新一条的场景下，更新也和软删除类似，以追加的方式实现。

最基础的更新方式就是通过 update 语句实现，如果要基于某些字段组成的键进行更新（该键存在）或追加（该键不存在），则可以使用 upsert! 函数。

例 1. 向快照表追加并更新买卖 1 档报价的信息。

dbName = "dfs://stock_lv2_snapshot"
tbName = "snapshot"

// 分布式表无法通过 update 直接追加字段，需要通过 addColumn
loadTable(dbName, tbName).addColumn(["OfferPx1", "BidPrice1"], [DOUBLE, DOUBLE])
// 更新更新买卖 1 档报价的信息
update loadTable(dbName, tbName) set OfferPx1 = OfferPrice[0], BidPrice1 = BidPrice[0] where TradeDate == 2022.12.01

例 2. 向快照表写入数据时，希望根据时间和股票代码进行去重写入。

// 模拟写入的数据
use MockData
date = 2023.01.03
t = stockSnapshot(tradeDate=date, securityNumber=1)
// 如果执行了增加列字段的代码，需要执行此步
update t set OfferPx1 = OfferPrice[0], BidPrice1 = BidPrice[0] 

// 向分布式表进行去重写入
loadTable(dbName, tbName).upsert!(t, keyColNames=["TradeDate", "TradeTime", "SecurityID"])

• SQL 编程：SQL 标准化 SQL 编写案例
• 增删改查：数据库操作 分布式表数据更新原理和性能 软删除

对于已经创建的库表，DolphinDB 可以借助 database 函数进行加载，参考脚本如下：

dbName = "dfs://stock_lv2_snapshot"
snapdb = database(dbName)
schema(snapdb)

如果是在对应数据目录（cataLog）下创建的库表，则可以用 “use catalog xxx” 的方式加载数据目录，例如：

use catalog trading
select count(*) from stock_lv2_snapshot.snapshot

详情请参考 数据目录 。

同样，分布式表的加载也可以借助 loadTable 函数进行，参考脚本如下：

dbName = "dfs://stock_lv2_snapshot"
tbName = "snapshot"

snaptb = loadTable(dbName, tbName)
select count(*) from snaptb
// 或者 select count(*) from loadTable(dbName, tbName)

除了借助函数，还可以通过 “数据库句柄”.”数据库表名” 访问分布式表：

dbName = "dfs://stock_lv2_snapshot"
snapdb = database(dbName)

select count(*) from snapdb.snapshot

DolphinDB 在 SQL 查询、更新和删除数据时，支持根据 where 条件对查询进行优化。最常用的优化之一就是如果过滤条件包含分区字段，则查询、更新和删除时，系统能够支持按照分区信息进行剪枝处理，减少扫描的分区数。

例 1. 查询 2022.12.01 ~ 2022.12.03 日期范围内的数据

触发分区剪枝的场景：使用 between 语句、非链式的过滤条件

select [HINT_EXPLAIN] * from loadTable(dbName, tbName) where TradeDate between 2022.12.01 and 2022.12.03
select [HINT_EXPLAIN] * from loadTable(dbName, tbName) where TradeDate >= 2022.12.01 and TradeDate <= 2022.12.03
// output:
// 可以看到查询遍历的分区数为 90
"map": {\n' +
'            "partitions": {\n' +
'                "local": 90,\n' +
'                "remote": 0\n' +
'            },\n' + "sql": ... 
'    }

无法触发分区剪枝的场景：链式条件过滤

select [HINT_EXPLAIN] * from loadTable(dbName, tbName) where 2022.12.01 <= TradeDate <= 2022.12.03

// output:
// 可以看到查询遍历的分区数为 10891
"map": {\n' +
'            "partitions": {\n' +
'                "local": 10891,\n' +
'                "remote": 0\n' +
'            }

例 2. 查询某只、某几只股票的数据

使用等值查询，in 查询都可以触发分区剪枝。

select [HINT_EXPLAIN] * from loadTable(dbName, tbName) where SecurityID = "000001"

// output:
// 可以看到查询遍历的分区数为 243
"map": {\n' +
'            "partitions": {\n' +
'                "local": 243,\n' +
'                "remote": 0\n' +
'            }
  
  
select [HINT_EXPLAIN] * from loadTable(dbName, tbName) where SecurityID in ["000001", "000002"]
// output:
// 可以看到查询遍历的分区数为 485
"map": {\n' +
'            "partitions": {\n' +
'                "local": 485,\n' +
'                "remote": 0\n' +
'            }

例 3. 查询股票 ”000001“ 在 2022 年 2 月的快照数据。

对于时间类型的分区列应用时间精度更低的函数，也可以进行分区剪枝，缩小分区范围。

select [HINT_EXPLAIN] * from loadTable(dbName, tbName) where month(TradeDate) = 2022.02M and SecurityID = "000001"

// output:
// 可以看到查询遍历的分区数为 16
"map": {\n' +
'            "partitions": {\n' +
'                "local": 16,\n' +
'                "remote": 0\n' +
'            }

与 OLAP 引擎不同，TSDB 引擎额外维护了一个键值索引（通过配置参数 sortColumns 指定），因此除了分区剪枝外，在过滤条件中指定键值字段，也可以加速查询。索引的组织形式详见 TSDB 存储引擎详解 3.3.1 节对索引键的说明。

可以通过 schema 函数查看 sortColumns 配置的值：

dbName = "dfs://stock_lv2_snapshot"
tbName = "snapshot"
loadTable(dbName, tbName).schema().sortColumns
// output: ["SecurityID","TradeTime"]

该sortColumns 属性显示索引键是 securityID（除最后一个时间列外的其他字段构成索引键），且每个索引对应的数据按照 TradeTime 顺序存储在磁盘上。SecurityID 字段既是分区列又是索引键，如果查询条件包含 securityID 不仅可以进行分区剪枝，还可以根据内存中维护的索引信息快速定位到数据块。

以查询某天某只股票的数据为例，通过 [HINT_EXPLAIN] 获取 SQL 执行计划，观察是否命中索引：

select [HINT_EXPLAIN] * from loadTable(dbName, tbName) where TradeDate = 2022.12.01 and SecurityID = "000002"
// output:
// 打印信息的 where 部分包含查询涉及扫描的数据块个数和命中的索引条件数
"where": {\n' +
'            "TSDBIndexPrefiltering": {\n' +
'                "blocksToBeScanned": 62,\n' +
'                "matchedWhereConditions": 1\n' +
'            }

在金融场景下，聚合指标是策略不可或缺的一部分，有助于帮助交易员、分析员快速识别市场趋势，做出基于数据的策略调整。以股票数据为例，常用的聚合方式有以下两种：

• 按照交易时间聚合
• 按股票标的聚合

下面综合上述场景，结合具体例子来进行说明。

例1：基于历史数据计算 5 分钟 K 线高开低收（OHLC）和总量。

// 模拟数据
n = 1000000
date = take(2019.11.07 2019.11.08, n)
time = (09:30:00.000 + rand(int(6.5*60*60*1000), n)).sort!()
timestamp = concatDateTime(date, time)
price = 100+cumsum(rand(0.02, n)-0.01)
volume = rand(1000, n)
symbol = rand(`AAPL`FB`AMZN`MSFT, n)
trade = table(symbol, date, time, timestamp, price, volume).sortBy!(`symbol`timestamp)

// 计算 OHLC 和 Volume
barMinutes = 5
OHLC = select first(price) as open, max(price) as high, min(price) as low, last(price) as close, sum(volume) as volume from trade group by symbol, date, bar(time, barMinutes*60*1000) as barStart

这里用到了四个函数 max, first, min, low 分别用于计算窗口内数据的高开低收聚合值，sum 用于计算总量。

K 线窗口借助 group by 分组语句进行划分：脚本 group by symbol, date, bar(time, barMinutes*60*1000) as barStart 表示按照股票代码、日期、5分钟时间窗口划分，其中 bar 函数用于按照时间将每 5 分钟的数据划分为一组。

例 1 只是计算 K 线最精简的一个例子，在实际场景中可能还会涉及非常复杂的数据处理过程，本节将在例 2 进行具体的讲解。

例2：基于历史快照数据合成分钟 K 线。如果某只股票在某一分钟无数据，则填充为 0。

Step1. 变量函数定义

• 自定义最高价最低价计算函数

// 自定义计算最高最低点的聚合函数
// defg 用于声明聚合函数

defg high(DeltasHighPrice, HighPrice, LastPrice){
	if(sum(DeltasHighPrice)>0.000001){
		return max(HighPrice)
	}
	else{
		return max(LastPrice)
	}
}

defg low(DeltasLowPrice, LowPrice, LastPrice){
	sumDeltas = sum(DeltasLowPrice)
	if(sumDeltas<-0.000001 and sumDeltas!=NULL){
		return min(iif(LowPrice==0.0, NULL, LowPrice))
	}
	else{
		return min(LastPrice)
	}
}

Step2. 数据预处理

• 把09:25:00-09:30:00的数据归入第一根 K 线：[09:30:00, 09:31:00]
• 计算同一个股票或者基金的两笔相邻快照的最高价和最低价的变化幅度
• 计算同一个股票或者基金的两笔相邻快照的成交量、成交金额和成交笔数的增量

dbName = "dfs://stock_lv2_snapshot"
tbName = "snapshot"

tempTB1 =	select	TradeDate,
			iif(TradeTime <=09:30:00.000, 09:30:00.000, TradeTime) as TradeTime,
			SecurityID,
			OpenPrice,
			PreCloPrice,
			HighPrice,
			LowPrice,
			LastPrice,
			PreCloseIOPV,
			IOPV,
			UpLimitPx,
			DownLimitPx,
			iif(deltas(HighPrice)>0.000001, 1, 0) as DeltasHighPrice,
			iif(abs(deltas(LowPrice))>0.000001, -1, 0) as DeltasLowPrice,
			iif(deltas(TotalVolumeTrade)==NULL, TotalVolumeTrade, deltas(TotalVolumeTrade)) as DeltasVolume,
			iif(deltas(TotalValueTrade)==NULL, TotalValueTrade, deltas(TotalValueTrade)) as DeltasTurnover,
			iif(deltas(NumTrades)==NULL, NumTrades, deltas(NumTrades)) as DeltasTradesCount
		from loadTable(dbName, tbName)
		where TradeTime >=09:25:00.000
		context by SecurityID

Step3. K 线计算

针对处理后的快照行情进行窗口为1分钟、步长为1分钟的滚动窗口计算。

re = select	firstNot(LastPrice, 0) as OpenPrice,
			high(DeltasHighPrice, HighPrice, LastPrice) as HighPrice,
			low(DeltasLowPrice, LowPrice, LastPrice) as LowPrice,
			last(LastPrice) as ClosePrice,
			sum(DeltasVolume) as Volume,
			sum(DeltasTurnover) as Turnover,
			sum(DeltasTradesCount) as TradesCount,
			last(PreCloPrice) as PreClosePrice,
			last(PreCloseIOPV) as PreCloseIOPV,
			last(IOPV) as IOPV,
			last(UpLimitPx) as UpLimitPx,
			last(DownLimitPx) as DownLimitPx,
			lastNot(LastPrice, 0)\firstNot(LastPrice, 0)-1 as FirstBarChangeRate	
		from tempTB1
		group by SecurityID, TradeDate, interval(X=TradeTime, duration=60s, label='left', fill=0) as TradeTime

该部分使用了一个关键的 SQL 语句 group by 进行分组聚合计算，group by 只能够搭配聚合指标使用，上述代码中的 firstNot, high, low, last, sum, lastNot 等等都是聚合函数。

group by 指定的分组指标是 SecurityID, TradeDate, interval(X=TradeTime, duration=60s, label='left', fill=0)，即股票字段、日期、分钟时间。interval 函数必须与 group by 语句搭配使用，通常用于时间精度的降维场景，此处 interval 函数将毫秒精度的时间字段，按照分钟窗口进行划分，缺失值填充为 0。

部分计算结果示意如下：

SecurityID TradeDate  TradeTime    OpenPrice HighPrice LowPrice ClosePrice Volume Turnover              TradesCount ...
---------- ---------- ------------ --------- --------- -------- ---------- ------ --------------------- ----------- ---
000002     2022.01.04 09:30:00.000 1.6       1.779     1.003    1.3        100    10022.200000000000727 94          ...
000002     2022.01.04 09:31:00.000 1.5       1.756     1.003    1.4        0      0                     0           ...
000002     2022.01.04 09:32:00.000 1.6       1.777     1.011    1.5        0      0                     0           ...
000002     2022.01.04 09:33:00.000 1.5       1.791     1.01     1.5        0      0                     0           ...
000002     2022.01.04 09:34:00.000 1.4       1.791     1.042    1.5        0      1.5                   -39         ...
000002     2022.01.04 09:35:00.000 1.5       1.794     1.003    1.5        0      0                     0           ...
...

先将上述 K 线的计算逻辑封装在一个函数内。

def calcKLine(snapshot){
    tempTB1 = select TradeDate,
			iif(TradeTime <=09:30:00.000, 09:30:00.000, TradeTime) as TradeTime,
			SecurityID,
			OpenPrice,
			PreCloPrice,
			HighPrice,
			LowPrice,
			LastPrice,
			PreCloseIOPV,
			IOPV,
			UpLimitPx,
			DownLimitPx,
			iif(deltas(HighPrice)>0.000001, 1, 0) as DeltasHighPrice,
			iif(abs(deltas(LowPrice))>0.000001, -1, 0) as DeltasLowPrice,
			iif(deltas(TotalVolumeTrade)==NULL, TotalVolumeTrade, deltas(TotalVolumeTrade)) as DeltasVolume,
			iif(deltas(TotalValueTrade)==NULL, TotalValueTrade, deltas(TotalValueTrade)) as DeltasTurnover,
			iif(deltas(NumTrades)==NULL, NumTrades, deltas(NumTrades)) as DeltasTradesCount
		from snapshot
		where TradeTime >= 09:25:00.000
		context by SecurityID

    re = select	firstNot(LastPrice, 0) as OpenPrice,
                high(DeltasHighPrice, HighPrice, LastPrice) as HighPrice,
                low(DeltasLowPrice, LowPrice, LastPrice) as LowPrice,
                last(LastPrice) as ClosePrice,
                sum(DeltasVolume) as Volume,
                sum(DeltasTurnover) as Turnover,
                sum(DeltasTradesCount) as TradesCount,
                last(PreCloPrice) as PreClosePrice,
                last(PreCloseIOPV) as PreCloseIOPV,
                last(IOPV) as IOPV,
                last(UpLimitPx) as UpLimitPx,
                last(DownLimitPx) as DownLimitPx,
                lastNot(LastPrice, 0)\firstNot(LastPrice, 0)-1 as FirstBarChangeRate	
            from tempTB1
            group by SecurityID, TradeDate, interval(X=TradeTime, duration=60s, label='left', fill=0) as TradeTime

    return re
}

然后利用 mr 函数指定 parallel 参数为 false 进行分区串行计算。此处选取了一个月的数据进行演示，其中 sqlDS 函数可以根据指定的 SQL 查询筛选数据，并按分区生成数据源列表。

beginDate = 2022.01.01 
endDate = 2022.01.31 
dbName = "dfs://stock_lv2_snapshot"
tbName = "snapshot"

ds = sqlDS(<select * from loadTable(dbName, tbName) where TradeDate between beginDate and endDate>)
re = mr(ds, calcKLine, , unionAll, false)

select top 10 * from re order by SecurityId

/* output:
SecurityID TradeDate  TradeTime    OpenPrice HighPrice LowPrice ClosePrice Volume Turnover              TradesCount ...
---------- ---------- ------------ --------- --------- -------- ---------- ------ --------------------- ----------- ---
000001     2022.01.04 09:30:00.000 1.6       1.794     1.064    1.3        0      10000.200000000000727 53          ...
000001     2022.01.04 09:31:00.000 1.4       1.767     1.011    1.3        0      0                     0           ...
000001     2022.01.04 09:32:00.000 1.6       1.8       1.046    1.4        0      0                     0           ...
000001     2022.01.04 09:33:00.000 1.6       1.736     1.033    1.4        0      0                     0           ...
000001     2022.01.04 09:34:00.000 1.6       1.772     1.063    1.3        0      0                     0           ...
000001     2022.01.04 09:35:00.000 1.3       1.758     1.035    1.6        0      0.799999999999272     -41         ...
000001     2022.01.04 09:36:00.000 1.4       1.784     1.013    1.5        0      0                     0           ...
000001     2022.01.04 09:37:00.000 1.4       1.79      1.006    1.5        0      0                     0           ...
000001     2022.01.04 09:38:00.000 1.6       1.781     1.024    1.5        0      0                     0           ...
000001     2022.01.04 09:39:00.000 1.5       1.697     1.024    1.6        0      0                     0           ...
*/

在金融领域，时间序列分析是一种重要的工具，它可以帮助理解数据的趋势、季节性、随机性等特征，并进行准确的预测。例如，在股票价格预测、货币汇率预测、通胀率预测等方面，时间序列分析被广泛应用于揭示金融数据随时间变化的规律。为应对此类场景，DolphinDB 提供了滑动窗口、累积窗口、序列计算等内置函数，以及 Talib、MyTT 算子指标库和 Alpha 因子库等模块，用户可以利用这些函数的组合和嵌套构建复杂的时序指标算子。

例：基于快照数据计算订单失衡率（SOIR）因子

该因子用于衡量买卖委托量在总量中的不均衡程度。t 时刻第 i 档的订单失衡率因子定义如下，具体参数说明参考 3.1.2 节。

使用 DolphinDB 的脚本实现上述因子的定义：此处使用 rowWavg 函数计算各档加权平均的买卖委托量不均衡程度因子，即订单失衡率因子。并使用 moving 系列函数对一段时间的指标进行移动标准化处理。

def wavgSOIR(bidQty,askQty,lag=20){
	imbalance= rowWavg((bidQty - askQty)\(bidQty + askQty), 10 9 8 7 6 5 4 3 2 1).ffill().nullFill(0)
	mean = mavg(prev(imbalance), (lag-1), 2)
	std = mstdp(prev(imbalance) * 1000000, (lag-1), 2) \ 1000000
	return iif(std >= 0.0000001,(imbalance - mean) \ std, NULL).ffill().nullFill(0)
}

re = select SecurityID, TradeDate, TradeTime, wavgSOIR(bidOrderQty, OfferOrderQty, lag=20) as HeightImbalance from loadTable(dbName, tbName) where TradeDate between beginDate and endDate context by SecurityID csort TradeTime

select top 10 * from re
/* output:
SecurityID TradeDate  TradeTime    HeightImbalance   
---------- ---------- ------------ ------------------
000001     2022.01.04 09:30:00.000 0                 
000001     2022.01.05 09:30:00.000 0                 
000001     2022.01.06 09:30:00.000 -4.710845145045806
000001     2022.01.07 09:30:00.000 10.571487776713668
000001     2022.01.10 09:30:00.000 1.688837114535889 
000001     2022.01.04 09:30:03.000 -0.761301248682889
000001     2022.01.05 09:30:03.000 -0.494321354433684
000001     2022.01.06 09:30:03.000 -0.963245349720814
000001     2022.01.07 09:30:03.000 1.708213822850803 
000001     2022.01.10 09:30:03.000 1.387502605818854
*/

除了纵向的时序计算外，在金融场景下还需要横向对同一时间截面上不同股票的数据进行一些比较或计算。要对比同一截面不同股票的指标，就需要将原本的窄表进行一些行列转换，使之转成面板数据。为此，DolphinDB 在 SQL 语义上拓展提供了 pivot by 语句，并开发了用于内存计算的 pivot、panel 函数。

pivot 操作基于指定的维度（通常是时间和标的）将某一个属性值展开成面板数据，通常用于将一个窄表转换成宽表/矩阵，然后进行计算。

在 DolphinDB 中使用 exec+pivot by 语句可以将一个表转换成一个面板矩阵，矩阵的行列标签记录了面板数据的维度信息。以一个简单的例子进行说明。

原表 t 定义如下：

sym = `C`MS`MS`MS`IBM`IBM`C`C`C
price= 49.6 29.46 29.52 30.02 174.97 175.23 50.76 50.32 51.29
qty = 2200 1900 2100 3200 6800 5400 1300 2500 8800
timestamp = [09:34:07,09:35:42,09:36:51,09:36:59,09:35:47,09:36:26,09:34:16,09:35:26,09:36:12]
t = table(timestamp, sym, qty, price);
/* output:
timestamp sym qty  price              
--------- --- ---- -------------------
09:34:07  C   2200 49.600000000000001 
09:35:42  MS  1900 29.46              
09:36:51  MS  2100 29.519999999999999 
09:36:59  MS  3200 30.019999999999999 
09:35:47  IBM 6800 174.969999999999998
09:36:26  IBM 5400 175.229999999999989
09:34:16  C   1300 50.759999999999997 
09:35:26  C   2500 50.32              
09:36:12  C   8800 51.289999999999999
*/

利用 exec + pivot by 将表 t 转成面板矩阵：

m = exec price from t pivot by timestamp, sym;
m
/* output:
         C                  IBM                 MS                
         ------------------ ------------------- ------------------
09:34:07|49.600000000000001                                       
09:34:16|50.759999999999997                                       
09:35:26|50.32                                                    
09:35:42|                                       29.46             
09:35:47|                   174.969999999999998                   
09:36:12|51.289999999999999                                       
09:36:26|                   175.229999999999989                   
09:36:51|                                       29.519999999999999
09:36:59|                                       30.019999999999999
*/

typestr m
// output: 'FAST DOUBLE MATRIX'

m.colNames() // output: ["C","IBM","MS"]
m.rowNames() // output: [09:34:07,09:34:16,09:35:26,09:35:42,09:35:47,09:36:12,09:36:26,09:36:51,09:36:59]

了解了 pivot by 的基本使用，下面结合具体的金融数据进行举例。

例：根据每天每只股票最后的 LastPrice 价格计算排名。

此处 pivot by 用于指定截面的维度信息，此处是日期和股票代码，由于同一个日期对应多个数据，因此计算时可以配合聚合指标使用。

m = exec last(LastPrice) from loadTable(dbName, tbName) pivot by TradeDate, SecurityID
rowRank(m)

/* output:
           000001 000002 000003 000004 000005 000006 000007 000008 000009 000010 000011 000012 000013 000014 000015 ...
           ------ ------ ------ ------ ------ ------ ------ ------ ------ ------ ------ ------ ------ ------ ------ ------ ---
2022.01.04|87     0      26     26     0      26     58     26     87     0      26     26     0      0      58     ...
2022.01.05|84     20     20     47     84     47     84     47     0      20     47     20     0      47     84     ...
2022.01.06|84     17     84     84     84     47     17     17     47     47     84     47     17     0      47     ...
2022.01.07|50     0      50     12     83     50     0      12     50     12     12     50     0      12     50     ...
2022.01.10|45     15     15     15     15     74     74     0      45     15     74     15     74     45     74     ...
...
*/

1. 如何在写入分布式表时对数据进行去重？
   • TSDB 引擎 sortColumns 去重：系统会根据索引配置字段 sortColumns 进行去重，但是该方式不适用于每个索引键（除最后一列时间列以外的 sortColumns 字段构成索引键）对应的数据量很少的场景，因为索引键的主要作用是数据索引，若设置为去重键会造成索引膨胀，影响查询效率。
   • 使用 upsert! 方法写入数据：upsert! 支持设置数据键 keyColNames，使用 upsert! 写入新数据时，系统会去检查原有数据中该数据键是否已经存在，若存在则更新主键的数据，否则追加。通过该方式以达到数据去重的目的。
2. 如何写脚本可以优化查询性能，查询很慢的原因是什么?

   一个查询是否可以优化，主要是看以下的方面：

   • 是否能够触发分区剪枝，缩小系统遍历范围？ → where 条件是否可以改写为触发剪枝的条件
   • 是否走了分布式查询的优化？ → 如果是自定义函数，无法进行分布式查询优化，是否可以通过 mr 函数进行逻辑改写。

   查询慢的原因可以参考文档 查询/写入慢 进行排查。

3. 查询的字段调用了自定义的函数，会对性能有影响吗，会走分布式查询的优化吗？

   如果没有通过 mapr 关键字声明，则自定义函数无法进行分布式查询优化，此时有几种优化方案：

   1. 计算数据不跨分区：在查询语句后添加 map 关键字。例如：select first(id), count(*) from pt map。
   2. 计算数据跨分区：通过函数 mr 定义分布式计算逻辑。参数 mapFunc 指定函数会映射到各个分区进行执行，并将计算结果通过 reduceFunc 进行处理汇总后，通过 finalFunc 进行最后的计算。案例参考 通用计算 。
4. 查询字段多造成脚本复杂，可以如何优化代码？

   可以通过元编程或者字段序列的方式去定义脚本。

   例：查询字段 col000~col999 的数据

   // 元编程
   cols="col" + lpad(string(0..999), 3, "0")
   <select _$$cols from t>.eval()
   
   // 字段序列
   select col000...col999 from t

   字段序列对于字段名和查询场景都有限制，与之相比元编程的使用范围更广。

   例：计算多个字段的最小二乘回归。

   colName = "y"
   x = `SH000001`SH000003`SH000005
   <select ols(_$colName, _$$x, 1, 2).Residual as residual from t>.eval()

   具体用法请参考 基于 SQL 的元编程 。

5. 如何进行表关联后只保留公共匹配列?

   DolphinDB 中表关联的结果会将两边公共列拆成两列返回，例如：

   t1= table(1 2 3  as id, 7.8 4.6 5.1  as value)
   t2 = table(1 3 5 as id, 300 500 800 as qty)
   fj(t1, t2, `id)
   /* output:
   id value t2_id qty
   -- ----- ----- ---
   1  7.8   1     300
   2  4.6            
   3  5.1   3     500
            5     800
   */

   如果需要将公共列 id 合成一列返回，可以参考下述脚本：

   select t1.id.nullFill(t2.id) as id, value, qty from fj(t1, t2, `id)

6. 共享表怎么删除，调用 dropTable 函数不起作用？

   dropTable 用于删除分布式表，对于共享内存表可以用 undef("st", SHARED) 进行删除。

7. update 和 delete 的事务开销很大，这是为什么？

   DolphinDB 的分布式数据是按照列式存储（OLAP）或者行列混存（TSDB）的，对于行级别的更新和删除操作，系统以分区为单位读出相关字段的数据到内存，然后进行更新和删除再写回磁盘。因此，如果事务涉及的分区多、分区数据量大，更新和删除的开销也会很大。

   如果数据以 TSDB 引擎进行存储，则在某些特定场景和配置下，其更新和删除操作有优化：

   • 设置去重策略 keepDuplicates=LAST 时，数据更新以数据追加方式进行。
   • 设置去重策略 keepDuplicates=LAST 并开启软删除机制 softDelete=true，则删除操作时，系统取出待删除的数据打上删除标记（软删除），再以追加的方式写回数据库。

   上述两种优化后的更新和删除方式效率更高。

8. 1. 如何编写查询脚本，使之可以通过命中内部维护的元数据信息，从而加速查询性能？

      DolphinDB 内部维护了分区信息、TSDB 引擎还额外维护了 zonemap、sortKey 等信息。理论上可以通过此类元数据信息加速相关的查询。下面以具体的例子进行说明：

      例 1：查询分区字段的元数据信息。

      场景：按照日期（天）和股票代码的 COMPO 分区，获取所有日期（天）值。

      以 4.2.1 导入的快照数据表 snapshot 为例，该表的结构为：

      engineType->TSDB
      keepDuplicates->ALL
      chunkGranularity->TABLE
      sortColumns->["SecurityID","TradeTime"]
      softDelete->0
      tableOwner->admin
      compressMethods->name               compressMethods
      ------------------ ---------------
      TradeDate          delta          
      TradeTime          delta          
      MDStreamID         lz4            
      SecurityID         lz4                    
      ...
      
      tableComment->
      colDefs->name               typeString typeInt extra comment     
      ------------------ ---------- ------- ----- ------------
      TradeDate          DATE       6             交易日期
      TradeTime          TIME       8             交易时间
      MDStreamID         SYMBOL     17                                           
      ...
      
      chunkPath->
      partitionColumnIndex->[0,3]
      partitionColumnName->["TradeDate","SecurityID"]
      partitionColumnType->[6,17]
      partitionType->[1,5]
      partitionTypeName->["VALUE","HASH"]
      partitionSchema->([2022.01.04,2022.01.05,2022.01.06,2022.01.07,2022.01.10,2022.01.11,2022.01.12,2022.01.13,2022.01.14,2022.01.17,2022.01.18,2022.01.19,2022.01.20,2022.01.21,2022.01.24,2022.01.25,2022.01.26,2022.01.27,2022.01.28,2022.02.07,2022.02.08,2022.02.09,2022.02.10,2022.02.11,2022.02.14,2022.02.15,2022.02.16,2022.02.17,2022.02.18,2022.02.21...],50)
      partitionSites->

      分别利用三种方式求所有数据日期的唯一值：

      dbName = "dfs://stock_lv2_snapshot"
      tbName = "snapshot"
      pt = loadTable(dbName, tbName)
      
      // （1）直接利用 distinct 函数求日期唯一值
      timer dateList = exec distinct(TradeDate) from pt
      // Time elapsed: 670.988 ms
      
      // （2）利用 group by 的分布式计算求日期唯一值
      timer dateList =  exec TradeDate from select count(*) from pt group by TradeDate
      // Time elapsed: 373.783 ms
      
      // （3）利用元数据信息求日期的唯一值
      timer{
          result = exec dfsPath from getTabletsMeta( dbName[5:]+ "/%", tbName, false, -1) where rowNum != 0
          dateList = substr(result, regexFind(result, "[0-9]{8}"), 8).distinct().temporalParse("yyyyMMdd").sort()
      }
      // Time elapsed: 35.623 ms

   比较查询耗时可以发现方案 3 性能最佳，因为它是通过直接解析分区元数据获取日期信息的。

   如果创建数据库时，设置的 VALUE 分区方案包含在数据范围内，则利用 VALUE 分区自动拓展的特性，可以通过 schema 函数直接获取日期分区字段的值。

   本例中 snapshot 表的分区信息可以通过schema(pt).partitionSchema[0]获取，获取结果是长度为 244 的向量，较真实的场景多了 2 天，因为 2024.01.01 和 2024.01.02 这两天也被包含了。这是由于在创建数据库时， snapshot 的建库语句为：create database "dfs://stock_lv2_snapshot" partitioned by VALUE(2024.01.01..2024.01.02), HASH([SYMBOL, 50]), engine='TSDB'，其中指定的 2024.01.01..2024.01.02 也被纳入了元数据信息中，即使该分区并无数据。

   利用该特性，我们在创建数据库时指定必然有数据的天作为分区 scheme，例如 create database "dfs://stock_lv2_snapshot" partitioned by VALUE(2022.01.04..2022.01.04), HASH([SYMBOL, 50]), engine='TSDB'，这样就可以直接借助 schema 函数更方便地获取元数据信息。

   例 2：限 TSDB 引擎：查询 count, min, max, sum 等 zonemap 包含的元数据信息。

   zonemap 存储了数据的预聚合信息（每列每个 sortKey 对应数据的 min，max，sum，notnullcount）。如果查询语句包含 count, min, max, sum 等预聚合指标，且用户于设置去重策略（keepDuplicates=ALL），就可以利用 zonemap 进行提速。

   不同去重策略下，查询的性能对比见 TSDB 存储引擎详解 5.2 节查询全表总记录数（命中元数据）部分。对于 ALL 去重策略，由于查询时无需去重，直接利用 zonemap 提速性能最高。

   

   例 3：限 TSDB 引擎：查询条件命中 sortKey 索引数据。

   dbName = "dfs://stock_lv2_snapshot"
   tbName = "snapshot"
   pt = loadTable(dbName, tbName)
   
   select [HINT_EXPLAIN] * from pt where SecurityID in ["000076", "000024"]
   
   // output：
   ...
   "explain": {
                 "where": {
                     "TSDBIndexPrefiltering": {
                         "blocksToBeScanned": 60, // 扫描的 block 个数
                         "matchedWhereConditions": 1 // 命中 sortKey 的过滤条件 
                     },
   ...

   可以通过 [HINT_EXPLAIN] 执行计划观察查询的过滤条件是否命中索引。

• 分区剪枝：查询数据 SQL 执行计划
• SQL 操作：量化金融范例 深度不平衡、买卖压力指标、波动率计算
  • 聚合计算：K 线计算 基于快照行情的股票和基金 K 线合成
  • 时序计算：窗口计算 处理 Level-2 行情数据实例
  • 截面计算：面板数据处理

流数据是基于事件持续生成的时间序列数据。与静态有界的历史数据不同，流数据具有以下特点：

• 动态：数据流持续动态生成，流的结束没有明确定义，数据的大小与结构也没有固定限制。
• 有序：每条流数据记录都具有时间戳或者序列号，标识了数据在流中的位置与顺序。
• 大规模：流数据通常以高速率生成，数据规模大，对处理引擎的并行处理性能和可扩展性有更高要求。
• 强时效：流数据的强时效性要求极低延迟的读取和处理能力，以最大化数据价值，驱动实时业务决策。

流数据处理是指在实时数据流上进行实时计算和分析的过程，也称为流计算。与批处理不同，流处理无需等待所有数据全部到位，即可按照时间顺序对数据进行增量处理。这种实时处理方式能够高效利用存储与计算资源，适用于需要快速响应和及时决策的应用场景。

表 7-1 批计算和流计算对比

在 DolphinDB 处理流数据有天然优势，这是因为 DolphinDB 内置了处理不同逻辑场景的流计算引擎，即使是复杂的因子计算逻辑，也可以通过逻辑拆分交给不同的引擎去实现，通过引擎的级联实现流水线式的计算。且 DolphinDB 的内置引擎支持的算子经过了算法上的优化，计算性能高效。用户使用时仅需要对引擎参数进行配置，即可实现复杂逻辑，节省了用户自己开发算法逻辑的成本。

根据各自适用的逻辑和场景，DolphinDB 内置的流计算引擎可以分为以下几类：

表 7-3-1 不同流计算场景下适用的引擎及其计算逻辑

例. 流式计算主买/主卖交易量（示例代码参考：流式计算中证 1000 指数主买/主卖交易量）

计算公式：主买/主卖交易量 = Σ(交易价格*交易量*权重因子)

拆分计算逻辑，得到的计算流程为：

（1）计算每分钟的成交量（按照股票分组）→ 时序聚合引擎

（2）按分钟累计成交量（按照股票分组） → 响应式状态引擎

（3）对股票成交量进行加权求和计算交易量 → 横截面引擎

在运行脚本前，可以先清理环境中的变量，并初始化定义一些公共变量：

undef all
go

dbName = "dfs://stock_trade"
tbName = "trade"
streamTbName = "trade_st"

try{unsubscribeTable(tableName=streamTbName, actionName="act_tsEngine")}catch(ex){print(ex)}
try{dropStreamEngine("rsEngine")}catch(ex){print(ex)}
try{dropStreamEngine("tsEngine")}catch(ex){print(ex)}
try{dropStreamEngine("csEngine")}catch(ex){print(ex)}

定义一个持久化流表 trade_st 用于接收实时的逐笔成交数据。

colName=loadTable(dbName, tbName).schema().colDefs.name
colType=loadTable(dbName, tbName).schema().colDefs.typeString
t = streamTable(100:0, colName, colType);

try{ 
    enableTableShareAndPersistence(table=t, tableName=streamTbName, asynWrite=true, compress=true, cacheSize=100000, retentionMinutes=1440, flushMode=0, preCache=10000)
}
catch(ex){ 
    print(ex) 
}

undef("t")
go

模拟创建成分股权重因子。

sids = exec distinct SecurityID from loadTable(dbName, tbName)
def createWeightDict(sids){

	return dict(sids, take(0.001, sids.size()))
}

weightDict = createWeightDict(sids)

模拟创建成分股权重因子。

// 输入表结构定义
tsEngineDummy = table(1:0, `SecurityID`TradeTime`TradePrice`TradeQty`TradeBSFlag, [SYMBOL, TIMESTAMP, DOUBLE, INT, SYMBOL])
rsEngineDummy = table(1:0, `TradeTime`SecurityID`SellTradeAmount`BuyTradeAmount, [TIMESTAMP, SYMBOL, DOUBLE, DOUBLE])
csEngineDummy = table(1:0, `SecurityID`TradeTime`SellTradeAmount`BuyTradeAmount, [SYMBOL, TIMESTAMP, DOUBLE, DOUBLE])

//创建存储结果的表
share(keyedTable(`Time, 2000:0, `Time`SellTradeAmount`BuyTradeAmount`UpdateTime, [TIMESTAMP, DOUBLE, DOUBLE, TIMESTAMP]), "tradeAmountIndex")
go

定义流计算引擎。由于引擎是通过输出表设置为下一个引擎，来进行级联的，为保证脚本正常解析，引擎定义的顺序和业务数据流的逻辑是相反的。如本例中计算逻辑是 时序引擎 → 响应式状态引擎 → 横截面引擎，脚本中定义顺序则相反：

//创建横截面引擎
csEngine = createCrossSectionalEngine(name="csEngine", metrics=<[wsum(SellTradeAmount, weightDict[SecurityID]), wsum(BuyTradeAmount, weightDict[SecurityID]), now()]>, dummyTable=csEngineDummy, outputTable=objByName("tradeAmountIndex"), keyColumn=`SecurityID, triggeringPattern="keyCount",  triggeringInterval=1000, useSystemTime=false, timeColumn=`TradeTime, lastBatchOnly=false)

//创建响应式状态引擎
rsEngine = createReactiveStateEngine(name="rsEngine", metrics=<[cummax(TradeTime), cumsum(SellTradeAmount), cumsum(BuyTradeAmount)]>, dummyTable=rsEngineDummy, outputTable=csEngine, keyColumn=`SecurityID)

//创建时间序列引擎
tsEngine = createTimeSeriesEngine(name="tsEngine", windowSize=60000, step=60000, metrics=<[sum(iif(TradeBSFlag=="S", 1, 0)*TradeQty*TradePrice), sum(iif(TradeBSFlag=="B", 1, 0)*TradeQty*TradePrice)]>, dummyTable=tsEngineDummy, outputTable=rsEngine, timeColumn=`TradeTime, keyColumn=`SecurityID, useWindowStartTime=true,  fill=[0, 0], forceTriggerTime=100)

订阅逐笔数据表，写入引擎流水线的第一个引擎：时序聚合引擎。

本例自定义了一个流数据处理函数 streamHandler，可以通过修改内部逻辑自定义数据处理方案。

// handler 数据处理函数
def streamHandler(mutable engine, msg){
    data = select SecurityID, concatDateTime(TradeDate, TradeTime) as TradeTime, TradePrice, TradeQty, TradeBSFlag from msg
    engine.append!(data)
}

// 订阅流数据表
subscribeTable(tableName=streamTbName, actionName="act_tsEngine", offset=0, handler=streamHandler{tsEngine}, msgAsTable=true, batchSize=10000, throttle=0.001)

通过 replayDS 函数获取历史数据源，然后借助 replay 函数模拟流数据回放过程。

// 后台提交数据回放任务，模拟实时数据注入
// 本例设置 timeRepartitionSchema 将数据源按照 15min 进行切片，这是为了保证回放时数据不会 OOM
ds = replayDS(<select * from loadTable(dbName, tbName)>, dateColumn="TradeDate", timeColumn="TradeTime", timeRepartitionSchema=09:30:00.000 + 15*1000*60*(0..22))

submitJob("relay", "replay to trade", replay{inputTables=ds, outputTables=streamTbName, dateColumn="TradeDate", timeColumn="TradeTime", replayRate=10000000, absoluteRate=true})
getRecentJobs(1)

回放进程可以通过函数 getRecentJobs 进程查询。

通过 SQL 查询输出的结果表：

select * from tradeAmountIndex limit 10

为了新用户能够更好理解引擎的计算，我们仔细观察三个计算引擎的流计算指标。

• 时序聚合引擎

  <[sum(iif(TradeBSFlag=="S", 1, 0)*TradeQty*TradePrice), 
  sum(iif(TradeBSFlag=="B", 1, 0)*TradeQty*TradePrice)]>

• 响应式状态引擎

<[cummax(TradeTime), cumsum(SellTradeAmount), cumsum(BuyTradeAmount)]>

• 横截面引擎

<[wsum(SellTradeAmount, weightDict[SecurityID]), wsum(BuyTradeAmount, weightDict[SecurityID]), now()]>

将上述流计算脚本转换为 SQL 批计算逻辑，可改写如下：

从分布式表中筛选一部分字段进行计算：

data = select SecurityID, concatDateTime(TradeDate, TradeTime) as TradeTime, TradePrice, TradeQty, TradeBSFlag from loadTable(dbName, tbName)

• 按时间进行分钟聚合（时序聚合）

t1 = select sum(iif(TradeBSFlag=="S", 1, 0)*TradeQty*TradePrice) as SellTradeAmount,
            sum(iif(TradeBSFlag=="B", 1, 0)*TradeQty*TradePrice) as BuyTradeAmount
    from data group by SecurityID, interval(TradeTime, 1m, 0) as TradeTime

• 累积计算成交量（状态计算）

t2 = select SecurityID, cummax(TradeTime) as Time, 
            cumsum(SellTradeAmount) as SellTradeAmount, 
            cumsum(BuyTradeAmount) as BuyTradeAmount 
     from t1 context by SecurityID

• 按股票因子加权计算成交量（截面计算）

t3 = select wsum(SellTradeAmount, weightDict[SecurityID]) as SellTradeAmount,
            wsum(BuyTradeAmount, weightDict[SecurityID]) as BuyTradeAmount 
     from t2 group by Time

对比批计算和流计算的脚本，可以发现不论是引擎的算子还是 SQL select 的计算主体，本质上都使用了一套函数去实现，这体现了 DolphinDB 流批一体的计算特点。

1. 流表数据是否支持部分记录的更新和删除？

   不支持。流表记录只能追加，一旦追加后就不能更新和删除。

   特殊地，对于键值流表（keyedStreamTable 和 latestKeyedStreamTable），系统支持在数据写入流表前，进行键值检查，以选择数据是追加还是丢弃。

2. 流表数据量大，内存占用不够怎么解决?

   可以将流表通过 enableTablePersistence 函数或者 enableTableShareAndPersistence 函数进行持久化，系统会将一部分数据持久化存储到磁盘指定目录（通过配置参数 persistenceDir 设置）。

3. 如何配置重启后自动加载订阅流表?

   将流数据表定义和流表订阅操作放在启动脚本 startup.dos （通过配置参数 startup 设置）内，这样节点重启后可以自动加载订阅流表。详情参考教程 启动脚本 和 节点启动时的流计算自动订阅。

4. 流表如何进行删除？一直无法删除如何解决？

   流表需要通过函数 dropStreamTable 进行删除，而不是通过 undef(`st, SHARED) 进行删除。

   如果流表一直无法删除，请通过 getStreamingStat 函数查看当前流表上是否存在未取消的订阅，只有订阅全被取消后（取消订阅函数 unsubscribeTable），才能删除流表。

5. 如何配置定期删除持久化流表持久化的数据？

   用户可以在通过 enableTablePersistence 函数或者 enableTableShareAndPersistence 函数持久化流表时，设置 retentionMinutes 参数，定期删除持久化的数据。

6. 如何查询当前流表是否存在？

   可以通过函数 existsStreamTable 查看指定流数据表是否存在。

   需要注意，用户常会误用 defined 函数进行判断，但是该函数对于共享内存表也会返回 true。

7. 流订阅或流计算异常如何排查？

   函数 getStreamingStat 返回了流订阅发布的详细信息，其中比较常用的信息有：

   • 通过 subWorkers 表的 lastErrMsg 字段可以排查订阅中是否存在异常；
   • 通过 pubConns 表和 subWorkers 表中的 queueDepth 字段可以排查订阅是否阻塞；

通过函数 getStreamEngineStat 排查流计算引擎的状态。

官方文档 流数据 整理了一些流计算相关的文档和教程说明，此处仅列举部分：

• 流数据架构：流数据功能简介 流数据表 发布与订阅
• 流计算引擎：内置流式计算引擎 内置多数据源流式关联引擎
• 流数据接入：Python API 接入数据 行情数据插件：NSQ
• 数据回放：股票行情回放
• 流批一体：流批一体 快速搭建 Level-2 快照数据流批一体因子计算平台最佳实践
• 场景案例：金融实时实际波动率预测 实时计算日累计逐单资金流 实时计算分钟资金流 实时选取外汇行情多价源最优价 金融因子流式实现
• 可视化展示：数据面板 Altair 连接 DolphinDB 数据源 DolphinDB Grafana DataSource Plugin