De acuerdo con la introducción de conjuntos de datos de Spark :

Mientras esperamos Spark 2.0, planeamos algunas mejoras interesantes para los conjuntos de datos, específicamente: ... Codificadores personalizados: si bien actualmente generamos codificadores automáticamente para una amplia variedad de tipos, nos gustaría abrir una API para objetos personalizados.

e intenta almacenar el tipo personalizado en un Datasetplomo para el siguiente error como:

No se puede encontrar el codificador para el tipo almacenado en un conjunto de datos. Los tipos primitivos (Int, String, etc.) y los tipos de productos (clases de caso) son compatibles con la importación de sqlContext.implicits. En futuras versiones se agregará soporte para serializar otros tipos.

o:

Java.lang.UnsupportedOperationException: no se ha encontrado ningún codificador para ...

¿Hay alguna solución existente?

Tenga en cuenta que esta pregunta existe solo como un punto de entrada para una respuesta Wiki de la comunidad. Siéntase libre de actualizar / mejorar tanto la pregunta como la respuesta.

Actualizar

Esta respuesta sigue siendo válida e informativo, aunque las cosas están ahora mejor desde 2.2 / 2.3, que añade soporte integrado para el codificador Set, Seq, Map, Date, Timestamp, y BigDecimal. Si te limitas a crear tipos con solo clases de casos y los tipos habituales de Scala, deberías estar bien solo con lo implícito SQLImplicits.

Desafortunadamente, prácticamente no se ha agregado nada para ayudar con esto. Buscar @since 2.0.0en Encoders.scalao SQLImplicits.scalaencuentra cosas principalmente relacionadas con tipos primitivos (y algunos ajustes de clases de casos). Entonces, lo primero que hay que decir: actualmente no hay un soporte realmente bueno para codificadores de clase personalizados . Con eso fuera del camino, lo que sigue son algunos trucos que hacen un trabajo tan bueno como podemos esperar, dado lo que tenemos actualmente a nuestra disposición. Como descargo de responsabilidad por adelantado: esto no funcionará perfectamente y haré todo lo posible para que todas las limitaciones sean claras y directas.

Cuál es exactamente el problema

Cuando desee crear un conjunto de datos, Spark "requiere un codificador (para convertir un objeto JVM de tipo T ay desde la representación interna de Spark SQL) que generalmente se crea automáticamente mediante implicits desde a SparkSession, o puede crearse explícitamente llamando a métodos estáticos on Encoders"(tomado de los documentos encreateDataset ). Un codificador tomará la forma Encoder[T]donde Testá el tipo que está codificando. La primera sugerencia es agregar import spark.implicits._(que le proporciona estos codificadores implícitos) y la segunda sugerencia es pasar explícitamente el codificador implícito utilizando este conjunto de funciones relacionadas con el codificador.

No hay codificador disponible para clases regulares, por lo que

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

le dará el siguiente error de tiempo de compilación relacionado implícito:

No se puede encontrar el codificador para el tipo almacenado en un conjunto de datos. Los tipos primitivos (Int, String, etc.) y los tipos de productos (clases de caso) son compatibles con la importación de sqlContext.implicits. En futuras versiones se agregará soporte para serializar otros tipos.

Sin embargo, si ajusta el tipo que acaba de usar para obtener el error anterior en alguna clase que se extiende Product, el error se retrasa confusamente en tiempo de ejecución, por lo que

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))))

Compila bien, pero falla en tiempo de ejecución con

java.lang.UnsupportedOperationException: no se ha encontrado ningún codificador para MyObj

La razón de esto es que los codificadores que Spark crea con los implicits en realidad solo se hacen en tiempo de ejecución (a través de la reescalación de escala). En este caso, todas las comprobaciones de Spark en tiempo de compilación es que la clase más externa se extiende Product(lo que hacen todas las clases de casos), y solo se da cuenta en el tiempo de ejecución de que todavía no sabe qué hacer MyObj(el mismo problema ocurre si intento hacer a Dataset[(Int,MyObj)]- Spark espera hasta el tiempo de ejecución para vomitar MyObj). Estos son problemas centrales que necesitan urgentemente ser solucionados:

• algunas clases que amplían la Productcompilación a pesar de fallar siempre en tiempo de ejecución y
• no hay forma de pasar codificadores personalizados para tipos anidados (no tengo forma de alimentar a Spark con un codificador solo para MyObjque sepa cómo codificar Wrap[MyObj]o (Int,MyObj)).

Solo usa kryo

La solución que todos sugieren es usar el kryocodificador.

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)))

Sin embargo, esto se vuelve bastante tedioso rápidamente. Especialmente si su código está manipulando todo tipo de conjuntos de datos, uniéndose, agrupando, etc. Usted termina acumulando un montón de problemas adicionales. Entonces, ¿por qué no hacer un implícito que hace todo esto automáticamente?

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

Y ahora, parece que puedo hacer casi cualquier cosa que quiera (el siguiente ejemplo no funcionará en el lugar spark-shelldonde spark.implicits._se importa automáticamente)

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!

