Spark Struct structfield 名称在 UDF 中发生变化

Question

我正在尝试将 spark 中的结构传递给 udf。它正在更改字段名称并重命名为列位置。我如何解决它？

object TestCSV {

          def main(args: Array[String]) {

            val conf = new SparkConf().setAppName("localTest").setMaster("local")
            val sc = new SparkContext(conf)
            val sqlContext = new SQLContext(sc)


            val inputData = sqlContext.read.format("com.databricks.spark.csv")
                  .option("delimiter","|")
                  .option("header", "true")
                  .load("test.csv")


            inputData.printSchema()

            inputData.show()

            val groupedData = inputData.withColumn("name",struct(inputData("firstname"),inputData("lastname")))

            val udfApply = groupedData.withColumn("newName",processName(groupedData("name")))

           udfApply.show()
          }



             def processName = udf((input:Row) =>{

                println(input)
                println(input.schema)

                Map("firstName" -> input.getAs[String]("firstname"), "lastName" -> input.getAs[String]("lastname"))

              })

        }

输出：

 root
 |-- id: string (nullable = true)
 |-- firstname: string (nullable = true)
 |-- lastname: string (nullable = true)

 +---+---------+--------+
 | id|firstname|lastname|
 +---+---------+--------+
 |  1|     jack| reacher|
 |  2|     john|     Doe|
 +---+---------+--------+

错误：

[jack,reacher] StructType(StructField(i[1],StringType,true), > StructField(i[2],StringType,true)) 17/03/08 09:45:35 ERROR Executor: Exception in task 0.0 in stage 2.0 (TID 2) java.lang.IllegalArgumentException: Field "firstname" does not exist.

Answer 1

你遇到的事情真奇怪。玩了一会儿后，我终于发现这可能与优化器引擎的问题有关。看来问题不是UDF而是struct函数。

当我 cache groupedData 时，我让它工作 (Spark 1.6.3)，没有缓存我得到你报告的异常：

import org.apache.spark.sql.Row
import org.apache.spark.sql.hive.HiveContext
import org.apache.spark.{SparkConf, SparkContext}


object Demo {

  def main(args: Array[String]): Unit = {

    val sc = new SparkContext(new SparkConf().setAppName("Demo").setMaster("local[1]"))
    val sqlContext = new HiveContext(sc)
    import sqlContext.implicits._
    import org.apache.spark.sql.functions._


    def processName = udf((input: Row) => {
      Map("firstName" -> input.getAs[String]("firstname"), "lastName" -> input.getAs[String]("lastname"))
    })


    val inputData =
      sc.parallelize(
        Seq(("1", "Kevin", "Costner"))
      ).toDF("id", "firstname", "lastname")


    val groupedData = inputData.withColumn("name", struct(inputData("firstname"), inputData("lastname")))
      .cache() // does not work without cache

    val udfApply = groupedData.withColumn("newName", processName(groupedData("name")))
    udfApply.show()
  }
}

或者你可以使用 RDD API 来创建你的结构，但这不是很好：

case class Name(firstname:String,lastname:String) // define outside main

val groupedData = inputData.rdd
    .map{r =>
        (r.getAs[String]("id"),
          Name(
            r.getAs[String]("firstname"),
            r.getAs[String]("lastname")
          )
        )
    }
   .toDF("id","name")

Spark Struct structfield 名称在 UDF 中发生变化

Spark Struct structfield names getting changed in UDF

apache-spark

struct

udf