spark udf 与 dataframe连用

示例场景：

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

object Test {
  def main(args: Array[String]): Unit = {
    val spark = SparkSession.builder().master("local[*]").appName("function").getOrCreate()
    import spark.implicits._

    val df = spark.createDataframe(Seq(
      (1, "a"),
      (2, "??"),
      (3, "nullnull"))).toDF("Id", "x4")

    val p: Pattern = Pattern.compile("(?i)(\s|null|nan|\?)*")

    val nullTrans: String => String = (s:String) =>{  //定义函数，观察注意3个符号，=>的左右是函数参数与函数体，= 右边是函数，：后边是类型
      if(s == null || p.matcher(s).matches()){
        null
      }else{
        s
      }
    }

    val nullTransFunc = udf(nullTrans)//udf使用

    def nullTransData(data:Dataframe) ={
      var df = data
      val cols: Array[String] = df.columns
      val schema1 = df.schema
      cols.foreach{
        col =>{
          val dtype = schema1(col).dataType
          dtype match {
            case StringType =>
              df = df.withColumn(col,nullTransFunc(df(col)))//withColumn(列名1，函数名(df(列名2)))，而且列名2中的每个元素类型是函数
                                                            //的输入类型
            case _ =>
          }

        }

      }
     df
    }
    nullTransData(df).show()
    //+---+----+
    //| Id|  x4|
    //+---+----+
    //|  1|   a|
    //|  2|null|
    //|  3|null|
    //+---+----+

  }
}