工业试验平台高频数据的存储和处理分析

DolphinDB 具有丰富而多样化的查询功能，其中包括总数查询、条件查询、单点查询和范围查询等多种功能，可以满足客户的不同查询需求。并且，DolphinDB 能够轻松地处理海量数据，实现毫秒级的查询响应。

/**** 数据查询 Demo ****/

//加载表，赋值给 pt （表变量赋值仅加载元数据）
pt=loadTable(database="dfs://db_test",tableName=`collect)  //pt(ts,deviceid,v1,...,v5000)

//查看数据总数
select count(*) from pt 

//数据检查
select max(ts),min(v1),max(v2),avg(v3),sum(v4),last(v1) from pt     //最大值、最小值、平均值、最后一条数据 的查询
select top 10 v1,v2,v3 from pt where ts between 2022.01.01 00:00:00.000 : 2022.01.01 00:01:00.000

//单点查询（返回单值）
timer t = select v1 from pt where deviceid='d001',ts=2022.01.01 00:00:00.001

//范围查询
timer t=select ts,v1 from pt where deviceid='d001', ts between 2022.01.01 00:00:00.000 : 2022.01.01 00:00:01.000

降采样是指将数字信号的采样率降低的过程，即通过去除部分采样点来减少信号的采样率。降采样通常用于降低信号的复杂度，以便减少计算量和存储空间。DolphinDB 为用户提供了 bar 函数，能够轻松实现降采样过程，帮助用户有效管理大规模数据。

滑动平均，也叫做指数加权平均，可以用来估计变量的局部均值，使得变量的更新与一段时间内的历史取值有关。通过 DolphinDB 提供的 mavg 函数，用户能够使用一条语句轻松实现滑动平均的计算，这极大地提高了数据的处理效率。

/**** 数据分析 ****/

//每 30 秒钟，取一条数据（降采样）
select ts,v1 from pt context by bar(ts,30s) limit -1        //limit -1 代表取最后一条记录

//30 秒钟滑动平均计算
timer t=select ts,deviceid,mavg(v1,30) as val from pt where deviceid=deviceid , ts between 2022.01.01 00:00:00.000 : 2022.01.01 00:01:00.000 context by deviceid

为了方便用户在不同数据源和目标之间进行数据转换和传输，DolphinDB 提供了多种数据导入导出方式。在数据导出方面，DolphinDB 支持两种方式：使用 saveTable 将一个表对象以二进制形式保存到文件中，或使用 saveText 将数据保存到文本文件中。

DolphinDB 还提供了多种灵活的数据导入方法，以帮助用户轻松地从多个数据源导入海量数据。对于本地数据，DolphinDB 提供了两种方式：CSV 文本文件导入和二进制文件导入。对于外部数据，DolphinDB 支持多种方式：二进制文件导入，ODBC、MySQL、MongoDB、HBase 等第三方数据库导入，还支持连接消息中间件以发布和订阅消息。

/**** 数据导出和导入 ****/

//数据准备：将1小时数据，赋值给临时表
t=NULL
t=select * from pt where ts between 2022.01.01 00:00:00.000 : 2022.01.01 01:00:00.000

//数据准备：路径，文件名
path=getHomeDir()  
filename="savetable_output.csv"

//1. 导出表
db=database(directory=path+"/output_db")
saveTable(dbHandle=db,table=t,tableName=`collect)

//2. 导出 csv 
saveText(obj=t, filename=path+"/"+filename)

