최신 정보

이 답변은 상황이 지금보다 나은 있지만, 여전히 유효하고 유익한 내장 추가합니다 인코더 지원을 위해 2.2 / 2.3 이후 Set, Seq, Map, Date, Timestamp,와 BigDecimal. 케이스 클래스와 일반적인 스칼라 유형으로 만 유형을 작성하는 경우 암시 적 in으로 만 괜찮을 것입니다 SQLImplicits.

불행히도, 이것을 돕기 위해 사실상 아무것도 추가되지 않았습니다. 검색 @since 2.0.0에서 Encoders.scala또는 SQLImplicits.scala발견 것들이 대부분 기본 유형 (및 사례 클래스의 일부 조정)을 수행하는. 따라서 가장 먼저 할 말 : 현재 커스텀 클래스 인코더에 대한 실질적인 지원은 없습니다 . 그 길을 따라 가면 다음에 오는 것은 현재 우리가 처분 할 수있는 것을 감안할 때 우리가 기대할 수있는만큼 좋은 일을하는 몇 가지 트릭입니다. 선결제 면책 조항 : 이것은 완벽하게 작동하지 않으며 모든 제한 사항을 명확하고 선결하게하기 위해 최선을 다할 것입니다.

정확히 무엇이 문제입니까

데이터 세트를 만들려면 Spark는 일반적으로의 암시 적을 통해 자동으로 생성 SparkSession되거나 정적 메서드를 호출하여 명시 적으로 생성 할 수 있는 인코더 (T 유형의 JVM 객체를 내부 Spark SQL 표현으로 변환하거나 내부 Spark SQL 표현으로 변환)가 필요 합니다. 에 Encoders“(의 문서에서createDataset 가져온 ). 인코더는 형태 걸릴 것입니다 당신이 인코딩하는 유형입니다. 최초의 제안에 추가한다 (을주는 이러한 암시 인코더) 및 제 제안을 명시 적으로 사용하여 암시 적 인코더에 전달하는 이 인코더 관련 기능의 세트.Encoder[T]Timport spark.implicits._

정규 수업에 사용할 수있는 인코더가 없으므로

import spark.implicits._
class MyObj(val i: Int)
// ...
val d = spark.createDataset(Seq(new MyObj(1),new MyObj(2),new MyObj(3)))

다음과 같은 암시 적 관련 컴파일 시간 오류가 발생합니다.

데이터 세트에 저장된 유형의 인코더를 찾을 수 없습니다. sqlContext.implicits를 가져 오면 기본 유형 (Int, String 등) 및 제품 유형 (케이스 클래스)이 지원됩니다. 다른 유형의 직렬화 지원은 다음 릴리스에서 추가 될 예정입니다.

그러나 확장 한 일부 클래스에서 위의 오류를 얻는 데 사용한 유형을 래핑 Product하면 오류가 혼란스럽게 런타임에 지연되므로

import spark.implicits._
case class Wrap[T](unwrap: T)
class MyObj(val i: Int)
// ...
val d = spark.createDataset(Seq(Wrap(new MyObj(1)),Wrap(new MyObj(2)),Wrap(new MyObj(3))))

컴파일은 잘되지만 런타임에 실패합니다.

java.lang.UnsupportedOperationException : MyObj에 대한 인코더가 없습니다.

그 이유는 Spark가 암시 적으로 생성하는 인코더는 실제로 런타임에만 (스칼라 관계를 통해) 만들어지기 때문입니다. 이 경우 컴파일 타임에 모든 Spark 검사는 가장 바깥 쪽 클래스가 확장되고 Product(모든 경우 클래스가 수행하는) 런타임에서만 수행해야 할 작업을 여전히 알지 못한다는 것을 인식합니다 MyObj(내가 만들려고해도 동일한 문제가 발생 함) a Dataset[(Int,MyObj)]-Spark는 런타임시 barf on까지 기다립니다 MyObj. 다음은 해결해야 할 핵심 문제입니다.

