DolphinDB实时聚合计算：多维度聚合-北京尧图网络

目录摘要一、聚合计算概述1.1 聚合类型1.2 聚合函数1.3 聚合维度二、基础聚合2.1 单表聚合2.2 分组聚合2.3 条件聚合三、多维度聚合3.1 多列分组3.2 Cube聚合3.3 Rollup聚合四、层级聚合4.1 组织层级4.2 时间层级4.3 上卷下钻五、实时聚合引擎5.1 时间序列聚合5.2 多度量聚合5.3 自定义聚合六、聚合优化6.1 增量聚合6.2 并行聚合6.3 预聚合七、实战案例7.1 完整实时聚合系统八、总结参考资料摘要本文深入讲解DolphinDB实时聚合计算技术。从聚合函数到多维度聚合从层级聚合到实时汇总从分组统计到聚合优化全面介绍实时聚合计算的核心方法。通过丰富的代码示例帮助读者掌握多维度聚合的核心技能。一、聚合计算概述1.1 聚合类型聚合计算单维度聚合聚合结果多维度聚合层级聚合1.2 聚合函数函数说明sum求和avg平均值max最大值min最小值count计数std标准差1.3 聚合维度维度说明时间维度按时间聚合设备维度按设备聚合产品维度按产品聚合区域维度按区域聚合二、基础聚合2.1 单表聚合//单表聚合defbasicAggregation(data){returnselectsum(temperature)astotal,avg(temperature)asmean,max(temperature)asmax_val,min(temperature)asmin_val,count(*)ascount,std(temperature)asstd_valfromdata}2.2 分组聚合//分组聚合defgroupAggregation(data,groupCol){returnselecteval(groupCol)asgroup_key,sum(temperature)astotal,avg(temperature)asmean,count(*)ascountfromdata group byeval(groupCol)}2.3 条件聚合//条件聚合defconditionalAggregation(data){returnselectsum(iif(temperature25,temperature,0))ashigh_temp_sum,sum(iif(temperature25,temperature,0))aslow_temp_sum,count(iif(temperature25,1,0))ashigh_count,count(iif(temperature25,1,0))aslow_countfromdata}三、多维度聚合3.1 多列分组//多列分组聚合defmultiDimAggregation(data){returnselect device_id,bar(timestamp,1h)ashour,sum(temperature)astotal,avg(temperature)asmean,max(temperature)asmax_val,min(temperature)asmin_val,count(*)ascountfromdata group by device_id,bar(timestamp,1h)}3.2 Cube聚合//Cube聚合多维度组合defcubeAggregation(data){//按设备聚合 byDeviceselect device_id,allashour,sum(temperature)astotal,avg(temperature)asmeanfromdata group by device_id//按时间聚合 byHourselectallasdevice_id,bar(timestamp,1h)ashour,sum(temperature)astotal,avg(temperature)asmeanfromdata group by bar(timestamp,1h)//按设备和时间聚合 byBothselect device_id,bar(timestamp,1h)ashour,sum(temperature)astotal,avg(temperature)asmeanfromdata group by device_id,bar(timestamp,1h)//合并returnbyDevice.union(byHour).union(byBoth)}3.3 Rollup聚合//Rollup聚合层级聚合defrollupAggregation(data){//层级设备-车间-工厂//设备级别 deviceLevelselect device_id,workshop,factory,sum(temperature)astotalfromdata group by device_id,workshop,factory//车间级别 workshopLevelselectallasdevice_id,workshop,factory,sum(temperature)astotalfromdata group by workshop,factory//工厂级别 factoryLevelselectallasdevice_id,allasworkshop,factory,sum(temperature)astotalfromdata group by factoryreturndeviceLevel.union(workshopLevel).union(factoryLevel)}四、层级聚合4.1 组织层级//组织层级聚合defhierarchyAggregation(data,hierarchy){resultsarray(ANY,0)for(levelinhierarchy){aggselecteval(level)aslevel_key,sum(temperature)astotal,avg(temperature)asmeanfromdata group byeval(level)results.append!(agg)}returnresults}4.2 时间层级//时间层级聚合deftimeHierarchyAggregation(data){//分钟级 minuteselect bar(timestamp,1m)astime,avg(temperature)asmeanfromdata group by bar(timestamp,1m)//小时级 hourselect bar(timestamp,1h)astime,avg(temperature)asmeanfromdata group by bar(timestamp,1h)//天级 dayselect date(timestamp)astime,avg(temperature)asmeanfromdata group by date(timestamp)returndict(STRING,ANY,[[minute,minute],[hour,hour],[day,day]])}4.3 上卷下钻//上卷聚合到更高层级defrollup(data,fromLevel,toLevel){returnselecteval(toLevel)aslevel,sum(temperature)astotal,avg(temperature)asmeanfromdata group byeval(toLevel)}//下钻展开到更低层级defdrilldown(data,fromLevel,toLevel,filter){filteredselect*fromdata whereeval(filter)returnselecteval(toLevel)aslevel,sum(temperature)astotal,avg(temperature)asmeanfromfiltered group byeval(toLevel)}五、实时聚合引擎5.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(sensor_agg,60000,[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)5.2 多度量聚合//多度量聚合 share table(1:0,time_windowdevice_idavg_tempavg_humidmax_tempmin_temp,[TIMESTAMP,SYMBOL,DOUBLE,DOUBLE,DOUBLE,DOUBLE])asmulti_agg multiAggEnginecreateTimeSeriesEngine(multi_agg,60000,[avg(temperature)asavg_temp,avg(humidity)asavg_humid,max(temperature)asmax_temp,min(temperature)asmin_temp],multi_agg,timestamp,device_id)subscribeTable(,sensor_stream,multi_agg,-1,multiAggEngine,true)5.3 自定义聚合//自定义聚合函数defcustomAgg(data){returndict(STRING,ANY,[[mean,avg(data)],[median,med(data)],[mode,mode(data)],[range,max(data)-min(data)],[iqr,percentile(data,75)-percentile(data,25)]])}六、聚合优化6.1 增量聚合//增量聚合 sharedict(STRING,ANY)asaggStatedefincrementalAgg(newData){for(rowinnewData){keyrow.device_idif(notaggState.has(key)){aggState[key]dict(STRING,ANY,[[sum,0.0],[count,0],[max,-infinity],[min,infinity]])}stateaggState[key]state[sum]row.temperature state[count]1state[max]max(state[max],row.temperature)state[min]min(state[min],row.temperature)}}6.2 并行聚合//并行聚合defparallelAgg(data,numWorkers4){resultsarray(ANY,0)//分区处理for(iin0..numWorkers){partitionselect*fromdata where device_id%numWorkersi results.append!(aggPartition(partition))}//合并结果returnmergeAggResults(results)}defmergeAggResults(results){totalSumsum(each(def(r){r.sum},results))totalCountsum(each(def(r){r.count},results))returndict(STRING,ANY,[[sum,totalSum],[count,totalCount],[avg,totalSum/totalCount]])}6.3 预聚合//预聚合表 share table(1:0,device_idhourpre_sumpre_countpre_maxpre_min,[SYMBOL,TIMESTAMP,DOUBLE,LONG,DOUBLE,DOUBLE])aspre_agg//定时预聚合defpreAggregationTask(){while(true){nownow()hourStartbar(now,1h)//聚合最近一小时数据 aggselect device_id,sum(temperature)aspre_sum,count(*)aspre_count,max(temperature)aspre_max,min(temperature)aspre_minfromsensor_stream where timestamphourStart group by device_id pre_agg.append!(agg)sleep(3600000)}}七、实战案例7.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_tempavg_humidmax_tempmin_tempcount,[TIMESTAMP,SYMBOL,DOUBLE,DOUBLE,DOUBLE,DOUBLE,LONG])asagg_result//3.创建聚合引擎 aggEnginecreateTimeSeriesEngine(sensor_agg,60000,[avg(temperature)asavg_temp,avg(humidity)asavg_humid,max(temperature)asmax_temp,min(temperature)asmin_temp,count(*)ascount],agg_result,timestamp,device_id)subscribeTable(,sensor_stream,agg,-1,aggEngine,true)//4.多维度聚合接口defgetMultiDimAgg(startTime,endTime){tloadTable(dfs://sensor_db,sensor_data)returnselect device_id,date(timestamp)asdate,bar(timestamp,1h)ashour,avg(temperature)asavg_temp,max(temperature)asmax_temp,min(temperature)asmin_temp,count(*)ascountfromt where timestamp between startTimeandendTime group by device_id,date(timestamp),bar(timestamp,1h)}addFunctionView(getMultiDimAgg)//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)print(实时聚合计算系统启动完成)八、总结本文详细介绍了DolphinDB实时聚合计算基础聚合单表聚合、分组聚合、条件聚合多维度聚合多列分组、Cube聚合、Rollup聚合层级聚合组织层级、时间层级、上卷下钻实时聚合引擎时间序列聚合、多度量聚合、自定义聚合聚合优化增量聚合、并行聚合、预聚合思考题如何设计高效的多维度聚合如何优化实时聚合性能如何处理聚合中的数据倾斜参考资料DolphinDB聚合函数DolphinDB时间序列引擎