//3. 导入表
db=database(directory=path+"/output_db")
t=loadTable(database=db,tableName=`collect)

//4. 导入 csv
t=NULL  //释放内存
ds=textChunkDS(path+"/"+filename,1000)
mr(ds,append!{loadTable("dfs://db_temp2","collect_temp")},,,false);

DolphinDB 提供了简单易用的数据备份和恢复功能，帮助用户保护重要数据。使用 DolphinDB 编程语言，用户可以自定义所需的备份策略和恢复方案。DolphinDB 以分区为单位进行数据备份，每个分区备份为一个数据文件。用户可以使用 backup 函数指定要备份的数据库、表或分区，以及备份文件的路径。使用 migrate 函数可以恢复数据库中已备份的数据。

/**** 数据备份和备份恢复 ****/

//1. 路径
path=getHomeDir()

//2. 备份
backup(path,<select * from loadTable("dfs://db_test","collect")>,true);

//3. 恢复备份
//使用 migrate 函数，完整恢复备份。恢复到新建的表中。
migrate(backupDir=path,backupDBPath="dfs://db_test",backupTableName="collect",newDBPath="dfs://db_temp3",newTableName=("collect_temp"));
loadTable(database="dfs://db_temp3",tableName="collect_temp") 

DolphinDB 对关系模型的支持包括以下几点：

• 除了包含时间戳的时序数据，还支持存储和处理不含时间戳的关系型数据
• 支持标准的 SQL 语言。包括常用的 select、insert、update 和 delete 等操作，group by 和 order by 等子句，还扩展了 context by 和 pivot by 以扩展数据的分析能力
• 支持复杂的多表关联，包括 inner join、left join、left semijoin 和 full join，已经拓展的 asof join 和 window join 以便用户能够方便地将不同数据表中的数据进行关联和整合。

这些功能的支持使得 DolphinDB 能够更好地满足用户对于数据处理和分析的需求，提高工作效率和数据处理的准确性。

/**** 关系模型 ****/
//关联查询
pt=loadTable("dfs://db_test",`collect)  //pt（ts,deviceid,v1,...,v5000） dt(deviceid,model,ip)
select pt.ts,pt.v1,pt.v2,dt.* from pt left join dt on pt.deviceid=dt.deviceid where pt.ts=2022.01.01 00:00:01.001

DolphinDB 提供了历史数据回放的功能。该功能可以帮助用户更加方便地发现历史数据中的异常，从而有助于排查试验故障。具体实现方式是通过 replay 函数，将内存表或数据库表中的记录按照一定速率写入到目标表中，以模拟实时数据流不断注入目标表的场景。

replay 函数支持单表回放和多表回放两种模式。在最基础的单表回放模式中，一个输入表将被回放至一个具有相同表结构的目标表中。而在多表回放模式中，多个输入表将被回放至多个目标表中，且输入表与目标表一一对应。

通过 DolphinDB 的历史数据回放功能，用户可以更加灵活地进行数据分析，并且能够更加快速地定位数据异常问题。这一功能的应用范围非常广泛，可以被广泛应用于各种数据分析场景中。

/**** 历史回放功能（可用于试验故障排查） ****/

//1. 新建流数据表，容纳回放数据
share streamTable(1000:0,`ts`deviceid,[TIMESTAMP,SYMBOL]) as replayStream
for(i in 1..3){  //取测点v1,v2,v3用于演示
    addColumn(replayStream,`v+string(i),DOUBLE)
}

//2. 数据回放（10 分钟）
rate=1                                  //回放倍速
begintime=2022.01.01 00:00:00.000       //数据开始时间
endtime  =2022.01.01 00:10:00.000       //数据结束时间
t=select ts,deviceid,v1,v2,v3 from pt where ts between begintime:endtime order by ts,deviceid

//3. 批处理调用 replay 函数，后台执行回放
submitJob("replay_output", "replay_output_stream", replay,  t,replayStream, `ts, `ts, rate,false)

//4. 展示回放数据（top 1000）
t=select * from replayStream limit 1000
plot(t.v1,t.ts,"top 1000 value 1")

数据回放

录制某段波形数据在工业场景中是一项很常见的需求，录制关键数据可以提高故障排查的效率。例如，当某个指标出现 20% 的幅度偏移时，记录前 10 秒和后 10 秒内的数据，自动保存备查。在 DolphinDB 中能通过自定义函数轻松实现波形录制功能，并且可以根据用户的需求进行个性化功能定制。

/**** 录制某段波形数据 ****/

//1. 建表，保存录制信息
n=5000 //采集点数量
dbname="dfs://db_recodeing"
tablename="recode"
cols_info=`ts`deviceid
cols_type=`TIMESTAMP`SYMBOL
t=table(1:0,cols_info,cols_type)

for(i in 1..n){
    addColumn(t,`v+string(i),DOUBLE)
}
  
db_hour=database(partitionType=VALUE,partitionScheme=[datehour(2022.10.01)])
db_hash=database(partitionType=HASH, partitionScheme=[SYMBOL,5])
db=database(directory=dbname,partitionType=COMPO,partitionScheme=[db_hour,db_hash],engine=`TSDB)
pt=createPartitionedTable(dbHandle=db,table=t,tableName=tablename,partitionColumns=`ts`deviceid,sortColumns=`deviceid`ts)

//2. 自定义录制函数
def funRecodeData(wtid,mutable begintime,second){

    //合法性判断
    if(second>180){
        strMsg='录制时间需小于3分钟'
        return strMsg
    }

    if(typestr(begintime)=="STRING"){ begintime=timestamp(begintime) }
    endtime=temporalAdd(begintime,second,`s)

    //录制未来数据
    diff=endtime-now()
    if(diff>0){
        sleep(diff)
    }

    //录制
    t=select * from loadTable("dfs://db_test",`collect) where ts between begintime:endtime and deviceid=wtid
    pt=loadTable(database="dfs://db_recodeing",tableName=`recode)
    pt.append!(t)

    //返回消息，供应用端展示
    rownum=t.size()
    strMsg='录制行数：'+string(rownum)+''
    return strMsg
}