O casi El problema es que el uso de kryoleads lleva a Spark a almacenar cada fila del conjunto de datos como un objeto binario plano. Para map, filter, foreachque es suficiente, pero para operaciones como join, Spark realmente las necesita para ser separados en columnas. Al inspeccionar el esquema para d2o d3, verá que solo hay una columna binaria:

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

Solución parcial para tuplas.

Entonces, usando la magia de las implicidades en Scala (más en 6.26.3 Resolución de sobrecarga ), puedo hacer una serie de implicidades que harán el mejor trabajo posible, al menos para las tuplas, y funcionarán bien con las implicidades existentes:

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

Luego, armado con estas dificultades, puedo hacer que mi ejemplo anterior funcione, aunque con un cambio de nombre de columna

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")

Todavía no he encontrado la manera de obtener los nombres de tupla esperados ( _1, _2, ...) de forma predeterminada sin cambiar el nombre de ellos - si alguien quiere jugar un poco con esto, este es el lugar donde el nombre "value"se introduce y esto es donde la tupla generalmente se agregan nombres. Sin embargo, el punto clave es que ahora tengo un buen esquema estructurado:

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)

Entonces, en resumen, esta solución alternativa:

• nos permite obtener columnas separadas para las tuplas (para que podamos unirnos nuevamente en las tuplas, ¡sí!)
• nuevamente podemos confiar en las implicidades (así que no hay necesidad de pasar por kryotodas partes)
• es casi totalmente compatible con versiones anteriores import spark.implicits._(con algunos cambios de nombre)
• no no vamos a unir en las kyrocolumnas binarios serializados, y mucho menos en los campos de los que pueden tener
• tiene el efecto secundario desagradable de cambiar el nombre de algunas de las columnas de tupla a "valor" (si es necesario, esto se puede deshacer mediante la conversión .toDF, la especificación de nuevos nombres de columna y la conversión de nuevo a un conjunto de datos, y los nombres de esquema parecen conservarse mediante combinaciones) , donde más se necesitan).

Solución parcial para clases en general.

Este es menos agradable y no tiene una buena solución. Sin embargo, ahora que tenemos la solución de tupla anterior, tengo el presentimiento de que la solución de conversión implícita de otra respuesta también será un poco menos dolorosa ya que puede convertir sus clases más complejas en tuplas. Luego, después de crear el conjunto de datos, probablemente cambie el nombre de las columnas utilizando el enfoque de marco de datos. Si todo va bien, esto es realmente una mejora ya que ahora puedo realizar uniones en los campos de mis clases. Si hubiera utilizado un kryoserializador binario plano, eso no hubiera sido posible.

Aquí está un ejemplo que hace un poco de todo: Tengo una clase MyObjque tiene campos de tipos Int, java.util.UUIDy Set[String]. El primero se cuida solo. El segundo, aunque podría serializar usando kryo, sería más útil si se almacena como un String(ya que los UUIDs suelen ser algo contra lo que me quiero unir). El tercero realmente solo pertenece a una columna binaria.

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)

Ahora, puedo crear un conjunto de datos con un buen esquema usando esta maquinaria:

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]

Y el esquema me muestra columnas I con los nombres correctos y con las dos primeras cosas con las que me puedo unir.

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

1. Usando codificadores genéricos.

   Hay dos codificadores genéricos disponibles por ahora kryoy javaSerializationdonde el último se describe explícitamente como:

   extremadamente ineficiente y solo debe usarse como último recurso.

   Asumiendo la siguiente clase

   class Bar(i: Int) {
     override def toString = s"bar $i"
     def bar = i
   }

   Puede usar estos codificadores agregando un codificador implícito:

   object BarEncoders {
     implicit def barEncoder: org.apache.spark.sql.Encoder[Bar] = 
     org.apache.spark.sql.Encoders.kryo[Bar]
   }

   que se pueden usar juntos de la siguiente manera:

   object Main {
     def main(args: Array[String]) {
       val sc = new SparkContext("local",  "test", new SparkConf())
       val sqlContext = new SQLContext(sc)
       import sqlContext.implicits._
       import BarEncoders._
   
       val ds = Seq(new Bar(1)).toDS
       ds.show
   
       sc.stop()
     }
   }

   Almacena objetos como binarycolumna, por lo que cuando se convierteDataFrame usted obtiene el siguiente esquema:

   root
    |-- value: binary (nullable = true)

   También es posible codificar tuplas usando kryo codificador para un campo específico:

   val longBarEncoder = Encoders.tuple(Encoders.scalaLong, Encoders.kryo[Bar])
   
   spark.createDataset(Seq((1L, new Bar(1))))(longBarEncoder)
   // org.apache.spark.sql.Dataset[(Long, Bar)] = [_1: bigint, _2: binary]

   Tenga en cuenta que no dependemos de codificadores implícitos aquí, sino que pasamos el codificador explícitamente, por lo que lo más probable es que esto no funcione con el toDSmétodo.

