快照引擎
1. 概述
在股票市场,交易员需要随时掌握每支股票的最新行情。在物联网领域,运维人员需要监控所有设备的最新状态数据。为了满足用户快速查询最新实时数据的需求,DolphinDB为分布式表(DFS tables)提供了快照引擎功能。快照引擎对每个分组最新插入的记录保存快照,查询最新记录时只需返回快照即可,因此性能出色,而且性能不受表中记录总数的影响。
注意:快照引擎目前仅支持单节点服务模式
2. 快照引擎的适用场景
若需快速查询各组最新记录,且数据是按时间顺序插入,系统内存资源充足,建议使用快照引擎查询最新记录。
若不能保证按时间顺序插入数据,则不适合使用快照引擎,否则得到的不能保证是最新数据。
因为快照引擎保存了各组的快照,当分组很多时,会占用较多内存。若系统内存资源紧张,也不建议使用快照引擎。
在数据节点启动后,需要注册快照引擎后,节点才会保存最新记录的快照。因此,若插入记录的时间间隔比较长,在重启后到新的数据写入之前,这段时间数据节点是没有保存快照的,所以这时若需要查询快照注册之前的最新记录,是不能用快照引擎查询到最新记录的。
3. 快照引擎的使用说明
3.1. 注册快照引擎
在查询之前,我们需要先为分布式表注册快照引擎,注册快照引擎的语法如下:
registerSnapshotEngine(dbName, tableName, keyColumnName)
其中参数dbName表示分布式数据库的名称,tableName表示分布式表的名称,keyColumnName表示分组列的名称。快照引擎将分布式表按照keyColumnName指定的列分组,获取每个组的最新记录。在实际使用时,这个分组列可以是设备的编号或IP地址、股票的股票代码等,这样用户就可以查到每个设备、每个股票的最新状态或行情数据。
下列代码创建分布式数据库compoDB和分布式表pt,然后注册快照引擎来获取分布式表pt中每个股票代码对应的最新记录。
db1=database("",VALUE,2018.09.01..2018.09.30)
db2=database("",VALUE,`AAPL`MSFT`MS`C)
db=database("dfs://compoDB",COMPO,[db1,db2])
t=table(1:0,`ts`sym`val,[DATETIME,SYMBOL,DOUBLE])
pt=db.createPartitionedTable(t,`pt,`ts`sym);
registerSnapshotEngine("dfs://compoDB","pt","sym")
需要注意的是,需要在集群中的每个节点都注册快照引擎(可以调用pnodeRun函数,例如pnodeRun(registerSnapshotEngine{dbName, tableName, keyColumnName})),而且数据节点重启后需要重新注册。
3.2. 查询快照
注册快照引擎后,要查询各组最新记录,只需在SQL语句中设置hint值为64。上述例子中要查询每个分组的最新记录,可以使用以下代码:
select [64] * from loadTable("dfs://compoDB","pt")
查询最近1分钟有行情数据更新的股票,可以使用以下代码:
select [64] sym from loadTable("dfs://compoDB","pt") where ts > datetimeAdd(now(),-1,`m)
3.3. 取消快照引擎
快照引擎为每个分组的最新记录在内存中保存快照。当分组较多时,会占用较多内存,所以不再需要查询最新记录时,建议取消快照引擎以释放内存。可以通过unregisterSnapshotEngine取消快照引擎:
unregisterSnapshotEngine(dbName, tableName)
其中参数dbName与tableName分别表示分布式数据库和分布式表的名称,须与注册时保持一致。上述例子中注册的快照引擎,我们可以用下列代码取消:
unregisterSnapshotEngine("dfs://compoDB","pt")
4. 快照引擎的性能优势
在DolphinDB中,除了使用快照引擎,还有三种通过select语句获取最新记录的方法。当快照引擎不适用时,可以考虑使用这些方法替代:
(1)通过group by对表内记录分组,然后用last函数查找每个分组的最后一行记录。但它也与快照引擎一样,不适合插入的数据是乱序的情况。
(2)通过group by对表内记录分组,然后用atImax函数找出每个分组的时间戳这列的最大值(即最新时间)所在行,然后返回该行的其他列数据。这种方法可用于插入数据是乱序的情况。
(3)通过SQL扩展语句context by对表内记录进行分组,然后在组内按时间戳列进行csort降序排序,再和top一并使用,就可获取每个分组中的最新记录。这种方法也可用于插入数据是乱序的情况。
本节将快照引擎与上述方法进行性能比较测试。
首先,准备测试环境。部署三台物理服务器的6数据节点的DolphinDB集群后,在DolphinDB GUI中用下列代码创建分布式数据库testDB和分布式表trainInfoTable,表中除了ts时间字段、trainID分组字段,还有3个指标字段。在 DolphinDB database中的分区方案是将时间ts作为第一个维度,每天一个分区,再将trainID作为分区的第二个维度,产生30个范围分区。
def createDatabase(dbName){
tableSchema = table(100:0,`trainID`ts`tag0001`tag0002`tag0003 ,[INT,TIMESTAMP, FLOAT,DOUBLE,LONG])
if(exists(dbName))
dropDatabase(dbName)
db1 = database("",VALUE,today()..(today()+3))
db2 = database("",RANGE,0..30*10+1)
db = database(dbName,COMPO,[db1,db2])
dfsTable = db.createPartitionedTable(tableSchema,"trainInfoTable",`ts`trainID)
}
login("admin","123456")
createDatabase("dfs://testDB")
用以下代码将300辆车的数据写入到分布式表中,每200毫秒写入一次。
//模拟数据产生
def simulateData(trainVector){
num = size(trainVector)
return table(take(trainVector,num) as trainID, take(now(),num) as ts, rand(20..41,num) as tag0001, rand(30..71,num) as tag0002, rand(70..151,num) as tag0003)
}
//将数据写入数据表。写入次数为batches。
def simulate(host,port,trainVector,batches){
h = xdb(host,port,"admin","123456")
for(i in 0:batches){
t=simulateData(trainVector)
h("append!{loadTable('dfs://testDB', 'trainInfoTable')}", t)
sleep(200)
}
}
然后,使用各种方法进行查询。
(1) 注册快照引擎,然后查询:
registerSnapshotEngine("dfs://testDB", "trainInfoTable", `trainID)
timer select [64] * from loadTable("dfs://testDB", "trainInfoTable")
(2) 通过group by对表内记录分组,然后用last函数查找每个分组的最后一行记录。为了减少查询范围,增加了一个查询条件,即限制ts为最近一小时。
timer select last(ts), last(tag0001), last(tag0002), last(tag0003)
from loadTable('dfs://testDB', 'trainInfoTable') where ts > datetimeAdd(now(),-1,`h) group by trainID
(3) 通过group by对表内记录分组,然后用atImax函数找出每个分组的时间戳这列的最大值(即最新时间)所在行,然后返回该行的其他列数据。atImax函数的语法是atImax(location, value),其中location和value是相同长度的向量或相同维度的矩阵。找出location中最大值所在的位置,就可以返回value中该位置对应的值。在查询表的数据时,location取时间戳列,value取其他状态列,就能查到表的最新记录。为了减少查询范围,增加了一个查询条件,即限制ts为最近一小时。
timer select atImax(ts,tag0001 ) as tag0001, atImax(ts,tag0002 ) as tag0002, atImax(ts,tag0003 ) as tag0003
from loadTable('dfs://testDB', 'trainInfoTable') where ts >datetimeAdd(now(),-1,`h) group by trainID
(4) 通过SQL扩展语句context by对表内记录进行分组,然后在组内按时间戳列进行csort降序排序,再和top一起使用,就可以用于获取每个分组中的最新记录。为了减少查询范围,增加了一个查询条件,即限制ts为最近一小时。
timer select top 1 * from loadTable('dfs://testDB', 'trainInfoTable') where ts >datetimeAdd(now(),-1,`h) context by trainID csort ts desc
进行2次测试。第一次是刚开始向数据库中写入数据时,测试时当天的分区中的数据记录大约有10万条。第二次是长时间写入后,测试时当天的分区中的数据记录已有约4千万条。
方法 | 第一次测试(刚开始写入时,单位:毫秒) | 第二次测试(长时间写入后,单位:毫秒) |
---|---|---|
Snapshot Engine | 53 | 55 |
last | 70 | 723 |
atImax | 71 | 743 |
context by | 72 | 781 |
从测试结果可以发现,用快照引擎查询最新记录的性能是最优的。而且由于它是返回内存中的快照,不受分区中记录数的影响,每次查询的时间都变化不大。而另外三种方法在查询时,会先把分区中的数据加载到内存中,所以当分区中的数据比较多时,会影响查询性能。