• Product런타임에 항상 충돌하더라도 컴파일 을 확장하는 일부 클래스
• 중첩 된 유형에 대한 정의 인코더에 전달하는 방법이 없습니다 (나는 불꽃에게 단지에 대한 인코더 공급의 방법이 없습니다 MyObj그 다음 어떻게 인코딩을 알고 같은 Wrap[MyObj]또는 (Int,MyObj)).

그냥 사용 kryo

모두가 제안하는 해결책은 kryo인코더 를 사용하는 것 입니다.

import spark.implicits._
class MyObj(val i: Int)
implicit val myObjEncoder = org.apache.spark.sql.Encoders.kryo[MyObj]
// ...
val d = spark.createDataset(Seq(new MyObj(1),new MyObj(2),new MyObj(3)))

그래도 꽤 지루합니다. 특히 코드가 모든 종류의 데이터 세트, 조인, 그룹화 등을 조작하는 경우 많은 암시 적 요소가 쌓이게됩니다. 그렇다면이 모든 것을 자동으로 수행하는 암시적인 이유는 무엇입니까?

import scala.reflect.ClassTag
implicit def kryoEncoder[A](implicit ct: ClassTag[A]) =
  org.apache.spark.sql.Encoders.kryo[A](ct)

그리고 지금, 내가 원하는 거의 모든 것을 할 수있는 것처럼 보입니다 (아래 예 는 자동으로 가져 오는 spark-shell곳에서는 작동하지 않습니다 spark.implicits._)

class MyObj(val i: Int)

val d1 = spark.createDataset(Seq(new MyObj(1),new MyObj(2),new MyObj(3)))
val d2 = d1.map(d => (d.i+1,d)).alias("d2") // mapping works fine and ..
val d3 = d1.map(d => (d.i,  d)).alias("d3") // .. deals with the new type
val d4 = d2.joinWith(d3, $"d2._1" === $"d3._1") // Boom!

아니면 거의. 문제는 kryo리드를 사용 하면 데이터 세트의 모든 행을 플랫 이진 객체로 저장하는 것입니다. 를 들어 map, filter, foreach충분하다,하지만 같은 작업을 위해 join, 스파크 정말이 컬럼으로 분리 될 필요합니다. d2또는 의 스키마를 검사하면 d3이진 열이 하나만있는 것을 알 수 있습니다.

d2.printSchema
// root
//  |-- value: binary (nullable = true)

튜플 용 부분 솔루션

따라서 Scala ( 6.26.3 Overloading Resolution ) 의 암시 적 마술을 사용하여 최소한 튜플에 대해 가능한 한 잘 작동하고 기존 암시 적과 잘 작동 하는 일련의 암시 적을 만들 수 있습니다.

import org.apache.spark.sql.{Encoder,Encoders}
import scala.reflect.ClassTag
import spark.implicits._  // we can still take advantage of all the old implicits

implicit def single[A](implicit c: ClassTag[A]): Encoder[A] = Encoders.kryo[A](c)

implicit def tuple2[A1, A2](
  implicit e1: Encoder[A1],
           e2: Encoder[A2]
): Encoder[(A1,A2)] = Encoders.tuple[A1,A2](e1, e2)

implicit def tuple3[A1, A2, A3](
  implicit e1: Encoder[A1],
           e2: Encoder[A2],
           e3: Encoder[A3]
): Encoder[(A1,A2,A3)] = Encoders.tuple[A1,A2,A3](e1, e2, e3)

// ... you can keep making these

그런 다음 이러한 암시로 무장하여 일부 열 이름을 바꾸더라도 위의 예제를 작동시킬 수 있습니다.

class MyObj(val i: Int)

val d1 = spark.createDataset(Seq(new MyObj(1),new MyObj(2),new MyObj(3)))
val d2 = d1.map(d => (d.i+1,d)).toDF("_1","_2").as[(Int,MyObj)].alias("d2")
val d3 = d1.map(d => (d.i  ,d)).toDF("_1","_2").as[(Int,MyObj)].alias("d3")
val d4 = d2.joinWith(d3, $"d2._1" === $"d3._1")

