如何在Spark SQL中定义自定义类型的模式？

Question

以下示例代码尝试将一些案例对象放入数据框中.代码包括案例对象层次结构的定义和使用此特征的案例类:

import org.apache.spark.{SparkContext, SparkConf}
import org.apache.spark.sql.SQLContext

sealed trait Some
case object AType extends Some
case object BType extends Some

case class Data( name : String, t: Some)

object Example {
  def main(args: Array[String]) : Unit = {
    val conf = new SparkConf()
      .setAppName( "Example" )
      .setMaster( "local[*]")

    val sc = new SparkContext(conf)
    val sqlContext = new SQLContext(sc)

    import sqlContext.implicits._

    val df = sc.parallelize( Seq( Data( "a", AType), Data( "b", BType) ), 4).toDF()
    df.show()
  }
}    

执行代码时,我遗憾地遇到以下异常:

java.lang.UnsupportedOperationException: Schema for type Some is not supported

问题

• 是否有可能为某些类型添加或定义模式(此处为类型Some)？
• 是否存在另一种表示此类枚举的方法？
  • 我试图Enumeration直接使用,但也没有成功.(见下文)

代码Enumeration:

object Some extends Enumeration {
  type Some = Value
  val AType, BType = Value
}

提前致谢.我希望,最好的方法不是使用字符串.

Answer 1

Spark 2.0.0+:

UserDefinedType已经在Spark 2.0.0中被私有化,并且目前它没有Dataset友好的替代品.

请参阅:SPARK-14155(在Spark 2.0中隐藏UserDefinedType)

静态输入的大部分时间Dataset都可以作为替换.有一个待定的Jira SPARK-7768可以使目标版本2.4再次公开UDT API.

另请参见如何在数据集中存储自定义对象？

Spark <2.0.0

是否有可能为某些类型添加或定义模式(此处键入Some)？

我想答案取决于你需要多么糟糕.它看起来像是可以创建一个UserDefinedType但它需要访问,DeveloperApi并且不是很简单或记录良好.

import org.apache.spark.sql.types._

@SQLUserDefinedType(udt = classOf[SomeUDT])
sealed trait Some
case object AType extends Some
case object BType extends Some

class SomeUDT extends UserDefinedType[Some] {
  override def sqlType: DataType = IntegerType

  override def serialize(obj: Any) = {
    obj match {
      case AType => 0
      case BType => 1
    }
  }

  override def deserialize(datum: Any): Some = {
    datum match {
      case 0 => AType
      case 1 => BType
    }
  }

  override def userClass: Class[Some] = classOf[Some]
}

你或许应该重写hashCode,并equals为好.

它的PySpark对应物可能如下所示:

from enum import Enum, unique
from pyspark.sql.types import UserDefinedType, IntegerType

class SomeUDT(UserDefinedType):
    @classmethod
    def sqlType(self):
        return IntegerType()

    @classmethod
    def module(cls):
        return cls.__module__

    @classmethod 
    def scalaUDT(cls): # Required in Spark < 1.5
        return 'net.zero323.enum.SomeUDT'

    def serialize(self, obj):
        return obj.value

    def deserialize(self, datum):
        return {x.value: x for x in Some}[datum]

@unique
class Some(Enum):
    __UDT__ = SomeUDT()
    AType = 0
    BType = 1

在Spark <1.5中,Python UDT需要一个成对的Scala UDT,但它看起来不再是1.5中的情况.

对于简单的UDT,您可以使用简单类型(例如,IntegerType而不是整体Struct).