将案例类传递给 Spark UDF答案

【问题标题】：Pass case class to Spark UDF将案例类传递给 Spark UDF
【发布时间】：2018-02-02 12:06:55
【问题描述】：

我有一个 scala-2.11 函数，它根据提供的类类型从 Map 创建一个案例类。

def createCaseClass[T: TypeTag, A](someMap: Map[String, A]): T = {

    val rMirror = runtimeMirror(getClass.getClassLoader)
    val myClass = typeOf[T].typeSymbol.asClass
    val cMirror = rMirror.reflectClass(myClass)

    // The primary constructor is the first one
    val ctor = typeOf[T].decl(termNames.CONSTRUCTOR).asTerm.alternatives.head.asMethod
    val argList = ctor.paramLists.flatten.map(param => someMap(param.name.toString))

    cMirror.reflectConstructor(ctor)(argList: _*).asInstanceOf[T]
  }

我正在尝试在 spark 数据帧的上下文中使用它作为 UDF。但是，我不确定通过案例课程的最佳方式是什么。下面的方法似乎不起作用。

def myUDF[T: TypeTag] = udf { (inMap: Map[String, Long]) =>
    createCaseClass[T](inMap)
  }

我正在寻找这样的东西-

case class MyType(c1: String, c2: Long)

val myUDF = udf{(MyType, inMap) => createCaseClass[MyType](inMap)}

感谢您提出解决此问题的想法和建议。

【问题讨论】：

标签： scala apache-spark user-defined-functions case-class

【解决方案1】：

但是，我不确定通过案例类的最佳方法是什么

不能将案例类用作用户定义函数的参数。 SQL StructTypes 被映射到动态类型（因为缺少更好的词）Row 对象。

如果要对静态类型的对象进行操作，请使用静态类型的Dataset。

【讨论】：

【解决方案2】：

通过尝试和错误，我了解到存储在 Dataframe 或 Dataset 中的任何数据结构都在使用 org.apache.spark.sql.types

你可以看到：

df.schema.toString

Int、Double 等基本类型的存储方式如下：

StructField(fieldname,IntegerType,true),StructField(fieldname,DoubleType,true)

复杂类型比如case class被转换成嵌套类型的组合：

StructType(StructField(..),StructField(..),StructType(..))

示例代码：

case class range(min:Double,max:Double)
org.apache.spark.sql.Encoders.product[range].schema

//Output:
 org.apache.spark.sql.types.StructType = StructType(StructField(min,DoubleType,false), StructField(max,DoubleType,false))

这种情况下的 UDF 参数类型是 Row，或者当您存储一个案例类数组时是 Seq[Row]

基本的调试技术是打印到字符串：

 val myUdf = udf( (r:Row) =>   r.schema.toString )

然后，看到发生了：

df.take(1).foreach(println) //

【讨论】：