Как определить схему для настраиваемого типа в Spark SQL?

Следующий пример кода пытается помещать некоторые объекты case в фреймворк данных. Код включает определение иерархии объекта case и класс case с использованием этого признака:

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
}

Спасибо заранее. Я надеюсь, что лучший подход - не использовать строки вместо этого.

Ответы

Ответ 1

Spark 2.0. 0+:

UserDefinedType был UserDefinedType в Spark 2.0.0, и на данный момент он не поддерживает замену Dataset.

См. SPARK-14155 (Hide UserDefinedType в Spark 2.0)

В большинстве случаев статически типизированный Dataset может служить заменой. Ожидается Jira SPARK-7768, чтобы снова открыть UDT API с целевой версией 2.4.

См. Также Как сохранить пользовательские объекты в наборе данных?

Искры <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).