DolphinDB实时统计分析：滑动窗口计算-北京尧图网络

摘要本文深入讲解DolphinDB实时统计分析技术。从滑动窗口原理到窗口函数应用从实时统计到趋势计算从多维度分析到性能优化全面介绍滑动窗口计算的核心方法。通过丰富的代码示例帮助读者掌握实时统计分析的核心技能。一、滑动窗口概述1.1 滑动窗口原理滑动窗口数据流窗口1窗口2窗口3统计结果1.2 窗口类型类型说明滚动窗口固定大小不重叠滑动窗口固定大小可重叠会话窗口基于活动间隔1.3 应用场景场景窗口类型分钟级统计滚动窗口移动平均滑动窗口会话分析会话窗口二、滚动窗口2.1 时间序列引擎//创建数据流 share streamTable(100000:0,device_idtimestamptemperaturehumidity,[SYMBOL,TIMESTAMP,DOUBLE,DOUBLE])assensor_stream//创建聚合结果表 share table(1:0,time_windowdevice_idavg_tempmax_tempmin_tempcount,[TIMESTAMP,SYMBOL,DOUBLE,DOUBLE,DOUBLE,LONG])asagg_result//创建时间序列引擎 aggEnginecreateTimeSeriesEngine(temp_agg,60000,//1分钟窗口[avg(temperature)asavg_temp,max(temperature)asmax_temp,min(temperature)asmin_temp,count(*)ascount],agg_result,timestamp,device_id)//订阅 subscribeTable(,sensor_stream,agg,-1,aggEngine,true)2.2 多窗口聚合//多窗口聚合defmultiWindowAgg(){//1分钟窗口 agg1mcreateTimeSeriesEngine(agg_1m,60000,[avg(temperature)asavg_temp],agg_1m_result,timestamp,device_id)//5分钟窗口 agg5mcreateTimeSeriesEngine(agg_5m,300000,[avg(temperature)asavg_temp],agg_5m_result,timestamp,device_id)//1小时窗口 agg1hcreateTimeSeriesEngine(agg_1h,3600000,[avg(temperature)asavg_temp],agg_1h_result,timestamp,device_id)}2.3 自定义聚合//自定义聚合函数defmyAgg(data){returndict(STRING,ANY,[[mean,avg(data)],[std,std(data)],[skewness,skewness(data)],[kurtosis,kurtosis(data)]])}三、滑动窗口3.1 移动平均//移动平均defmovingAverage(data,window){returnselect timestamp,mavg(temperature,window)asmafromdata}//使用 ttable(1..100asid,now()1..100*1000astimestamp,rand(20.0..30.0,100)astemperature)resultmovingAverage(t,10)3.2 移动统计//移动统计defmovingStats(data,window){returnselect timestamp,temperature,mavg(temperature,window)asmoving_avg,mstd(temperature,window)asmoving_std,mmax(temperature,window)asmoving_max,mmin(temperature,window)asmoving_min,msum(temperature,window)asmoving_sumfromdata}3.3 指数移动平均//指数移动平均defema(data,alpha0.1){resultarray(DOUBLE,data.rows())result[0]data[0]for(iin1..data.rows()){result[i]alpha*data[i](1-alpha)*result[i-1]}returnresult}四、实时统计4.1 实时计数//实时计数 share table(1:0,time_windowcount,[TIMESTAMP,LONG])ascount_result countEnginecreateTimeSeriesEngine(count_engine,60000,[count(*)ascount],count_result,timestamp)subscribeTable(,sensor_stream,count,-1,countEngine,true)4.2 实时求和//实时求和 share table(1:0,time_windowtotal_temp,[TIMESTAMP,DOUBLE])assum_result sumEnginecreateTimeSeriesEngine(sum_engine,60000,[sum(temperature)astotal_temp],sum_result,timestamp)subscribeTable(,sensor_stream,sum,-1,sumEngine,true)4.3 实时百分位//实时百分位defpercentileAgg(data,p){returnpercentile(data,p)}//使用 share table(1:0,time_windowp50p90p99,[TIMESTAMP,DOUBLE,DOUBLE,DOUBLE])aspercentile_result percentileEnginecreateTimeSeriesEngine(percentile_engine,60000,[percentile(temperature,50)asp50,percentile(temperature,90)asp90,percentile(temperature,99)asp99],percentile_result,timestamp)subscribeTable(,sensor_stream,percentile,-1,percentileEngine,true)五、趋势计算5.1 变化率//变化率defcalculateChangeRate(data){returnselect timestamp,temperature,deltas(temperature)aschange,deltas(temperature)/prev(temperature)*100aschange_ratefromdata}5.2 累计统计//累计统计defcumulativeStats(data){returnselect timestamp,temperature,cumsum(temperature)ascum_sum,cumavg(temperature)ascum_avg,cummax(temperature)ascum_max,cummin(temperature)ascum_minfromdata}5.3 趋势检测//趋势检测defdetectTrend(data,window10){resultselect timestamp,temperature,mavg(temperature,window)asma,iif(temperaturemavg(temperature,window),up,iif(temperaturemavg(temperature,window),down,stable))astrendfromdatareturnresult}六、多维度分析6.1 分组统计//分组统计defgroupStats(data,groupCol){returnselect avg(temperature)asavg_temp,max(temperature)asmax_temp,min(temperature)asmin_temp,std(temperature)asstd_temp,count(*)ascountfromdata group byeval(groupCol)}6.2 多维度聚合//多维度聚合defmultiDimAgg(data){returnselect device_id,bar(timestamp,1h)ashour,avg(temperature)asavg_temp,max(temperature)asmax_temp,min(temperature)asmin_tempfromdata group by device_id,bar(timestamp,1h)}6.3 交叉分析//交叉分析defcrossAnalysis(data){returnselect device_id,iif(temperature25,high,low)astemp_level,count(*)ascount,avg(humidity)asavg_humidityfromdata group by device_id,iif(temperature25,high,low)}七、性能优化7.1 增量计算//增量计算defincrementalAgg(newData,existingStats){//更新统计量 newCountexistingStats.countnewData.rows()newSumexistingStats.sumsum(newData.temperature)newAvgnewSum/newCountreturndict(STRING,ANY,[[count,newCount],[sum,newSum],[avg,newAvg]])}7.2 并行计算//并行计算defparallelAgg(data,numPartitions4){resultsarray(ANY,0)for(iin0..numPartitions){partitionselect*fromdata where device_id%numPartitionsi results.append!(aggPartition(partition))}//合并结果returnmergeResults(results)}八、实战案例8.1 完整实时统计系统//实时统计分析系统//1.创建数据流 share streamTable(100000:0,device_idtimestamptemperaturehumiditypressure,[SYMBOL,TIMESTAMP,DOUBLE,DOUBLE,DOUBLE])assensor_stream enableTablePersistence(sensor_stream,true,true,1000000)//2.创建聚合结果表 share table(1:0,time_windowdevice_idavg_tempmax_tempmin_tempstd_tempcount,[TIMESTAMP,SYMBOL,DOUBLE,DOUBLE,DOUBLE,DOUBLE,LONG])asagg_result//3.创建时间序列引擎 aggEnginecreateTimeSeriesEngine(sensor_agg,60000,[avg(temperature)asavg_temp,max(temperature)asmax_temp,min(temperature)asmin_temp,std(temperature)asstd_temp,count(*)ascount],agg_result,timestamp,device_id)subscribeTable(,sensor_stream,agg,-1,aggEngine,true)//4.移动统计 share table(1:0,timestampdevice_idtemperaturema_10ma_30,[TIMESTAMP,SYMBOL,DOUBLE,DOUBLE,DOUBLE])asma_resultdefcalculateMA(data){insert into ma_result select timestamp,device_id,temperature,mavg(temperature,10)asma_10,mavg(temperature,30)asma_30fromdata context by device_id}subscribeTable(,sensor_stream,ma,-1,def(msg){calculateMA(msg)},true)//5.模拟数据defgenerateMockData(){while(true){datatable(take(1..10,10)asdevice_id,take(now(),10)astimestamp,rand(20.0..30.0,10)astemperature,rand(40.0..60.0,10)ashumidity,rand(1000.0..1020.0,10)aspressure)sensor_stream.append!(data)sleep(5000)}}submitJob(mock_data,模拟数据,generateMockData)//6.查询接口defgetLatestStats(){returnselect*fromagg_result order by time_window desc limit10}addFunctionView(getLatestStats)print(实时统计分析系统启动完成)九、总结本文详细介绍了DolphinDB实时统计分析滚动窗口时间序列引擎、多窗口聚合滑动窗口移动平均、移动统计、指数移动平均实时统计实时计数、实时求和、实时百分位趋势计算变化率、累计统计、趋势检测多维度分析分组统计、多维度聚合、交叉分析性能优化增量计算、并行计算思考题如何选择合适的窗口大小如何优化滑动窗口计算性能如何处理乱序数据参考资料DolphinDB时间序列引擎DolphinDB流计算系列预告下一篇将介绍实时关联分析敬请期待

