spark udf 和 dataframe连用

示例场景：

将dataframe中非法字符替代为 null  类型，方便后续空值填充

1.  
   object Test {
2.  
   def main(args: Array[String]): Unit = {
3.  
   val spark = SparkSession.builder().master("local[*]").appName("function").getOrCreate()
4.  
   import spark.implicits._
5.  
    
6.  
   val df = spark.createDataFrame(Seq(
7.  
   (1, "a"),
8.  
   (2, "??"),
9.  
   (3, "nullnull"))).toDF("Id", "x4")
10.  
     
11.  
    val p: Pattern = Pattern.compile("(?i)(\\s|null|nan|\\?)*")
12.  
     
13.  
    val nullTrans: String => String = (s:String) =>{ //定义函数，观察注意3个符号，=>的左右是函数参数与函数体，= 右边是函数，：后边是类型
14.  
    if(s == null || p.matcher(s).matches()){
15.  
    null
16.  
    }else{
17.  
    s
18.  
    }
19.  
    }
20.  
     
21.  
    val nullTransFunc = udf(nullTrans)//udf使用
22.  
     
23.  
    def nullTransData(data:DataFrame) ={
24.  
    var df = data
25.  
    val cols: Array[String] = df.columns
26.  
    val schema1 = df.schema
27.  
    cols.foreach{
28.  
    col =>{
29.  
    val dtype = schema1(col).dataType
30.  
    dtype match {
31.  
    case StringType =>
32.  
    df = df.withColumn(col,nullTransFunc(df(col)))//withColumn(列名1，函数名(df(列名2)))，而且列名2中的每个元素类型是函数
33.  
    //的输入类型
34.  
    case _ =>
35.  
    }
36.  
     
37.  
    }
38.  
     
39.  
    }
40.  
    df
41.  
    }
42.  
    nullTransData(df).show()
43.  
    // --- ----
44.  
    //| Id| x4|
45.  
    // --- ----
46.  
    //| 1| a|
47.  
    //| 2|null|
48.  
    //| 3|null|
49.  
    // --- ----
50.  
     
51.  
    }
52.  
    }

这篇好文章是转载于：学新通技术网