subscribeTable

例1

下面是关于流计算的例子。在本例中，集群有两个节点：DFS_NODE1 和 DFS_NODE2。我们需要在 cluster.cfg 中指定 maxPubConnections和subPort 参数来启动发布/订阅功能。例如：

maxPubConnections=32
DFS_NODE1.subPort=9010
DFS_NODE1.persistenceDir=C:/DolphinDB/Data
DFS_NODE2.subPort=9011

在 DFS_NODE1 上执行以下脚本：

1. 创建一个共享的流数据表 trades_stream，并以同步模式保存。此时，表 trades_stream 有0行记录。

   n=20000000
   colNames = `time`sym`qty`price
   colTypes = [TIME,SYMBOL,INT,DOUBLE]
   enableTableShareAndPersistence(table=streamTable(n:0, colNames, colTypes), tableName="trades_stream", asynWrite=false, cacheSize=n)
   go

2. 创建一个分布式表 trades。此时，表 trades 有0行记录。

   if(existsDatabase("dfs://STREAM_TEST")){
        dropDatabase("dfs://STREAM_TEST")
   }
   dbDate = database(directory="", partitionType=VALUE, partitionScheme=temporalAdd(date(today()),0..30,'d'))
   dbSym= database(directory="", partitionType=RANGE, partitionScheme=string('A'..'Z') join "ZZZZ")
   db = database(directory="dfs://STREAM_TEST", partitionType=COMPO, partitionScheme=[dbDate, dbSym])
   colNames = `date`time`sym`qty`price
   colTypes = [DATE,TIME,SYMBOL,INT,DOUBLE]
   trades = db.createPartitionedTable(table=table(1:0, colNames, colTypes), tableName="trades", partitionColumns=`date`sym)

3. 创建表 trades_stream 的本地订阅。使用 saveTradesToDFS 函数把表 trades_stream 的流数据和今天的日期保存至表 trades。

   def saveTradesToDFS(mutable dfsTrades, msg): dfsTrades.append!(select today() as date,* from msg)
   subscribeTable(tableName="trades_stream", actionName="trades", offset=0, handler=saveTradesToDFS{trades}, msgAsTable=true, batchSize=100000, throttle=60)

4. 创建表 trades_stream 的另一个本地订阅。使用每分钟的流数据计算成交量加权平均价格（vwap），并以异步模式把结果保存至共享的流数据表 vwap_stream 中。

   n=1000000
   tmpTrades = table(n:0, colNames, colTypes)
   lastMinute = [00:00:00.000]
   colNames = `time`sym`vwap
   colTypes = [MINUTE,SYMBOL,DOUBLE]
   enableTableShareAndPersistence(table=streamTable(n:0, colNames, colTypes), tableName="vwap_stream")
   go
   
   def calcVwap(mutable vwap, mutable tmpTrades, mutable lastMinute, msg){
       tmpTrades.append!(msg)
       curMinute = time(msg.time.last().minute()*60000l)
       t = select wavg(price, qty) as vwap from tmpTrades where time < curMinute, time >= lastMinute[0] group by time.minute(), sym
       if(t.size() == 0) return
       vwap.append!(t)
       t = select * from tmpTrades where time >= curMinute
       tmpTrades.clear!()
       lastMinute[0] = curMinute
       if(t.size() > 0) tmpTrades.append!(t)
   }
   subscribeTable(tableName="trades_stream", actionName="vwap", offset=0, handler=calcVwap{vwap_stream, tmpTrades, lastMinute}, msgAsTable=true, batchSize=100000, throttle=60)

在 DFS_NODE2 上执行以下脚本，创建表 trades_stream 的远程订阅，并以异步模式把流数据保存至表 trades_stream_slave 中。

n=20000000
colNames = `time`sym`qty`price
colTypes = [TIME,SYMBOL,INT,DOUBLE]
enableTableShareAndPersistence(table=streamTable(n:0, colNames, colTypes), tableName="trades_stream_slave", cacheSize=n)
go
subscribeTable(server="DFS_NODE1", tableName="trades_stream", actionName="slave", offset=0, handler=trades_stream_slave)

在 DFS_NODE1 上执行以下脚本，模拟3支股票在10分钟内的流数据。每支股票每分钟生成2,000,000条记录。每分钟的数据被插入到流数据表 trades_stream 的600个数据块中。每两个数据块有100毫秒的时间间隔。

n=10
ticks = 2000000
rows = ticks*3
startMinute = 09:30:00.000
blocks=600
for(x in 0:n){
    time = startMinute + x*60000 + rand(60000, rows)
    indices = isort(time)
    time = time[indices]
    sym = array(SYMBOL,0,rows).append!(take(`IBM,ticks)).append!(take(`MSFT,ticks)).append!(take(`GOOG,ticks))[indices]
    price = array(DOUBLE,0,rows).append!(norm(153,1,ticks)).append!(norm(91,1,ticks)).append!(norm(1106,20,ticks))[indices]
    indices = NULL
    blockSize = rows / blocks
    for(y in 0:blocks){
        range =pair(y * blockSize, (y+1)* blockSize)
        insert into trades_stream values(subarray(time,range), subarray(sym,range), 10+ rand(100, blockSize), subarray(price,range))
        sleep(100)
    }

    blockSize = rows % blocks
    if(blockSize > 0){
        range =pair(rows - blockSize, rows)
        insert into trades_stream values(subarray(time,range), subarray(sym,range), 10+ rand(100, blockSize), subarray(price,range))
    }
}

在 DFS_NODE1 上执行以下脚本来检查结果：

trades=loadTable("dfs://STREAM_TEST", `trades)
select count(*) from trades

预期结果有 60,000,000 条记录。

select * from vwap_stream

表 vwap_stream 预期有 27 条记录。

在 DFS_NODE2 上执行以下脚本：

select count(*) from trades_stream_slave

我们看到的结果小于 60,000,000 行，因为部分表的记录已经保存到磁盘中。