DolphinDB实时统计分析：滑动窗口计算

相关推荐

鸿蒙 ArkTS 实战：Travel Budget Ledger 从出行记录到状态反馈完整解析

第七届心理健康与教育、人文发展国际学术会议（MHEHD 2026）

【高届数艺术人文CPCI会议】第五届公共艺术与人文发展国际学术会议 (ICPAHD 2026)

更多资讯

Midscene.js 实战入门：AI 视觉驱动的跨平台自动化，5 分钟跑通第一个脚本

PSCAD齿轮箱参数配置与机电耦合仿真实践

Python开发者实战项目集：从爬虫到机器学习

基于主从博弈的共享储能微网优化运行MATLAB实现

背包问题解析：从贪心算法到实际应用

4 步跑通：kkFileView 文档预览元数据接入 KingbaseES 的数据库备份完整做法

今日更新

SpacetimeDB C++ 模块库实战：用 C++20 在数据库内构建 WebAssembly 模块

别踩雷！不是所有 AI 写作工具都靠谱，2026 高校认可工具精选

SSM+MySQL幼儿园管理系统源码解析：从部署到优化

本周精选

超人会飞不算本事：系统稳定依赖清晰规则与边界设计

超人VS蜘蛛侠：拆解超级IP的影响力与传播方法论

基于CNN的调制信号识别：MATLAB实现时频图分类实战

本月热门

基于YOLO的焊缝缺陷检测：从数据标注到工业部署全流程实践

分布式服务的决策记录

创业产品的首版范围

读完文章，想聊聊您的网站？