摘要本文深入讲解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实时统计分析技术。从滑动窗口原理到窗口函数应用从实时统计到趋势计算从多维度分析到性能优化全面介绍滑动窗口计算的核心方法。通过丰富的代码示例帮助读者掌握实时统计分析的核心技能。一、滑动窗口概述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流计算系列预告下一篇将介绍实时关联分析敬请期待
)
