用户定义的聚合函数(UDAFs)<一个类="headerlink" href="//www.neidfyre.com/docs.gcp/sql/language-manual/#user-defined-aggregate-functions-udafs" title="">

进口java.io.Serializable;进口org.apache.spark.sql.Dataset;进口org.apache.spark.sql.Encoder;进口org.apache.spark.sql.Encoders;进口org.apache.spark.sql.Row;进口org.apache.spark.sql.SparkSession;进口org.apache.spark.sql.expressions.Aggregator;进口org.apache.spark.sql.functions;公共静态类平均实现了可序列化的{私人长总和;私人长数;/ /构造函数,getter、setter……}公共静态类MyAverage扩展聚合器<长,平均,双>{/ /一个零值聚合。应该满足属性+ 0 = b吗公共平均零(){返回新平均(0l,0l);}/ /合并两个值来产生一个新值。针对性能、功能修改“缓冲”/ /并返回它,而不是构造一个新对象公共平均减少(平均缓冲,长数据){长newSum=缓冲。getSum()+数据;长newCount=缓冲。getCount()+1;缓冲。setSum(newSum);缓冲。setCount(newCount);返回缓冲;}/ /合并两个中间值公共平均合并(平均b1,平均b2){长mergedSum=b1。getSum()+b2。getSum();长mergedCount=b1。getCount()+b2。getCount();b1。setSum(mergedSum);b1。setCount(mergedCount);返回b1;}/ /转换的输出公共双完成(平均减少){返回((双)减少。getSum())/减少。getCount();}/ /中间值类型的编码器公共编码器<平均>bufferEncoder(){返回编码器。豆(平均。类);}/ /最终的输出值类型的编码器公共编码器<双>outputEncoder(){返回编码器。双();}}/ /注册函数来访问它火花。udf()。注册(“myAverage”,功能。udaf(新MyAverage(),编码器。长()));数据集<行>df=火花。读()。格式(“json”)。负载(“例子/ src / main /资源/ employees.json”);df。createOrReplaceTempView(“员工”);df。显示();/ / + - - - - - - - - - - - - - +/ /工资| | |名称/ / + - - - - - - - - - - - - - +/ / |迈克尔| 3000 |/ / |安迪| 4500 |/ / |贾斯汀| 3500 |/ / | Berta | 4000 |/ / + - - - - - - - - - - - - - +数据集<行>结果=火花。sql(“选择myAverage(工资)作为average_salary从员工”);结果。显示();/ / + - - - - - - - - - - - - - - - - +/ / | average_salary |/ / + - - - - - - - - - - - - - - - - +/ / | 3750.0 |/ / + - - - - - - - - - - - - - - - - +