在spark中使用sql時一些功能需要自定義方法實現(xiàn)，這時候就可以使用UDF功能來實現(xiàn)

多參數(shù)支持

UDF不支持參數(shù)*的方式輸入多個參數(shù)，例如String*，不過可以使用array來解決這個問題。

定義udf方法，此處功能是將多個字段合并為一個字段

def allInOne(seq: Seq[Any], sep: String): String = seq.mkString(sep)

在sql中使用

sqlContext.udf.register("allInOne", allInOne _)

//將col1,col2,col3三個字段合并，使用','分割
val sql =
    """
      |select allInOne(array(col1,col2,col3),",") as col
      |from tableName
    """.stripMargin
sqlContext.sql(sql).show()

在DataFrame中使用

import org.apache.spark.sql.functions.{udf，array,lit}
val myFunc = udf(allInOne _)
val cols = array("col1","col2","col3")
val sep = lit(",")
df.select(myFunc(cols,sep).alias("col")).show()

一些簡單的例子

1.個數(shù)統(tǒng)計

表結構如下，統(tǒng)計出每個人的愛好個數(shù)

# 將某個字段中逗號分隔的數(shù)量統(tǒng)計出來
sqlContext.udf.register("hobby_num", (s: String) => s.split(',').size)
sqlContext.sql("select *,hobby_num(hobbies) as hobby_num from table")

結果

2.空值填補

表結構如下

# 填補第一個字段的空值
sqlContext.udf.register("combine", (s1: String,s2: String)=> {if(s1 == null) s2 else s1})
sqlContext.sql("select combine(A,B) as A from table")

結果

3. 類型轉化

類型轉化,將 String 轉化為 Int

sqlContext.udf.register("str2Int", (s: String) => s.toInt)

或者直接使用cast

sqlContext.sql("select cast(a AS Int) from table")

4. 綜合運用

原始數(shù)據(jù)，ID(用戶名)，loginIP(帳號登錄的ip地址)

計算每個用戶在哪些ip登錄過，并統(tǒng)計數(shù)量

//統(tǒng)計數(shù)量
sqlContext.udf.register("list_size", (s: String) => s.split(',').size)
val sql =
    """select ID,ip_list,list_size(ip_list) as loginIP_num
      |from (select ID,concat_ws(',',collect_set(loginIP)) as ip_list from table)
    """.stripMargin
sqlContext.sql(sql)