나는 아직 (예상 튜플 이름을 얻는 방법을 파악하지 않은 _1, _2그 이름을 변경하지 않고 기본적으로, …) – 다른 사람이 놀러하고자하는 경우, 이 이름이 어디 "value"도입 도착하고 이 곳 튜플 이름이 일반적으로 추가됩니다. 그러나 요점은 이제 멋진 구조화 된 스키마가 있다는 것입니다.

d4.printSchema
// root
//  |-- _1: struct (nullable = false)
//  |    |-- _1: integer (nullable = true)
//  |    |-- _2: binary (nullable = true)
//  |-- _2: struct (nullable = false)
//  |    |-- _1: integer (nullable = true)
//  |    |-- _2: binary (nullable = true)

요약하면이 해결 방법은 다음과 같습니다.

• 튜플에 대해 별도의 열을 얻을 수 있습니다 (따라서 튜플에 다시 가입 할 수 있습니다).
• 우리는 다시 암시 적에 의존 할 수 있습니다 (따라서 kryo모든 곳을 지나갈 필요가 없습니다 )
• 거의 전적으로 이전 버전과 호환됩니다 import spark.implicits._(일부 이름 변경 포함)
• 않습니다 하지 우리가에 가입하자 kyro연재 진 열, 사람들이있을 수 있습니다 필드에 혼자하자
• 일부 튜플 열의 이름을 “값”으로 변경하면 불쾌한 부작용이 있습니다 (필요한 경우, 변환 .toDF, 새 열 이름을 지정하고 데이터 집합으로 다시 변환하여 실행 취소 할 수 있음). 조인을 통해 스키마 이름이 유지되는 것처럼 보입니다. 가장 필요한 곳).

일반적인 수업을위한 부분 솔루션

이것은 덜 유쾌하며 좋은 해결책이 없습니다. 그러나 이제 우리는 위의 튜플 솔루션을 얻었으므로 더 복잡한 클래스를 튜플로 변환 할 수 있기 때문에 다른 답변의 암시 적 변환 솔루션도 약간 덜 고통 스럽습니다. 그런 다음 데이터 집합을 만든 후에는 데이터 프레임 접근 방식을 사용하여 열의 이름을 바꿀 수 있습니다. 모든 것이 잘 진행된다면, 이제 수업 분야에서 조인을 수행 할 수 있기 때문에 이것은 실제로 개선 된 것입니다. 플랫 바이너리 kryo시리얼 라이저를 하나만 사용 했다면 불가능했을 것입니다.

여기에 모든 것을 조금을 수행하는 예입니다 : 나는 수업이 MyObj유형의 필드가 Int, java.util.UUID하고 Set[String]. 첫 번째는 스스로 처리합니다. 두 번째로, 직렬화 할 수는 있지만 ( 보통 내가 참여하고 싶은 kryo것이기 때문에) 더 유용합니다 . 세 번째는 실제로 이진 열에 속합니다.StringUUID

class MyObj(val i: Int, val u: java.util.UUID, val s: Set[String])

// alias for the type to convert to and from
type MyObjEncoded = (Int, String, Set[String])

// implicit conversions
implicit def toEncoded(o: MyObj): MyObjEncoded = (o.i, o.u.toString, o.s)
implicit def fromEncoded(e: MyObjEncoded): MyObj =
  new MyObj(e._1, java.util.UUID.fromString(e._2), e._3)

이제이 메커니즘을 사용하여 멋진 스키마로 데이터 세트를 만들 수 있습니다.

val d = spark.createDataset(Seq[MyObjEncoded](
  new MyObj(1, java.util.UUID.randomUUID, Set("foo")),
  new MyObj(2, java.util.UUID.randomUUID, Set("bar"))
)).toDF("i","u","s").as[MyObjEncoded]

그리고 스키마는 올바른 이름과 내가 조인 할 수있는 처음 두 가지가있는 열을 보여줍니다.

d.printSchema
// root
//  |-- i: integer (nullable = false)
//  |-- u: string (nullable = true)
//  |-- s: binary (nullable = true)

