摘要

本文深入讲解DolphinDB实时聚合计算技术。从聚合函数到多维度聚合，从层级聚合到实时汇总，从分组统计到聚合优化，全面介绍实时聚合计算的核心方法。通过丰富的代码示例，帮助读者掌握多维度聚合的核心技能。

一、聚合计算概述

1.1 聚合类型

1.2 聚合函数

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(temperature>25,temperature,0))ashigh_temp_sum,sum(iif(temperature<=25,temperature,0))aslow_temp_sum,count(iif(temperature>25,1,0))ashigh_count,count(iif(temperature<=25,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){//按设备聚合 byDevice=select device_id,"all"ashour,sum(temperature)astotal,avg(temperature)asmeanfromdata group by device_id//按时间聚合 byHour=select"all"asdevice_id,bar(timestamp,1h)ashour,sum(temperature)astotal,avg(temperature)asmeanfromdata group by bar(timestamp,1h)//按设备和时间聚合 byBoth=select 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){//层级：设备->车间->工厂//设备级别 deviceLevel=select device_id,workshop,factory,sum(temperature)astotalfromdata group by device_id,workshop,factory//车间级别 workshopLevel=select"all"asdevice_id,workshop,factory,sum(temperature)astotalfromdata group by workshop,factory//工厂级别 factoryLevel=select"all"asdevice_id,"all"asworkshop,factory,sum(temperature)astotalfromdata group by factoryreturndeviceLevel.union(workshopLevel).union(factoryLevel)}

四、层级聚合

4.1 组织层级

//组织层级聚合defhierarchyAggregation(data,hierarchy){results=array(ANY,0)for(levelinhierarchy){agg=selecteval(level)aslevel_key,sum(temperature)astotal,avg(temperature)asmeanfromdata group byeval(level)results.append!(agg)}returnresults}

4.2 时间层级

//时间层级聚合deftimeHierarchyAggregation(data){//分钟级 minute=select bar(timestamp,1m)astime,avg(temperature)asmeanfromdata group by bar(timestamp,1m)//小时级 hour=select bar(timestamp,1h)astime,avg(temperature)asmeanfromdata group by bar(timestamp,1h)//天级 day=select 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=""){filtered=select*fromdata whereeval(filter)returnselecteval(toLevel)aslevel,sum(temperature)astotal,avg(temperature)asmeanfromfiltered group byeval(toLevel)}

五、实时聚合引擎

5.1 时间序列聚合

//创建流表 share streamTable(100000:0,`device_id`timestamp`temperature`humidity,[SYMBOL,TIMESTAMP,DOUBLE,DOUBLE])assensor_stream//创建聚合结果表 share table(1:0,`time_window`device_id`avg_temp`max_temp`min_temp`count,[TIMESTAMP,SYMBOL,DOUBLE,DOUBLE,DOUBLE,LONG])asagg_result//创建聚合引擎 aggEngine=createTimeSeriesEngine("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_window`device_id`avg_temp`avg_humid`max_temp`min_temp,[TIMESTAMP,SYMBOL,DOUBLE,DOUBLE,DOUBLE,DOUBLE])asmulti_agg multiAggEngine=createTimeSeriesEngine("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){key=row.device_idif(notaggState.has(key)){aggState[key]=dict(STRING,ANY,[["sum",0.0],["count",0],["max",-infinity],["min",infinity]])}state=aggState[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,numWorkers=4){results=array(ANY,0)//分区处理for(iin0..numWorkers){partition=select*fromdata where device_id%numWorkers=i results.append!(aggPartition(partition))}//合并结果returnmergeAggResults(results)}defmergeAggResults(results){totalSum=sum(each(def(r){r.sum},results))totalCount=sum(each(def(r){r.count},results))returndict(STRING,ANY,[["sum",totalSum],["count",totalCount],["avg",totalSum/totalCount]])}

6.3 预聚合

//预聚合表 share table(1:0,`device_id`hour`pre_sum`pre_count`pre_max`pre_min,[SYMBOL,TIMESTAMP,DOUBLE,LONG,DOUBLE,DOUBLE])aspre_agg//定时预聚合defpreAggregationTask(){while(true){now=now()hourStart=bar(now,1h)//聚合最近一小时数据 agg=select device_id,sum(temperature)aspre_sum,count(*)aspre_count,max(temperature)aspre_max,min(temperature)aspre_minfromsensor_stream where timestamp>=hourStart group by device_id pre_agg.append!(agg)sleep(3600000)}}

七、实战案例

7.1 完整实时聚合系统

//==========实时聚合计算系统==========//1.创建数据流 share streamTable(100000:0,`device_id`timestamp`temperature`humidity`pressure,[SYMBOL,TIMESTAMP,DOUBLE,DOUBLE,DOUBLE])assensor_stream enableTablePersistence(sensor_stream,true,true,1000000)//2.创建聚合结果表 share table(1:0,`time_window`device_id`avg_temp`avg_humid`max_temp`min_temp`count,[TIMESTAMP,SYMBOL,DOUBLE,DOUBLE,DOUBLE,DOUBLE,LONG])asagg_result//3.创建聚合引擎 aggEngine=createTimeSeriesEngine("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){t=loadTable("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){data=table(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实时聚合计算：

1. 基础聚合：单表聚合、分组聚合、条件聚合
2. 多维度聚合：多列分组、Cube聚合、Rollup聚合
3. 层级聚合：组织层级、时间层级、上卷下钻
4. 实时聚合引擎：时间序列聚合、多度量聚合、自定义聚合
5. 聚合优化：增量聚合、并行聚合、预聚合

思考题：

1. 如何设计高效的多维度聚合？
2. 如何优化实时聚合性能？
3. 如何处理聚合中的数据倾斜？

参考资料

• DolphinDB聚合函数
• DolphinDB时间序列引擎