Spark SQL學習——UDF、UDAF和開窗函數
其餘更多java基礎文章：
java基礎學習(目錄)html
概述
SparkSQL中的UDF至關因而1進1出，UDAF至關因而多進一出，相似於聚合函數。sql
開窗函數通常分組取topn時經常使用。express
UDF
SparkConf conf = new SparkConf();
conf.setMaster("local");
conf.setAppName("udf");
JavaSparkContext sc = new JavaSparkContext(conf);
SQLContext sqlContext = new SQLContext(sc);
JavaRDD<String> parallelize = sc.parallelize(Arrays.asList("zhansan","lisi","wangwu"));
JavaRDD<Row> rowRDD = parallelize.map(new Function<String, Row>() {
/**
*
*/
private static final long serialVersionUID = 1L;
@Override
public Row call(String s) throws Exception {
return RowFactory.create(s);
}
});
List<StructField> fields = new ArrayList<StructField>();
fields.add(DataTypes.createStructField("name", DataTypes.StringType,true));
StructType schema = DataTypes.createStructType(fields);
DataFrame df = sqlContext.createDataFrame(rowRDD,schema);
df.registerTempTable("user");
/**
* 根據UDF函數參數的個數來決定是實現哪個UDF UDF1，UDF2。。。。UDF1xxx
*/
sqlContext.udf().register("StrLen", new UDF1<String,Integer>() {
/**
*
*/
private static final long serialVersionUID = 1L;
@Override
public Integer call(String t1) throws Exception {
return t1.length();
}
}, DataTypes.IntegerType);
sqlContext.sql("select name ,StrLen(name) as length from user").show();
複製代碼
new UDF1<String,Integer>()這些參數須要對應，UDF2就是表示傳兩個參數，UDF3就是傳三個參數。例如new UDF2<String, Integer, Integer>()，表示傳入兩個參數，第一個爲String類型，第二個爲Integer類型，返回Integer類型的數據。apache
UDAF
UDAF: 用戶自定義聚合函數，user defined aggreagatefunction
實現UDAF函數若是要自定義類要繼承UserDefinedAggregateFunction類api
package com.spark.sparksql.udf_udaf;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.api.java.function.Function;
import org.apache.spark.sql.DataFrame;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.RowFactory;
import org.apache.spark.sql.SQLContext;
import org.apache.spark.sql.expressions.MutableAggregationBuffer;
import org.apache.spark.sql.expressions.UserDefinedAggregateFunction;
import org.apache.spark.sql.types.DataType;
import org.apache.spark.sql.types.DataTypes;
import org.apache.spark.sql.types.StructField;
import org.apache.spark.sql.types.StructType;
/**
* UDAF 用戶自定義聚合函數
* @author root
*
*/
public class UDAF {
public static void main(String[] args) {
SparkConf conf = new SparkConf();
conf.setMaster("local").setAppName("udaf");
JavaSparkContext sc = new JavaSparkContext(conf);
SQLContext sqlContext = new SQLContext(sc);
JavaRDD<String> parallelize = sc.parallelize(
Arrays.asList("zhangsan","lisi","wangwu","zhangsan","zhangsan","lisi"));
JavaRDD<Row> rowRDD = parallelize.map(new Function<String, Row>() {
/**
*
*/
private static final long serialVersionUID = 1L;
@Override
public Row call(String s) throws Exception {
return RowFactory.create(s);
}
});
List<StructField> fields = new ArrayList<StructField>();
fields.add(DataTypes.createStructField("name", DataTypes.StringType, true));
StructType schema = DataTypes.createStructType(fields);
DataFrame df = sqlContext.createDataFrame(rowRDD, schema);
df.registerTempTable("user");
/**
* 註冊一個UDAF函數,實現統計相同值得個數
* 注意：這裏能夠自定義一個類繼承UserDefinedAggregateFunction類也是能夠的
*/
sqlContext.udf().register("StringCount",new UserDefinedAggregateFunction() {
/**
*
*/
private static final long serialVersionUID = 1L;
/**
* 初始化一個內部的本身定義的值,在Aggregate以前每組數據的初始化結果
*/
@Override
public void initialize(MutableAggregationBuffer buffer) {
buffer.update(0, 0);
}
/**
* 更新 能夠認爲一個一個地將組內的字段值傳遞進來 實現拼接的邏輯
* buffer.getInt(0)獲取的是上一次聚合後的值
* 至關於map端的combiner，combiner就是對每個map task的處理結果進行一次小聚合
* 大聚和發生在reduce端.
* 這裏便是:在進行聚合的時候，每當有新的值進來，對分組後的聚合如何進行計算
*/
@Override
public void update(MutableAggregationBuffer buffer, Row arg1) {
buffer.update(0, buffer.getInt(0)+1);
}
/**
* 合併 update操做，多是針對一個分組內的部分數據，在某個節點上發生的 可是可能一個分組內的數據，會分佈在多個節點上處理
* 此時就要用merge操做，將各個節點上分佈式拼接好的串，合併起來
* buffer1.getInt(0) : 大聚合的時候 上一次聚合後的值
* buffer2.getInt(0) : 此次計算傳入進來的update的結果
* 這裏便是：最後在分佈式節點完成後須要進行全局級別的Merge操做
* 也能夠是一個節點裏面的多個executor合併
*/
@Override
public void merge(MutableAggregationBuffer buffer1, Row buffer2) {
buffer1.update(0, buffer1.getInt(0) + buffer2.getInt(0));
}
/**
* 在進行聚合操做的時候所要處理的數據的結果的類型
*/
@Override
public StructType bufferSchema() {
return DataTypes.createStructType(Arrays.asList(DataTypes.createStructField("bffer111", DataTypes.IntegerType, true)));
}
/**
* 最後返回一個和DataType的類型要一致的類型，返回UDAF最後的計算結果
*/
@Override
public Object evaluate(Row row) {
return row.getInt(0);
}
/**
* 指定UDAF函數計算後返回的結果類型
*/
@Override
public DataType dataType() {
return DataTypes.IntegerType;
}
/**
* 指定輸入字段的字段及類型
*/
@Override
public StructType inputSchema() {
return DataTypes.createStructType(Arrays.asList(DataTypes.createStructField("nameeee", DataTypes.StringType, true)));
}
/**
* 確保一致性 通常用true,用以標記針對給定的一組輸入，UDAF是否老是生成相同的結果。
*/
@Override
public boolean deterministic() {
return true;
}
});
sqlContext.sql("select name ,StringCount(name) as strCount from user group by name").show();
sc.stop();
}
}
複製代碼
開窗函數
row_number() 開窗函數是按照某個字段分組，而後取另外一字段的前幾個的值，至關於分組取topN
開窗函數格式：row_number() over (partitin by XXX order by XXX)bash
package com.spark.sparksql.windowfun;
import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.sql.DataFrame;
import org.apache.spark.sql.SaveMode;
import org.apache.spark.sql.hive.HiveContext;
/**是hive的函數，必須在集羣中運行。
* row_number()開窗函數：
* 主要是按照某個字段分組，而後取另外一字段的前幾個的值，至關於 分組取topN
* row_number() over (partition by xxx order by xxx desc) xxx
* 注意：
* 若是SQL語句裏面使用到了開窗函數，那麼這個SQL語句必須使用HiveContext來執行，HiveContext默認狀況下在本地沒法建立
* @author root
*
*/
public class RowNumberWindowFun {
public static void main(String[] args) {
SparkConf conf = new SparkConf();
conf.setAppName("windowfun");
JavaSparkContext sc = new JavaSparkContext(conf);
HiveContext hiveContext = new HiveContext(sc);
hiveContext.sql("use spark");
hiveContext.sql("drop table if exists sales");
hiveContext.sql("create table if not exists sales (riqi string,leibie string,jine Int) "
+ "row format delimited fields terminated by '\t'");
hiveContext.sql("load data local inpath '/root/test/sales' into table sales");
/**
* 開窗函數格式：
* 【 row_number() over (partition by XXX order by XXX) as rank】//起個別名
* 注意：rank 從1開始
*/
/**
* 以類別分組，按每種類別金額降序排序，顯示 【日期，種類，金額】 結果，如：
*
* 1 A 100
* 2 B 200
* 3 A 300
* 4 B 400
* 5 A 500
* 6 B 600
* 排序後：
* 5 A 500 --rank 1
* 3 A 300 --rank 2
* 1 A 100 --rank 3
* 6 B 600 --rank 1
* 4 B 400 --rank 2
* 2 B 200 --rank 3
*
*/
DataFrame result = hiveContext.sql("select riqi,leibie,jine "
+ "from ("
+ "select riqi,leibie,jine,"
+ "row_number() over (partition by leibie order by jine desc) rank "
+ "from sales) t "
+ "where t.rank<=3");
result.show(100);
/**
* 將結果保存到hive表sales_result
*/
result.write().mode(SaveMode.Overwrite).saveAsTable("sales_result");
sc.stop();
}
}
複製代碼
相關文章
- 1. Spark SQL 中 UDF 和 UDAF 的使用
- 2. spark自定義hive的UDF/UDAF函數
- 3. 【Spark篇】---SparkSQL中自定義UDF和UDAF，開窗函數的應用
- 4. SparkSQL的UDF函數和UDAF函數
- 5. SparkSQL 用戶自定義函數（UDF、UDAF、開窗）
- 6. Spark SQL源碼函數解讀及UDF/UDAF例子 spark研習第六集
- 7. 詳解spark sql用戶自定義函數:UDF與UDAF
- 8. [2.5]詳解spark sql用戶自定義函數:UDF與UDAF
- 9. Spark SQL（二十二）用戶自定義的UDF、UDAF函數
- 10. Spark 2.4.0編程指南--Spark SQL UDF和UDAF
- 更多相關文章...
- • 我們已經學習了 SQL，下一步學習什麼呢？ - SQL 教程
- • 您已經學習了 XML Schema，下一步學習什麼呢？ - XML Schema 教程
- • 適用於PHP初學者的學習線路和建議
- • Tomcat學習筆記（史上最全tomcat學習筆記）
相關標籤/搜索
每日一句
-
每一个你不满意的现在，都有一个你没有努力的曾经。
最新文章
- 1. Appium入門
- 2. Spring WebFlux 源碼分析(2)-Netty 服務器啓動服務流程 --TBD
- 3. wxpython入門第六步（高級組件）
- 4. CentOS7.5安裝SVN和可視化管理工具iF.SVNAdmin
- 5. jedis 3.0.1中JedisPoolConfig對象缺少setMaxIdle、setMaxWaitMillis等方法，問題記錄
- 6. 一步一圖一代碼，一定要讓你真正徹底明白紅黑樹
- 7. 2018-04-12—（重點）源碼角度分析Handler運行原理
- 8. Spring AOP源碼詳細解析
- 9. Spring Cloud（1）
- 10. python簡單爬去油價信息發送到公衆號
歡迎關注本站公眾號,獲取更多信息
相關文章
- 1. Spark SQL 中 UDF 和 UDAF 的使用
- 2. spark自定義hive的UDF/UDAF函數
- 3. 【Spark篇】---SparkSQL中自定義UDF和UDAF，開窗函數的應用
- 4. SparkSQL的UDF函數和UDAF函數
- 5. SparkSQL 用戶自定義函數（UDF、UDAF、開窗）
- 6. Spark SQL源碼函數解讀及UDF/UDAF例子 spark研習第六集
- 7. 詳解spark sql用戶自定義函數:UDF與UDAF
- 8. [2.5]詳解spark sql用戶自定義函數:UDF與UDAF
- 9. Spark SQL（二十二）用戶自定義的UDF、UDAF函數
- 10. Spark 2.4.0編程指南--Spark SQL UDF和UDAF