답변

trait CustomEnum { def value:String }
case object Foo extends CustomEnum  { val value = "F" }
case object Bar extends CustomEnum  { val value = "B" }
object CustomEnum {
  def fromString(str:String) = Seq(Foo, Bar).find(_.value == str).get
}

// First define a UDT class for it:
class CustomEnumUDT extends UserDefinedType[CustomEnum] {
  override def sqlType: DataType = org.apache.spark.sql.types.StringType
  override def serialize(obj: CustomEnum): Any = org.apache.spark.unsafe.types.UTF8String.fromString(obj.value)
  // Note that this will be a UTF8String type
  override def deserialize(datum: Any): CustomEnum = CustomEnum.fromString(datum.toString)
  override def userClass: Class[CustomEnum] = classOf[CustomEnum]
}

// Then Register the UDT Class!
// NOTE: you have to put this file into the org.apache.spark package!
UDTRegistration.register(classOf[CustomEnum].getName, classOf[CustomEnumUDT].getName)

case class UsingCustomEnum(id:Int, en:CustomEnum)

val seq = Seq(
  UsingCustomEnum(1, Foo),
  UsingCustomEnum(2, Bar),
  UsingCustomEnum(3, Foo)
).toDS()
seq.filter(_.en == Foo).show()
println(seq.collect())

trait CustomPoly
case class FooPoly(id:Int) extends CustomPoly
case class BarPoly(value:String, secondValue:Long) extends CustomPoly

case class UsingPoly(id:Int, poly:CustomPoly)

Seq(
  UsingPoly(1, new FooPoly(1)),
  UsingPoly(2, new BarPoly("Blah", 123)),
  UsingPoly(3, new FooPoly(1))
).toDS

polySeq.filter(_.poly match {
  case FooPoly(value) => value == 1
  case _ => false
}).show()

class CustomPolyUDT extends UserDefinedType[CustomPoly] {
  val kryo = new Kryo()

