案例与脚本实践：DolphinDB 轻量级实时数仓的构建与应用

def clearEnv(){
    //取消订阅
    unsubscribeTable(tableName=`inputStream, actionName="dispatch1")
    unsubscribeTable(tableName=`inputStream, actionName="dispatch2")
    unsubscribeTable(tableName=`oneMinuteResult, actionName="calcHalfHour")
    unsubscribeTable(tableName=`oneMinuteResult, actionName="calcOneHour")
    unsubscribeTable(tableName=`fiveMinuteResult, actionName="calcOneDay")
    unsubscribeTable(tableName = `oneDayResultSimulate,actionName=`calcOneYear)

    unsubscribeTable(tableName = `oneMinuteResult,actionName=`appendInToDFS)
    unsubscribeTable(tableName = `fiveMinuteResult,actionName=`appendInToDFS)
    unsubscribeTable(tableName = `halfHourResult,actionName=`appendInToDFS)
    unsubscribeTable(tableName = `oneHourResult,actionName=`appendInToDFS)
    unsubscribeTable(tableName = `oneDayResult,actionName=`appendInToDFS)
    unsubscribeTable(tableName = `oneYearResult,actionName=`appendInToDFS)

    //删除流计算引擎
    for(i in 1..2){
        try{dropStreamEngine(`dispatchDemo+string(i))}catch(ex){print(ex)}
    }
    for(i in 1..5){
        try{dropStreamEngine(`oneMinuteCalc+string(i))}catch(ex){print(ex)}
        try{dropStreamEngine(`fiveMinuteCalc+string(i))}catch(ex){print(ex)}
    }
    try{dropStreamEngine(`halfHourCalc)}catch(ex){print(ex)}
    try{dropStreamEngine(`oneHourCalc)}catch(ex){print(ex)}
    try{dropStreamEngine(`oneDayCalc)}catch(ex){print(ex)}
    try{dropStreamEngine(`oneYearCalc)}catch(ex){print(ex)}

    //删除流数据表
    try{dropStreamTable(`inputStream)}catch(ex){print(ex)}
    try{dropStreamTable(`oneMinuteResult)}catch(ex){print(ex)}
    try{dropStreamTable(`fiveMinuteResult)}catch(ex){print(ex)}
    try{dropStreamTable(`halfHourResult)}catch(ex){print(ex)}
    try{dropStreamTable(`oneHourResult)}catch(ex){print(ex)}
    try{dropStreamTable(`oneDayResult)}catch(ex){print(ex)}
    try{dropStreamTable(`oneDayResultSimulate)}catch(ex){print(ex)}
    try{dropStreamTable(`oneYearResult)}catch(ex){print(ex)}

}