//3. 调用函数
deviceid = 'd001'
second = 5
begintime = 2022.01.01 00:10:00.000
msg=funRecodeData(deviceid,begintime,second)
msg

DolphinDB 提供了函数视图来灵活地控制用户对数据库和表的访问。使得用户即使不具备读写数据库原始数据的权限，也可通过执行函数视图，间接访问数据库，得到所需计算结果。例如，用户不能查看个别患者的名称和年龄等原始数据，但是可以获取患者的总数与平均年龄。

/**** 存储过程的实现（函数视图） ****/

//建立函数视图（存储过程），可通过外部 API 调用该函数
addFunctionView(funRecodeData)

//删除函数视图（存储过程）
dropFunctionView(`funRecodeData)

在工业场景中，统计设备使用时长的需求十分常见。通过对设备使用时长的统计，可以实现设备保养预警和设备生命周期管理，同时也能够帮助运维人员更好地进行设备维护，从而延长设备的使用寿命。DolphinDB 提供了统计设备使用时长的功能，可以为运维人员制定维保计划提供数据支持，实现预防性维护。

具体实现过程中，设备信息表会记录不同设备的开关机时间，表中包含时间戳、设备 ID 和设备状态三个字段。统计设备使用时长需要计算设备的关机时间和开机时间之差，并将差值进行总和。

DolphinDB 使用列式存储，可以轻松地计算出每列相邻元素之间的差值，因此可以快速计算出设备使用时长。相比之下，MySQL 和 PostgreSQL 等传统的关系数据库使用行式存储，需要迭代才能计算出每列相邻元素之间的差值，这种方法繁琐复杂且耗时长。

/**** 统计设备使用时长 ****/
//1. 生成数据
n=100
times=datetime(2022.01.01)+1..864000
ts=sort(rand(times,n))
status=rand(1..3,n)
deviceid=rand("D"+lpad(string(1..9),2,"00"),n)
t=table(ts,deviceid, status)

//2. 汇总设备状态时长
dt = select *, deltas(ts) as duration from t context by deviceid
dt=select sum(duration)/3600 as totalhour from dt where status=1 group by deviceid, status 

//3. 返回使用时长
select deviceid,string(totalhour)+"小时" as totalUse from dt order by totalhour desc

DolphinDB 数据库功能强大，可以在实时数据流计算场景中高效地进行计算。用户可以通过简单的表达式实现复杂的流计算。DolphinDB 的流式数据处理引擎能够实现低延迟高吞吐的流式数据分析，并提供了十种不同的引擎以满足不同的计算需求。此外，DolphinDB 数据库支持多种数据终端输出，如共享内存表、流数据表、消息中间件、数据库和 API 等。在计算复杂表达式时，用户还可以通过级联多个流数据引擎来实现复杂的数据流拓扑。此外，DolphinDB 数据库还提供了 createTimeSeriesEngine 函数，可以进一步提高计算效率，实现基于时间的滑动窗口或滚动窗口进行实时计算。

/**** 时间序列聚合引擎 ****/

//定义保存计算结果的流表
share streamTable(1000:0, `ts`deviceid`v1`v2`v3,[TIMESTAMP,SYMBOL,FLOAT,FLOAT,FLOAT]) as calc_result

//计算引擎，实时计算均值指标（窗口 10s，步长 1s）
tsAggr_iot = createTimeSeriesEngine(name="tsAggr_iot",  windowSize=10*1000, step=1*1000, metrics=<[avg(v1),avg(v2),max(v3)]>, dummyTable=replayStream, outputTable=calc_result, timeColumn=`ts, keyColumn=`deviceid)
subscribeTable(tableName="replayStream", actionName="act_tsAggr_iot", offset=0, handler=append!{tsAggr_iot}, msgAsTable=true, batchSize=50, throttle=1, hash=0)

t=select top 100 * from calc_result order by ts desc
title=select min(ts),max(ts) from t
plot(t.v1,t.ts,title.min_ts[0]+"  ~  "+title.max_ts[0])ame="collect", diskUsage=true,top=0})