registerSnapshotEngine

语法

registerSnapshotEngine(dbName, tableName, keyColumnName, [dedupColumnName])

参数

dbName 是一个字符串,表示分布式数据库的名称。

tableName 是一个字符串,表示分布式表的名称。

keyColumnName 是一个字符串,表示分组列的名称。

dedupColumnName 是一个字符串,引擎会在分组内根据该列对数据进行倒序排序,并保留排序后的第一条数据作为唯一记录。如果在集群中省略此参数,保留结果可能是随机的。

详情

为分布式表注册一个快照引擎。快照引擎将分布式表按照 keyColumnName 指定的列分组,获取每个组的最新记录。注册快照引擎后,我们可以通过 select [HINT_SNAPSHOT] * from loadTable(dbName,tableName) 来获取每个分组的最新记录。

使用 unregisterSnapshotEngine 函数可以取消注册快照引擎。

注:
  • 该功能目前不支持在包含计算节点的集群模式下运行。
  • 一个分布式表只能注册一个快照引擎。
  • 系统重启后,需要重新注册快照引擎。集群模式下,任一数据节点重启,整个集群都需要重新注册快照引擎,即先调用 unregisterSnapshotEngine 函数清空其他节点的 Snapshot 信息,再调用 registerSnapshotEngine 函数重新注册。

例子

创建分布式表:

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,`date`sym`val,[DATE,SYMBOL,DOUBLE])
pt=db.createPartitionedTable(t,`pt,`date`sym);

注册快照引擎来获取分布式表 pt 中每个股票代码对应的最新记录。

registerSnapshotEngine("dfs://compoDB","pt","sym");

写入数据,获取每个股票代码对应的最新记录。本例使用的是随机生成的数据,因此执行的结果有可能会与本例的结果不同。

def writeData(batch){
    pt=loadTable("dfs://compoDB","pt")
    tmp=table(batch:0,`date`sym`val,[DATE,SYMBOL,DOUBLE])
    dates=take(2018.09.01..2018.09.30,batch)
    syms=take(`AAPL`MSFT`MS`C,batch)
    vals=rand(100.0,batch)
    insert into tmp values(dates,syms,vals)
    pt.append!(tmp)
}

writeData(1000);

select [HINT_SNAPSHOT] * from loadTable("dfs://compoDB","pt");
date sym val
2018.09.29 AAPL 24.945753
2018.09.29 MS 14.034453
2018.09.30 C 3.89175
2018.09.30 MSFT 17.720025
writeData(1000);
select [HINT_SNAPSHOT] * from loadTable("dfs://compoDB","pt");
date sym val
2018.09.29 AAPL 86.296883
2018.09.29 MS 48.17885
2018.09.30 C 83.7821
2018.09.30 MSFT 44.415456

上例如果在集群中执行,在不同节点查询的结果可能会不同。为避免这种情况,可以设置参数 dedupColumnName "date",实现组内按照 date 列排序后保留最近一条。

unregisterSnapshotEngine(dbName="dfs://compoDB", tableName="pt")
registerSnapshotEngine(dbName="dfs://compoDB", tableName="pt",keyColumnName="sym", dedupColumnName="date");
writeData(1000);

// 以下查询在不同节点执行将获得相同结果
select [HINT_SNAPSHOT] * from loadTable("dfs://compoDB","pt");