Spark SQL 中UDF的講解

User Define Function, 用戶自定義函數,簡稱UDF,存在與很多組件中。

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

SparkSQL中的UDF相當于是1進1出，UDAF相當于是多進一出，類似于聚合函數。

開窗函數一般分組取topn時常用。

可以理解為自己定義函數，來獲取自己想要的結果！ 案例借鑒于網絡！ 需求：計算文本中每一個單詞的長度！ 代碼： 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("劉亦菲", "張柏芝", "馮提模","陳一發兒")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)} } 運行結果： +----+------+
|name|length|
+----+------+
| 劉亦菲|? ? ?3|
| 張柏芝|? ? ?3|
| 馮提模|? ? ?3|
|陳一發兒|? ? ?4|
+----+------+

[劉亦菲,3]
[張柏芝,3]
[馮提模,3]
[陳一發兒,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("陳一發兒"); 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)); }}); /** * 使用動態編程方式，將RDD轉換為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); /** * 注冊臨時表 */ nameDF.registerTempTable("user"); /** * 根據UDF函數參數的個數來決定是實現哪一個UDF UDF1，UDF2, 表明包含幾個參數傳入 * UDF1<String, Integer> 表示 傳入參數 String 輸出參數為 Integer * call方法為 自定義的函數！ * DataTypes.IntegerType 必須與輸出參數的類型一致即 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 * 在臨時表user中 查找name StrLen(name) == name的長度 * StrLen(name) as length 表示將獲取到的name的長度 例如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(); } } 輸出結果：同上！
Java版本中，主要之一到UDFX 方法，以及傳入參數的個數類型，以及輸出類型，最終要的是文本最后的 DataTypes.IntegerType? 類型要與輸出類型相同！

總結

以上是生活随笔為你收集整理的Spark SQL 中UDF的讲解的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。