DolphinDB实时聚合计算：多维度聚合

相关推荐

微信小程序二维码生成深度解析：weapp-qrcode架构设计与最佳实践

2026年，专业全屋定制设计公司揭秘

GitHub开源项目日报 · 2026年7月15日 · AI编码代理技能集爆火

更多资讯

高压电工证复训新规：不知道去哪报名怕被中介坑？费用明细全在这了

芒市考电工证几号怎么选培训机构附官方报名入口

临平考取电工证多少钱？实操考试心里没底怕挂科怎么破？

电梯安装需要电工证全流程办理指南，全国通用证件怎么考更省心

三门峡人考叉车证电工证刘佳别慌！良心建议帮你少走弯路

电力电缆证是什么证值不值得考电工从业者最怕的3个问题全解析

今日更新

金坛办电工证在哪里？考前押题能帮你少走弯路

怎样查看真假电工证怎么考

史密斯内部电工证值不值得考？怕考不过白交培训费的看过来

本周精选

安监局电工证分数查询怎么查才能知道含金量高不高

新疆电工作业干活累吗良心建议助你避开考不过的坑

中牟县电工证怎么报名外省证能转吗

本月热门

面向复杂光照与角度的宠物面部识别鲁棒性增强策略

基于DRV8818与STM32的双极步进电机驱动方案实战解析

DRV8818PWPR+PIC18F45K80双极步进电机控制方案实战

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