2. Usando conversiones implícitas:

   Proporcione conversiones implícitas entre la representación que se puede codificar y la clase personalizada, por ejemplo:

   object BarConversions {
     implicit def toInt(bar: Bar): Int = bar.bar
     implicit def toBar(i: Int): Bar = new Bar(i)
   }
   
   object Main {
     def main(args: Array[String]) {
       val sc = new SparkContext("local",  "test", new SparkConf())
       val sqlContext = new SQLContext(sc)
       import sqlContext.implicits._
       import BarConversions._
   
       type EncodedBar = Int
   
       val bars: RDD[EncodedBar]  = sc.parallelize(Seq(new Bar(1)))
       val barsDS = bars.toDS
   
       barsDS.show
       barsDS.map(_.bar).show
   
       sc.stop()
     }
   }

Preguntas relacionadas:

• ¿Cómo crear un codificador para el constructor de tipo Opción, por ejemplo, Opción [Int]?

Puede usar UDTRegistration y luego Case Classes, Tuples, etc. ¡todo funciona correctamente con su tipo definido por el usuario!

Digamos que quieres usar una enumeración personalizada:

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
}

Regístralo así:

// 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)

Entonces úsalo!

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())

Digamos que quieres usar un registro polimórfico:

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

... y lo usamos así:

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()

Puede escribir un UDT personalizado que codifique todo en bytes (estoy usando la serialización de Java aquí, pero probablemente sea mejor instrumentar el contexto Kryo de Spark).

Primero defina la clase UDT:

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]
}

Entonces regístralo:

// 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)

¡Entonces puedes usarlo!

// 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()

Los codificadores funcionan más o menos de la misma manera Spark2.0. Y Kryosigue siendo la serializationopción recomendada .

Puedes ver el siguiente ejemplo con spark-shell

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.

Hasta ahora] no había appropriate encodersalcance actual, por lo que nuestras personas no estaban codificadas como binaryvalores. Pero eso cambiará una vez que proporcionemos algunos implicitcodificadores utilizando la Kryoserialización.

// 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.

En el caso de la clase Java Bean, esto puede ser útil

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

Ahora puede simplemente leer el DataFrame como DataFrame personalizado

dataFrame.as[MyClass]

Esto creará un codificador de clase personalizado y no uno binario.

Mis ejemplos estarán en Java, pero no creo que sea difícil adaptarse a Scala.

He tenido bastante éxito la conversión RDD<Fruit>a Dataset<Fruit>utilizar spark.createDataset y Encoders.bean el tiempo que Fruites un simple Java Bean .

Paso 1: Crea el Java Bean simple.

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
}

Me apegaría a las clases con tipos primitivos y String como campos antes de que la gente de DataBricks refuerce sus codificadores. Si tiene una clase con objeto anidado, cree otro Java Bean simple con todos sus campos aplanados, para que pueda usar transformaciones RDD para asignar el tipo complejo al más simple.Claro que es un poco de trabajo extra, pero imagino que ayudará mucho en el rendimiento al trabajar con un esquema plano.

Paso 2: Obtenga su conjunto de datos del RDD

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);

¡Y voilá! Enjabonar, enjuagar, repetir.

Para aquellos que puedan en mi situación, pongo mi respuesta aquí también.

Ser especifico,

1. Estaba leyendo 'Establecer datos escritos' de SQLContext. Entonces el formato de datos original es DataFrame.

   val sample = spark.sqlContext.sql("select 1 as a, collect_set(1) as b limit 1") sample.show()

   +---+---+ | a| b| +---+---+ | 1|[1]| +---+---+

2. Luego conviértalo a RDD usando rdd.map () con el tipo mutable.WrappedArray.

   sample .rdd.map(r => (r.getInt(0), r.getAs[mutable.WrappedArray[Int]](1).toSet)) .collect() .foreach(println)

   Resultado:

   (1,Set(1))

Además de las sugerencias ya dadas, otra opción que descubrí recientemente es que puede declarar su clase personalizada, incluido el rasgo org.apache.spark.sql.catalyst.DefinedByConstructorParams.

Esto funciona si la clase tiene un constructor que usa tipos que el ExpressionEncoder puede entender, es decir, valores primitivos y colecciones estándar. Puede ser útil cuando no puede declarar la clase como una clase de caso, pero no quiere usar Kryo para codificarla cada vez que se incluye en un conjunto de datos.

Por ejemplo, quería declarar una clase de caso que incluyera un vector Breeze. El único codificador que podría manejar eso normalmente sería Kryo. Pero si declaraba una subclase que extendía Breeze DenseVector y DefinedByConstructorParams, ExpressionEncoder entendía que podía serializarse como una matriz de Dobles.

Así es como lo declaró:

class SerializableDenseVector(values: Array[Double]) extends breeze.linalg.DenseVector[Double](values) with DefinedByConstructorParams
implicit def BreezeVectorToSerializable(bv: breeze.linalg.DenseVector[Double]): SerializableDenseVector = bv.asInstanceOf[SerializableDenseVector]

Ahora puedo usarlo SerializableDenseVectoren un conjunto de datos (directamente o como parte de un producto) usando un ExpressionEncoder simple y sin Kryo. Funciona igual que un Breeze DenseVector pero se serializa como una matriz [doble].