def createStreamTable(){
    //定义输入流表
    enableTableShareAndPersistence(table = streamTable(1000:0,`Time`deviceId`value,`TIMESTAMP`SYMBOL`DOUBLE),
            tableName = `inputStream,cacheSize = 1000000,precache=1000000)

    colName = `Time`deviceId`filterTime`MAX`MIN`MEAN`MED`P95`P5`CHANGE`CHANGE_RATE`first`last`endTime
    colType = `TIMESTAMP`SYMBOL`NANOTIMESTAMP join take(`DOUBLE,10) join `NANOTIMESTAMP
    //定义1分钟窗口计算结果流表            
    enableTableShareAndPersistence(table = streamTable(1000:0,colName,colType),
        tableName = `oneMinuteResult,cacheSize = 1000000,precache=1000000)

    //定义5分钟窗口计算结果流表            
    enableTableShareAndPersistence(table = streamTable(1000:0,colName,colType),
        tableName = `fiveMinuteResult,cacheSize = 1000000,precache=1000000)

    //定义30分钟窗口计算结果流表            
    enableTableShareAndPersistence(table = streamTable(1000:0,colName,colType),
        tableName = `halfHourResult,cacheSize = 1000000,precache=1000000)

    //定义1小时窗口计算结果流表            
    enableTableShareAndPersistence(table = streamTable(1000:0,colName,colType),
        tableName = `oneHourResult,cacheSize = 1000000,precache=1000000)
    
    //定义24小时窗口计算结果流表            
    enableTableShareAndPersistence(table = streamTable(1000:0,colName,colType),
        tableName = `oneDayResult,cacheSize = 1000000,precache=1000000)

    //定义模拟24小时窗口计算结果流表
    colName = `TIME`deviceId`MAX`MIN`MEAN`MED`P95`P5`CHANGE`CHANGE_RATE`first`last
    colType = `DATE`SYMBOL join take(`DOUBLE,10)           
    enableTableShareAndPersistence(table = streamTable(1000:0,colName,colType),
        tableName = `oneDayResultSimulate,cacheSize = 1000000,precache=1000000)

    //定义1年窗口计算结果流表
    colName = `Time`deviceId`filterTime`MAX`MIN`MEAN`MED`P95`P5`CHANGE`CHANGE_RATE`first`last`endTime
    colType = `DATE`SYMBOL`NANOTIMESTAMP join take(`DOUBLE,10) join `NANOTIMESTAMP            
    enableTableShareAndPersistence(table = streamTable(1000:0,colName,colType),
       tableName = `oneYearResult,cacheSize = 1000000,precache=1000000)    
}

def createDFS(){
    //创建存储计算1分钟窗口计算结果表
    if(existsDatabase("dfs://oneMinuteCalc")){dropDatabase("dfs://oneMinuteCalc")}
    db1 = database(, VALUE,2023.01.01..2023.01.03)
    db2 = database(, HASH,[SYMBOL,20])
    db = database(directory="dfs://oneMinuteCalc", partitionType=COMPO, partitionScheme=[db1,db2],engine="TSDB")
    colName =  `Time`deviceId`filterTime`MAX`MIN`MEAN`MED`P95`P5`CHANGE`CHANGE_RATE`first`last`endTime
    colType = `TIMESTAMP`SYMBOL`NANOTIMESTAMP join take(`DOUBLE,10) join `NANOTIMESTAMP
    t = table(1:0,colName,colType)
    pt = db.createPartitionedTable(table=t,tableName ="test" ,partitionColumns = ["Time","deviceId"],
            sortColumns =["deviceId","Time"],compressMethods={Time:"delta"})

    //创建存储计算5分钟窗口计算结果表        
    if(existsDatabase("dfs://fiveMinuteCalc")){dropDatabase("dfs://fiveMinuteCalc")}
    db = database(directory="dfs://fiveMinuteCalc", partitionType=VALUE,
           partitionScheme=2023.01.01..2023.01.03,engine="TSDB")
    t = table(1:0,colName,colType)
    pt = db.createPartitionedTable(table=t,tableName ="test" ,partitionColumns = ["Time"],
            sortColumns =["deviceId","Time"],compressMethods={Time:"delta"},
            sortKeyMappingFunction=[hashBucket{,100}])        

    //创建存储计算30分钟窗口计算结果表
    if(existsDatabase("dfs://halfHourCalc")){dropDatabase("dfs://halfHourCalc")}
    db = database(directory="dfs://halfHourCalc", partitionType=VALUE, 
            partitionScheme=2023.01.01..2023.01.03,engine="TSDB")
    t = table(1:0,colName,colType)
    pt = db.createPartitionedTable(table=t,tableName ="test" ,partitionColumns = ["Time"],
            sortColumns =["deviceId","Time"],compressMethods={Time:"delta"})  

    //创建存储计算1小时窗口计算结果表
    if(existsDatabase("dfs://oneHourCalc")){dropDatabase("dfs://oneHourCalc")}
    db = database(directory="dfs://oneHourCalc", partitionType=VALUE,
             partitionScheme=2023.01.01..2023.01.03,engine="TSDB")
    t = table(1:0,colName,colType)
    pt = db.createPartitionedTable(table=t,tableName ="test" ,partitionColumns = ["Time"],
            sortColumns =["deviceId","Time"],compressMethods={Time:"delta"})
            
    //创建存储计算24小时窗口计算结果表
    if(existsDatabase("dfs://oneDayCalc")){dropDatabase("dfs://oneDayCalc")}
    db = database(directory="dfs://oneDayCalc", partitionType=VALUE, 
          partitionScheme=2023.01.01..2023.01.03,engine="TSDB")
    t = table(1:0,colName,colType)
    pt = db.createPartitionedTable(table=t,tableName ="test" ,partitionColumns = ["Time"],
            sortColumns =["deviceId","Time"],compressMethods={Time:"delta"})

    //创建存储计算1年窗口计算结果表
    if(existsDatabase("dfs://oneYearCalc")){dropDatabase("dfs://oneYearCalc")}
    db = database(directory="dfs://oneYearCalc", partitionType=VALUE, 
          partitionScheme=2023.01.01..2023.01.03,engine="TSDB")
    t = table(1:0,colName,colType)
    pt = db.createPartitionedTable(table=t,tableName ="test" ,partitionColumns = ["Time"],
            sortColumns =["deviceId","Time"],compressMethods={Time:"delta"})
}

//1分钟窗口计算过滤函数
def filter1(msg){
    t = select *,now(true) as filterTime from msg 
    getStreamEngine(`dispatchDemo1).append!(t)
}

//5分钟窗口计算过滤函数
def filter2(msg){
    t = select *,now(true) as filterTime from msg 
    getStreamEngine(`dispatchDemo2).append!(t)
}

//30分钟窗口计算过滤函数
def filter3(msg){
    t = select *,now(true) as filterTime2 from msg 
    getStreamEngine(`halfHourCalc).append!(t)
}

//1小时窗口计算
def filter4(msg){
    t = select *,now(true) as filterTime2 from msg 
    getStreamEngine(`oneHourCalc).append!(t)
}

//24小时窗口计算
def filter5(msg){
    t = select *,now(true) as filterTime2 from msg 
    getStreamEngine(`oneDayCalc).append!(t)
}

clearEnv();
createStreamTable();
createDFS();

schemas1 = table(1:0,`Time`deviceId`value`filterTime,`TIMESTAMP`SYMBOL`DOUBLE`NANOTIMESTAMP)
metrics1 = <[first(filterTime),max(value),min(value),mean(value),med(value),percentile(value,95),
              percentile(value,5),last(value)-first(value),
              (last(value)-first(value))/first(value),first(value),last(value),now(true)]>
//创建1分钟窗口聚合计算引擎
for(i in 1..5){
    engine1 = createTimeSeriesEngine(name="oneMinuteCalc"+string(i), windowSize=60000, step=60000,
                metrics=metrics1 , dummyTable=schemas1 , outputTable=objByName(`oneMinuteResult),
                timeColumn = `Time, useSystemTime=false, keyColumn = `deviceId)
}

//创建5分钟窗口聚合计算引擎
for(i in 1..5){
    engine2 = createTimeSeriesEngine(name="fiveMinuteCalc"+string(i), windowSize=300000, step=300000, 
                metrics=metrics1 , dummyTable=schemas1 , outputTable=objByName(`fiveMinuteResult),
                timeColumn = `Time, useSystemTime=false, keyColumn = `deviceId)
}

//1分钟、5分钟窗口聚合计算分发引擎
dispatchEngine1=createStreamDispatchEngine(name="dispatchDemo1", dummyTable=schemas1, keyColumn=`deviceId, 
        outputTable=[getStreamEngine("oneMinuteCalc1"),getStreamEngine("oneMinuteCalc2"),
                        getStreamEngine("oneMinuteCalc3"),getStreamEngine("oneMinuteCalc4"),
                        getStreamEngine("oneMinuteCalc5")])
dispatchEngine2=createStreamDispatchEngine(name="dispatchDemo2", dummyTable=schemas1, keyColumn=`deviceId, 
        outputTable=[getStreamEngine("fiveMinuteCalc1"),getStreamEngine("fiveMinuteCalc2"),
                        getStreamEngine("fiveMinuteCalc3"),getStreamEngine("fiveMinuteCalc4"),
                        getStreamEngine("fiveMinuteCalc5")])


colName =  `Time`deviceId`filterTime`MAX`MIN`MEAN`MED`P95`P5`CHANGE`CHANGE_RATE`first`last`endTime`filterTime2
colType = `TIMESTAMP`SYMBOL`NANOTIMESTAMP join take(`DOUBLE,10) join `NANOTIMESTAMP`NANOTIMESTAMP
schemas2 = table(1:0,colName,colType)
metrics2 = <[first(filterTime2),max(MAX),min(MIN),mean(MEAN),med(MED),avg(P95),avg(P5),last(last)-first(first),
                (last(last)-first(first))/first(first),first(first),last(last),now(true)]>
//创建30分钟窗口聚合计算引擎
engine3 = createTimeSeriesEngine(name="halfHourCalc", windowSize=1800000, step=1800000, metrics=metrics2 , 
                dummyTable=schemas2 , outputTable=objByName(`halfHourResult),
                timeColumn = `Time, useSystemTime=false, keyColumn = `deviceId)

//创建1小时窗口聚合计算引擎
engine4 = createTimeSeriesEngine(name="oneHourCalc", windowSize=3600000, step=3600000, metrics=metrics2 , 
                dummyTable=schemas2 , outputTable=objByName(`oneHourResult),
                timeColumn = `Time, useSystemTime=false, keyColumn = `deviceId)

//创建24小时窗口聚合计算引擎
engine5 = createTimeSeriesEngine(name="oneDayCalc", windowSize=86400000, step=86400000, 
                metrics=metrics2 , dummyTable=schemas2 , outputTable=objByName(`oneDayResult),
                timeColumn = `Time, useSystemTime=false, keyColumn = `deviceId)


//订阅
subscribeTable(tableName=`inputStream, actionName="dispatch1", handler=filter1, msgAsTable = true,
          batchSize = 10240)
subscribeTable(tableName=`inputStream, actionName="dispatch2", handler=filter2, msgAsTable = true,
          batchSize = 10240)

subscribeTable(tableName=`oneMinuteResult, actionName="calcHalfHour", handler=filter3,
           msgAsTable = true,batchSize = 10240)
subscribeTable(tableName=`oneMinuteResult, actionName="calcOneHour", handler=filter4, 
          msgAsTable = true,batchSize = 10240)
subscribeTable(tableName=`fiveMinuteResult, actionName="calcOneDay", handler=filter5, 
          msgAsTable = true,batchSize = 10240)


subscribeTable(tableName = `oneMinuteResult,actionName=`appendInToDFS,offset=0,
          handler=loadTable("dfs://oneMinuteCalc","test"),
          msgAsTable=true,batchSize=10240)
subscribeTable(tableName = `fiveMinuteResult,actionName=`appendInToDFS,offset=0,
          handler=loadTable("dfs://fiveMinuteCalc","test"),
          msgAsTable=true,batchSize=10240)
subscribeTable(tableName = `halfHourResult,actionName=`appendInToDFS,offset=0,
          handler=loadTable("dfs://halfHourCalc","test"),
          msgAsTable=true,batchSize=10240)
subscribeTable(tableName = `oneHourResult,actionName=`appendInToDFS,offset=0,
          handler=loadTable("dfs://oneHourCalc","test"),
          msgAsTable=true,batchSize=10240)
subscribeTable(tableName = `oneDayResult,actionName=`appendInToDFS,offset=0,
          handler=loadTable("dfs://oneDayCalc","test"),
          msgAsTable=true,batchSize=10240)


def filter6(msg){
    tmp = select * ,now(true) as filterTime from msg 
    getStreamEngine(`oneYearCalc).append!(tmp)
}        

colName =  `Time`deviceId`MAX`MIN`MEAN`MED`P95`P5`CHANGE`CHANGE_RATE`first`last`filterTime
colType = `DATE`SYMBOL join take(`DOUBLE,10) join `NANOTIMESTAMP
schemas3 = table(1:0,colName,colType)
metrics3 = <[last(filterTime),max(MAX),min(MIN),mean(MEAN),med(MED),avg(P95),avg(P5),last(last)-first(first),
                (last(last)-first(first))/first(first),first(first),last(last),now(true)]>
engine6 = createTimeSeriesEngine(name="oneYearCalc", windowSize=365, step=1, metrics=metrics3 , 
            dummyTable=schemas3 , outputTable=objByName(`oneYearResult),
            timeColumn = `Time, useSystemTime=false, keyColumn = `deviceId)

subscribeTable(tableName = `oneDayResultSimulate,actionName=`calcOneYear, handler=filter6, 
            msgAsTable = true,batchSize = 10240)
subscribeTable(tableName = `oneYearResult,actionName=`appendInToDFS,offset=0,
            handler=loadTable("dfs://oneYearCalc","test"),
            msgAsTable=true)



deviceIdList = lapd(string(rand(10000,700)),6,"0")  //测点id

//模拟数据的函数，一共模拟1小时的数据
def simulateData(deviceIdList){
    num = deviceIdList.size()
    startTime = timestamp(2023.01.01)
    do{
        Time = take(startTime,num)
        deviceId = deviceIdList
        value = rand(100.0,num)
        objByName(`inputStream).append!(table(Time,deviceId,value))
        startTime = startTime+1000
        sleep(100)
    }while(startTime<=2023.01.02T00:00:10.000)
}

def simulateOneDay(deviceIdList){
    num = deviceIdList.size()
    startTime =2022.01.01
    do{
        Time = take(startTime,num)
        deviceId = deviceIdList
        MAX = rand(100.0,num)
        MIN = rand(100.0,num)
        MEAN = rand(100.0,num)
        MED = rand(100.0,num)
        P95 = rand(100.0,num)
        P5 = rand(100.0,num)
        CHANGE = rand(100.0,num)
        CHANGE_RATE = rand(100.0,num)
        first = rand(100.0,num)
        last = rand(100.0,num)
        tmp = table(Time,deviceId,MAX,MIN,MEAN,MED,P95,P5,CHANGE,CHANGE_RATE,first,last)
        objByName(`oneDayResultSimulate).append!(tmp)
        startTime = startTime+1
        sleep(500)
    }while(startTime<=2023.12.31)
}


submitJob("simulateData","write",simulateData,deviceIdList)
submitJob("simulateOneDay","write",simulateOneDay,deviceIdList)


//耗时统计
tmp1 = select Time,deviceId,filterTime,endTime from loadTable("dfs://oneYearCalc","test") order by Time,deviceId
tmp2 = select Time,deviceId,next(filterTime) as startTime,endTime from tmp1 context by deviceId
select avg(endTime-startTime)\1000\1000 as timeUsed from tmp2 group by deviceId  //统计单个测点的计算耗时
tmp3 = select min(startTime) as st,max(endTime) as dt from tmp2 group by Time  
select (dt-st)\1000\1000 as used from tmp3 //统计整个时间窗口的计算耗时