用戶定義的聚合函數(UDAFs)<一個類="headerlink" href="//m.eheci.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 |/ / + - - - - - - - - - - - - - - - - +