DolphinDB自定义聚合函数：UDAF详解

摘要

本文深入讲解DolphinDB自定义聚合函数开发。从聚合函数原理到状态管理，从map-reduce模式到性能优化，从窗口聚合到分布式聚合，全面介绍UDAF开发的核心方法。通过丰富的代码示例，帮助读者掌握自定义聚合函数的核心技能。

一、自定义聚合函数概述

1.1 什么是聚合函数

聚合函数将多行数据聚合为一个结果：

1.2 为什么需要UDAF

二、聚合函数原理

2.1 Map-Reduce模式

2.2 状态管理

//聚合函数需要维护状态//例如：计算平均值需要维护sum和 count//状态结构classAggState{sum=0count=0defupdate(value){sum+=value count+=1}deffinalize(){returnsum/count}}

三、创建UDAF

3.1 使用defg定义

//使用defg定义聚合函数 defg mySum(x){returnsum(x)}//使用 t=table(1..10asvalue)select mySum(value)fromt//55

3.2 使用map-reduce

//Map-Reduce聚合函数defmyAvgMap(x){return[sum(x),count(x)]}defmyAvgReduce(mapResults){totalSum=sum(mapResults[0])totalCount=sum(mapResults[1])returntotalSum/totalCount}//注册聚合函数 addAggregator("myAvg",myAvgMap,myAvgReduce)//使用 t=table(1..10asvalue)select myAvg(value)fromt//5.5

3.3 完整UDAF示例

//计算加权平均defweightedAvgMap(values,weights){return[sum(values*weights),sum(weights)]}defweightedAvgReduce(mapResults){totalWeightedSum=sum(mapResults[0])totalWeights=sum(mapResults[1])returntotalWeightedSum/totalWeights}addAggregator("weightedAvg",weightedAvgMap,weightedAvgReduce)//使用 t=table(1..10asvalue,[1,1,1,1,1,2,2,2,2,2]asweight)select weightedAvg(value,weight)fromt

四、窗口聚合

4.1 累计聚合

//累计聚合函数 defg cumAvg(x){returncumsum(x)\ cumcount(x)}//使用 t=table(1..10asvalue)select value,cumAvg(value)ascum_avgfromt

4.2 滑动窗口聚合

//滑动窗口聚合defmovingStd(x,window){returnmstd(x,window)}//使用 t=table(1..100asvalue)select value,movingStd(value,10)asmoving_stdfromt

4.3 时间窗口聚合

//时间窗口聚合deftimeWindowAvg(timestamp,value,window){returnmavg(value,window)}//使用 t=table(2024.01.01T00:00:00+0..99*60000astimestamp,rand(20.0..30.0,100)astemperature)select timestamp,temperature,timeWindowAvg(timestamp,temperature,10)asavg_10fromt

五、分布式聚合

5.1 分布式聚合原理

5.2 分布式聚合示例

//创建分布式表 db=database("dfs://agg_db",VALUE,1..100)schema=table(1:0,`device_id`timestamp`value,[INT,TIMESTAMP,DOUBLE])db.createPartitionedTable(schema,`sensor_data,`device_id)//插入数据 loadTable("dfs://agg_db","sensor_data").append!(table(take(1..100,100000)asdevice_id,take(now(),100000)astimestamp,rand(20.0..30.0,100000)asvalue))//分布式聚合 t=loadTable("dfs://agg_db","sensor_data")//使用自定义聚合函数 select device_id,myAvg(value)asavg_valuefromt group by device_id

5.3 分区聚合优化

//分区聚合：利用分区并行计算 select device_id,avg(value)asavg_valuefromt group by device_id//分区裁剪优化 select avg(value)asavg_valuefromt where device_idin1..10

六、实战案例

6.1 统计指标聚合

//==========统计指标聚合函数==========//计算变异系数 defg cv(x){returnstd(x)/avg(x)}//计算偏度 defg skewness(x){n=count(x)m=avg(x)s=std(x)returnsum((x-m)^3)/(n*s^3)}//计算峰度 defg kurtosis(x){n=count(x)m=avg(x)s=std(x)returnsum((x-m)^4)/(n*s^4)-3}//使用 t=table(rand(20.0..30.0,1000)asvalue)select cv(value)ascv,skewness(value)asskew,kurtosis(value)askurtfromt

6.2 时间序列聚合

//==========时间序列聚合函数==========//计算时间序列斜率 defg slope(timestamp,value){n=count(value)sumX=sum(timestamp)sumY=sum(value)sumXY=sum(timestamp*value)sumX2=sum(timestamp*timestamp)return(n*sumXY-sumX*sumY)/(n*sumX2-sumX*sumX)}//计算时间序列截距 defg intercept(timestamp,value){n=count(value)avgX=avg(timestamp)avgY=avg(value)returnavgY-slope(timestamp,value)*avgX}//使用 t=table(1..100astime,2*(1..100)+rand(-5.0..5.0,100)asvalue)select slope(time,value)asslope,intercept(time,value)asinterceptfromt

6.3 工业指标聚合

//==========工业指标聚合函数==========//计算OEE defg calculateOEE(availability,performance,quality){returnavg(availability*performance*quality)*100}//计算合格率 defg passRate(values,lowerLimit,upperLimit){returnsum(values>=lowerLimitandvalues<=upperLimit)/count(values)*100}//计算CPK defg cpk(values,lowerLimit,upperLimit){m=avg(values)s=std(values)cpu=(upperLimit-m)/(3*s)cpl=(m-lowerLimit)/(3*s)returnmin(cpu,cpl)}//使用 t=table(rand(95.0..105.0,1000)asmeasurement)select passRate(measurement,90,110)aspass_rate,cpk(measurement,90,110)ascpkfromt

七、性能优化

7.1 向量化计算

//向量化：使用向量化操作 defg fastSum(x){returnsum(x)//向量化}//避免循环 defg slowSum(x){total=0for(vinx){total+=v//非向量化，慢}returntotal}

7.2 状态优化

//状态优化：减少中间结果 defg optimizedAvg(x){//直接计算，不存储中间结果returnsum(x)/count(x)}

7.3 内存优化

//内存优化：使用流式计算 defg streamingAgg(x){//流式计算，不存储全部数据returnsum(x)}

八、总结

本文详细介绍了DolphinDB自定义聚合函数：

1. 聚合原理：Map-Reduce模式、状态管理
2. 创建方法：defg定义、map-reduce注册
3. 窗口聚合：累计聚合、滑动窗口、时间窗口
4. 分布式聚合：分布式原理、分区优化
5. 实战应用：统计指标、时间序列、工业指标
6. 性能优化：向量化、状态优化、内存优化

思考题：

1. 如何设计分布式聚合函数？
2. 窗口聚合和普通聚合有什么区别？
3. 如何优化聚合函数性能？

参考资料

• DolphinDB自定义聚合函数
• DolphinDB聚合函数