  override def sqlType: DataType = org.apache.spark.sql.types.BinaryType
  override def serialize(obj: CustomPoly): Any = {
    val bos = new ByteArrayOutputStream()
    val oos = new ObjectOutputStream(bos)
    oos.writeObject(obj)

    bos.toByteArray
  }
  override def deserialize(datum: Any): CustomPoly = {
    val bis = new ByteArrayInputStream(datum.asInstanceOf[Array[Byte]])
    val ois = new ObjectInputStream(bis)
    val obj = ois.readObject()
    obj.asInstanceOf[CustomPoly]
  }

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

// NOTE: The file you do this in has to be inside of the org.apache.spark package!
UDTRegistration.register(classOf[CustomPoly].getName, classOf[CustomPolyUDT].getName)

// As shown above:
case class UsingPoly(id:Int, poly:CustomPoly)

Seq(
  UsingPoly(1, new FooPoly(1)),
  UsingPoly(2, new BarPoly("Blah", 123)),
  UsingPoly(3, new FooPoly(1))
).toDS

polySeq.filter(_.poly match {
  case FooPoly(value) => value == 1
  case _ => false
}).show()

scala> import spark.implicits._
import spark.implicits._

scala> import org.apache.spark.sql.Encoders
import org.apache.spark.sql.Encoders

scala> case class NormalPerson(name: String, age: Int) {
 |   def aboutMe = s"I am ${name}. I am ${age} years old."
 | }
defined class NormalPerson

scala> case class ReversePerson(name: Int, age: String) {
 |   def aboutMe = s"I am ${name}. I am ${age} years old."
 | }
defined class ReversePerson

scala> val normalPersons = Seq(
 |   NormalPerson("Superman", 25),
 |   NormalPerson("Spiderman", 17),
 |   NormalPerson("Ironman", 29)
 | )
normalPersons: Seq[NormalPerson] = List(NormalPerson(Superman,25), NormalPerson(Spiderman,17), NormalPerson(Ironman,29))

scala> val ds1 = sc.parallelize(normalPersons).toDS
ds1: org.apache.spark.sql.Dataset[NormalPerson] = [name: string, age: int]

scala> val ds2 = ds1.map(np => ReversePerson(np.age, np.name))
ds2: org.apache.spark.sql.Dataset[ReversePerson] = [name: int, age: string]

scala> ds1.show()
+---------+---+
|     name|age|
+---------+---+
| Superman| 25|
|Spiderman| 17|
|  Ironman| 29|
+---------+---+

scala> ds2.show()
+----+---------+
|name|      age|
+----+---------+
|  25| Superman|
|  17|Spiderman|
|  29|  Ironman|
+----+---------+

scala> ds1.foreach(p => println(p.aboutMe))
I am Ironman. I am 29 years old.
I am Superman. I am 25 years old.
I am Spiderman. I am 17 years old.

scala> val ds2 = ds1.map(np => ReversePerson(np.age, np.name))
ds2: org.apache.spark.sql.Dataset[ReversePerson] = [name: int, age: string]

scala> ds2.foreach(p => println(p.aboutMe))
I am 17. I am Spiderman years old.
I am 25. I am Superman years old.
I am 29. I am Ironman years old.

// Provide Encoders

scala> implicit val normalPersonKryoEncoder = Encoders.kryo[NormalPerson]
normalPersonKryoEncoder: org.apache.spark.sql.Encoder[NormalPerson] = class[value[0]: binary]

scala> implicit val reversePersonKryoEncoder = Encoders.kryo[ReversePerson]
reversePersonKryoEncoder: org.apache.spark.sql.Encoder[ReversePerson] = class[value[0]: binary]

// Ecoders will be used since they are now present in Scope

scala> val ds3 = sc.parallelize(normalPersons).toDS
ds3: org.apache.spark.sql.Dataset[NormalPerson] = [value: binary]

scala> val ds4 = ds3.map(np => ReversePerson(np.age, np.name))
ds4: org.apache.spark.sql.Dataset[ReversePerson] = [value: binary]

// now all our persons show up as binary values
scala> ds3.show()
+--------------------+
|               value|
+--------------------+
|[01 00 24 6C 69 6...|
|[01 00 24 6C 69 6...|
|[01 00 24 6C 69 6...|
+--------------------+

scala> ds4.show()
+--------------------+
|               value|
+--------------------+
|[01 00 24 6C 69 6...|
|[01 00 24 6C 69 6...|
|[01 00 24 6C 69 6...|
+--------------------+

// Our instances still work as expected    

scala> ds3.foreach(p => println(p.aboutMe))
I am Ironman. I am 29 years old.
I am Spiderman. I am 17 years old.
I am Superman. I am 25 years old.

scala> ds4.foreach(p => println(p.aboutMe))
I am 25. I am Superman years old.
I am 29. I am Ironman years old.
I am 17. I am Spiderman years old.

import spark.sqlContext.implicits._
import org.apache.spark.sql.Encoders
implicit val encoder = Encoders.bean[MyClasss](classOf[MyClass])

dataFrame.as[MyClass]

public class Fruit implements Serializable {
    private String name  = "default-fruit";
    private String color = "default-color";

    // AllArgsConstructor
    public Fruit(String name, String color) {
        this.name  = name;
        this.color = color;
    }

    // NoArgsConstructor
    public Fruit() {
        this("default-fruit", "default-color");
    }

    // ...create getters and setters for above fields
    // you figure it out
}

SparkSession spark = SparkSession.builder().getOrCreate();
JavaSparkContext jsc = new JavaSparkContext();

List<Fruit> fruitList = ImmutableList.of(
    new Fruit("apple", "red"),
    new Fruit("orange", "orange"),
    new Fruit("grape", "purple"));
JavaRDD<Fruit> fruitJavaRDD = jsc.parallelize(fruitList);


RDD<Fruit> fruitRDD = fruitJavaRDD.rdd();
Encoder<Fruit> fruitBean = Encoders.bean(Fruit.class);
Dataset<Fruit> fruitDataset = spark.createDataset(rdd, bean);