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时间序列引擎