Spark SQL 中UDF的講解

User Define Function, 用戶自定義函數(shù),簡(jiǎn)稱UDF,存在與很多組件中。

在使用Sparksql的人都遇到了Sparksql所支持的函數(shù)太少了的難處，除了最基本的函數(shù)，Sparksql所能支撐的函數(shù)很少，肯定不能滿足正常的項(xiàng)目使用，UDF可以解決問(wèn)題。

SparkSQL中的UDF相當(dāng)于是1進(jìn)1出，UDAF相當(dāng)于是多進(jìn)一出，類(lèi)似于聚合函數(shù)。

開(kāi)窗函數(shù)一般分組取topn時(shí)常用。

可以理解為自己定義函數(shù)，來(lái)獲取自己想要的結(jié)果！ 案例借鑒于網(wǎng)絡(luò)！ 需求：計(jì)算文本中每一個(gè)單詞的長(zhǎng)度！ 代碼： Scala版本： package com.bynear.Scalaimport org.apache.spark.sql.types.{StringType, StructField, StructType} import org.apache.spark.sql.{Row, SQLContext} import org.apache.spark.{SparkConf, SparkContext}object UDF {def main(args: Array[String]): Unit = {val conf = new SparkConf().setAppName("SparkSQL_UDF").setMaster("local")val sc = new SparkContext(conf)val sqlContext = new SQLContext(sc)val names = Array("劉亦菲", "張柏芝", "馮提模","陳一發(fā)兒")val nameRDD = sc.parallelize(names, 5)val nameRowRDD = nameRDD.map(name => Row(name))val structType = StructType(Array(StructField("name", StringType, true)))val namesDF = sqlContext.createDataFrame(nameRowRDD, structType)namesDF.registerTempTable("names")sqlContext.udf.register("strLen", (str: String) => str.length)sqlContext.sql("select name,strLen(name) as length from names").show()sqlContext.sql("select name,strLen(name) as length from names").collect().foreach(println)} } 運(yùn)行結(jié)果： +----+------+
|name|length|
+----+------+
| 劉亦菲|? ? ?3|
| 張柏芝|? ? ?3|
| 馮提模|? ? ?3|
|陳一發(fā)兒|? ? ?4|
+----+------+

[劉亦菲,3]
[張柏芝,3]
[馮提模,3]
[陳一發(fā)兒,4]

Java版本： package com.bynear.spark_sql; 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.api.java.UDF1; import org.apache.spark.sql.types.DataTypes; import org.apache.spark.sql.types.StructField; import org.apache.spark.sql.types.StructType; import java.util.ArrayList; public class JavaUDF {public static void main(String[] args) {SparkConf conf = new SparkConf().setAppName("JavaUDF").setMaster("local"); JavaSparkContext sc = new JavaSparkContext(conf); SQLContext sqlContext = new SQLContext(sc.sc()); ArrayList<String> names = new ArrayList<String>(); names.add("劉亦菲"); names.add("張柏芝"); names.add("馮提模"); names.add("陳一發(fā)兒"); JavaRDD<String> nameRDD = sc.parallelize(names); JavaRDD<Row> nameRowRDD = nameRDD.map(new Function<String, Row>() {@Override public Row call(String line) throws Exception {return RowFactory.create(String.valueOf(line)); }}); /** * 使用動(dòng)態(tài)編程方式，將RDD轉(zhuǎn)換為Dataframe */ ArrayList<StructField> fields = new ArrayList<StructField>(); fields.add(DataTypes.createStructField("name", DataTypes.StringType, true)); StructType structType = DataTypes.createStructType(fields); DataFrame nameDF = sqlContext.createDataFrame(nameRowRDD, structType); /** * 注冊(cè)臨時(shí)表 */ nameDF.registerTempTable("user"); /** * 根據(jù)UDF函數(shù)參數(shù)的個(gè)數(shù)來(lái)決定是實(shí)現(xiàn)哪一個(gè)UDF UDF1，UDF2, 表明包含幾個(gè)參數(shù)傳入 * UDF1<String, Integer> 表示 傳入?yún)?shù) String 輸出參數(shù)為 Integer * call方法為 自定義的函數(shù)！ * DataTypes.IntegerType 必須與輸出參數(shù)的類(lèi)型一致即 Integer */ sqlContext.udf().register("StrLen", new UDF1<String, Integer>() {@Override public Integer call(String s) throws Exception {return s.length(); }}, DataTypes.IntegerType); /** * select name ,StrLen(name) as length from user * 在臨時(shí)表user中 查找name StrLen(name) == name的長(zhǎng)度 * StrLen(name) as length 表示將獲取到的name的長(zhǎng)度 例如15 15作為一列 as length 列名為 length */ sqlContext.sql("select name ,StrLen(name) as length from user").show(); Row[] rows = sqlContext.sql("select name ,StrLen(name) as length from user").collect(); for (Row row : rows) {System.out.println(row); }sc.close(); } } 輸出結(jié)果：同上！
Java版本中，主要之一到UDFX 方法，以及傳入?yún)?shù)的個(gè)數(shù)類(lèi)型，以及輸出類(lèi)型，最終要的是文本最后的 DataTypes.IntegerType? 類(lèi)型要與輸出類(lèi)型相同！

總結(jié)

以上是生活随笔為你收集整理的Spark SQL 中UDF的讲解的全部?jī)?nèi)容，希望文章能夠幫你解決所遇到的問(wèn)題。

如果覺(jué)得生活随笔網(wǎng)站內(nèi)容還不錯(cuò)，歡迎將生活随笔推薦給好友。