Selon Présentation des ensembles de données Spark :

Alors que nous attendons avec impatience Spark 2.0, nous prévoyons quelques améliorations intéressantes aux ensembles de données, en particulier: ... Encodeurs personnalisés - alors que nous générons actuellement des encodeurs pour une grande variété de types, nous aimerions ouvrir une API pour les objets personnalisés.

et tente de stocker un type personnalisé dans une Datasetpiste d'erreur suivante comme:

Impossible de trouver le codeur pour le type stocké dans un ensemble de données. Les types primitifs (Int, String, etc.) et les types de produit (classes de cas) sont pris en charge par l'importation de sqlContext.implicits._ La prise en charge de la sérialisation d'autres types sera ajoutée dans les versions futures

ou:

Java.lang.UnsupportedOperationException: aucun encodeur trouvé pour ....

Existe-t-il des solutions de contournement?

Notez que cette question n'existe que comme point d'entrée pour une réponse du wiki communautaire. N'hésitez pas à mettre à jour / améliorer la question et la réponse.

Mettre à jour

Cette réponse est toujours valide et informatif, bien que les choses sont maintenant mieux depuis 2.2 / 2.3, qui ajoute le support intégré de codeur pour Set, Seq, Map, Date, Timestampet BigDecimal. Si vous vous en tenez à créer des types avec uniquement des classes de cas et les types Scala habituels, vous devriez vous contenter de l'implicite in SQLImplicits.

Malheureusement, pratiquement rien n'a été ajouté pour y remédier. Rechercher @since 2.0.0dans Encoders.scalaou SQLImplicits.scalatrouver des choses principalement à voir avec les types primitifs (et quelques ajustements des classes de cas). Donc, première chose à dire: il n'y a actuellement pas vraiment de bon support pour les encodeurs de classe personnalisés . Avec cela à l'écart, ce qui suit est quelques trucs qui font un aussi bon travail que nous pouvons jamais espérer, étant donné ce que nous avons actuellement à notre disposition. En guise d'avertissement initial: cela ne fonctionnera pas parfaitement et je ferai de mon mieux pour clarifier toutes les limitations.

Quel est le problème exactement

Lorsque vous souhaitez créer un ensemble de données, Spark "nécessite un encodeur (pour convertir un objet JVM de type T vers et à partir de la représentation SQL Spark interne) qui est généralement créé automatiquement via des implicits à partir d'un SparkSession, ou peut être créé explicitement en appelant des méthodes statiques on Encoders"(extrait de la documentationcreateDataset ). Un encodeur prendra la forme Encoder[T]où Test le type que vous encodez. La première suggestion est d'ajouter import spark.implicits._(ce qui vous donne ces encodeurs implicites) et la deuxième suggestion est de passer explicitement l'encodeur implicite en utilisant cet ensemble de fonctions liées à l'encodeur.

Il n'y a pas d'encodeur disponible pour les classes régulières, donc

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

vous donnera l'erreur de compilation implicite suivante:

Impossible de trouver le codeur pour le type stocké dans un ensemble de données. Les types primitifs (Int, String, etc.) et les types de produit (classes de cas) sont pris en charge par l'importation de sqlContext.implicits._ La prise en charge de la sérialisation d'autres types sera ajoutée dans les versions futures

Cependant, si vous encapsulez le type que vous venez d'utiliser pour obtenir l'erreur ci-dessus dans une classe qui s'étend Product, l'erreur est retardée de manière déroutante à l'exécution, donc

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

Compile très bien, mais échoue à l'exécution avec

java.lang.UnsupportedOperationException: aucun encodeur trouvé pour MyObj

La raison en est que les encodeurs créés par Spark avec les implicits ne sont en fait créés qu'au moment de l'exécution (via la relfection de scala). Dans ce cas, toutes les vérifications Spark au moment de la compilation sont que la classe la plus externe s'étend Product(ce que font toutes les classes de cas), et ne se rend compte qu'au moment de l'exécution qu'elle ne sait toujours pas quoi faire MyObj(le même problème se produit si j'essaye de faire a Dataset[(Int,MyObj)]- Spark attend que l'exécution soit activée MyObj). Ce sont des problèmes centraux qui doivent absolument être résolus:

• certaines classes qui s'étendent se Productcompilent malgré toujours des plantages à l'exécution
• il n'y a aucun moyen de passer des encodeurs personnalisés pour les types imbriqués (je n'ai aucun moyen de fournir à Spark un encodeur pour MyObjqu'il sache ensuite encoder Wrap[MyObj]ou (Int,MyObj)).

Juste utiliser kryo

La solution que tout le monde suggère est d'utiliser l' kryoencodeur.

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

Cela devient vite assez fastidieux. Surtout si votre code manipule toutes sortes d'ensembles de données, jointures, regroupements, etc. Vous finissez par accumuler un tas d'implicits supplémentaires. Alors, pourquoi ne pas simplement faire un implicite qui fait tout cela automatiquement?

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

Et maintenant, il semble que je puisse faire presque tout ce que je veux (l'exemple ci-dessous ne fonctionnera pas dans l' spark-shellendroit où spark.implicits._est automatiquement importé)

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!

Ou presque. Le problème est que l'utilisation kryode Spark conduit à stocker simplement chaque ligne de l'ensemble de données sous la forme d'un objet binaire plat. Pour map, filter, foreachcela suffit, mais pour des opérations telles que join, Spark a vraiment besoin de ceux - ci soient séparés en colonnes. En inspectant le schéma pour d2ou d3, vous voyez qu'il n'y a qu'une seule colonne binaire:

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

Solution partielle pour les tuples

Donc, en utilisant la magie des implicits dans Scala (plus en 6.26.3 Overloading Resolution ), je peux me créer une série d'implicits qui feront le meilleur travail possible, au moins pour les tuples, et fonctionnera bien avec les implicits existants:

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

Ensuite, armé de ces implicits, je peux faire fonctionner mon exemple ci-dessus, mais avec un changement de nom de colonne

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

Je ne l' ai pas encore compris comment obtenir les noms de tuple attendus ( _1, _2...) par défaut sans les renommer - si quelqu'un d' autre veut jouer avec cela, c'est où le nom se présente et c'est là tuple les noms sont généralement ajoutés. Cependant, le point clé est que j'ai maintenant un joli schéma structuré:"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)

Donc, en résumé, cette solution de contournement:

• nous permet d'obtenir des colonnes séparées pour les tuples (afin que nous puissions rejoindre à nouveau les tuples, yay!)
• on peut à nouveau se fier aux implicits (donc pas besoin de passer kryopartout)
• est presque entièrement rétrocompatible avec import spark.implicits._(avec certains changements de nom impliqués)
• ne nous permet pas de nous joindre sur les kyrocolonnes binaires sérialisées, encore moins sur les champs
• a pour effet secondaire désagréable de renommer certaines des colonnes de tuple en «valeur» (si nécessaire, cela peut être annulé en convertissant .toDF, en spécifiant de nouveaux noms de colonnes et en les reconvertissant en un ensemble de données - et les noms de schéma semblent être préservés grâce aux jointures , là où ils sont le plus nécessaires).

Solution partielle pour les classes en général

Celui-ci est moins agréable et n'a pas de bonne solution. Cependant, maintenant que nous avons la solution de tuple ci-dessus, j'ai le sentiment que la solution de conversion implicite d'une autre réponse sera également un peu moins pénible puisque vous pouvez convertir vos classes plus complexes en tuples. Ensuite, après avoir créé l'ensemble de données, vous renommeriez probablement les colonnes en utilisant l'approche dataframe. Si tout se passe bien, c'est vraiment une amélioration puisque je peux désormais effectuer des jointures sur les champs de mes cours. Si j'avais juste utilisé un kryosérialiseur binaire plat, cela n'aurait pas été possible.

Voici un exemple qui fait un peu de tout: j'ai une classe MyObjqui a des champs de types Int, java.util.UUIDet Set[String]. Le premier prend soin de lui-même. Le second, bien que je puisse sérialiser en utilisant kryoserait plus utile s'il est stocké en tant que String(puisque les UUIDs sont généralement quelque chose contre lequel je voudrais me joindre). Le troisième appartient vraiment à une colonne binaire.

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)

Maintenant, je peux créer un ensemble de données avec un joli schéma en utilisant cette machine:

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]

Et le schéma me montre des colonnes avec les bons noms et avec les deux premières choses contre lesquelles je peux me joindre.

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

1. Utilisation d'encodeurs génériques.

   Il existe deux encodeurs génériques disponibles pour l'instant kryoet javaSerializationoù le dernier est explicitement décrit comme:

   extrêmement inefficace et ne doit être utilisé qu'en dernier recours.

   En supposant la classe suivante

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

   vous pouvez utiliser ces encodeurs en ajoutant un encodeur implicite:

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

   qui peuvent être utilisés ensemble comme suit:

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

   Il stocke les objets sous forme de binarycolonne, de sorte qu'une fois converti, DataFramevous obtenez le schéma suivant:

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

   Il est également possible d'encoder des tuples en utilisant un kryoencodeur pour un champ spécifique:

   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]

   Veuillez noter que nous ne dépendons pas des encodeurs implicites ici, mais transmettons explicitement l'encodeur, donc cela ne fonctionnera probablement pas avec la toDSméthode.

2. Utilisation de conversions implicites:

   Fournissez des conversions implicites entre la représentation qui peut être encodée et la classe personnalisée, par exemple:

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

Questions connexes:

• Comment créer un encodeur pour le constructeur de type Option, par exemple Option [Int]?

Vous pouvez utiliser UDTRegistration, puis les classes de cas, les tuples, etc. fonctionnent tous correctement avec votre type défini par l'utilisateur!

Supposons que vous souhaitiez utiliser un Enum personnalisé:

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
}

Enregistrez-le comme ceci:

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

Alors UTILISEZ-LE!

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

Supposons que vous souhaitiez utiliser un enregistrement polymorphe:

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

... et utilisez-le comme ceci:

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

Vous pouvez écrire un UDT personnalisé qui encode tout en octets (j'utilise la sérialisation java ici mais il est probablement préférable d'instrumenter le contexte Kryo de Spark).

Définissez d'abord la classe 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]
}

Puis enregistrez-le:

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

Ensuite, vous pouvez l'utiliser!

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

Les encodeurs fonctionnent plus ou moins de la même manière dans Spark2.0. Et Kryoc'est toujours le recommandéserialization choix .

Vous pouvez regarder l'exemple suivant avec 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.

Jusqu'à présent] il n'y avait pas appropriate encodersde portée actuelle, donc nos personnes n'étaient pas encodées en tant que binaryvaleurs. Mais cela changera une fois que nous fournirons certains implicitencodeurs utilisant la Kryosérialisation.

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

Dans le cas de la classe Java Bean, cela peut être utile

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

Vous pouvez maintenant simplement lire le dataFrame en tant que DataFrame personnalisé

dataFrame.as[MyClass]

Cela créera un encodeur de classe personnalisé et non un encodeur binaire.

Mes exemples seront en Java, mais je n'imagine pas qu'il soit difficile de s'adapter à Scala.

J'ai réussi à convertir RDD<Fruit>en Dataset<Fruit>utilisant spark.createDataset et Encoders.bean tant qu'il Fruits'agit d'un simple Java Bean .

Étape 1: Créez le simple Java Bean.

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
}

Je m'en tiendrais aux classes avec des types primitifs et String comme champs avant que les gens de DataBricks renforcent leurs encodeurs. Si vous avez une classe avec un objet imbriqué, créez un autre Java Bean simple avec tous ses champs aplatis, de sorte que vous puissiez utiliser les transformations RDD pour mapper le type complexe vers le plus simple.Bien sûr, c'est un peu de travail supplémentaire, mais j'imagine que cela aidera beaucoup sur les performances de travailler avec un schéma plat.

Étape 2: Obtenez votre ensemble de données à partir du 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);

Et voila! Faire mousser, rincer, répéter.

Pour ceux qui peuvent dans ma situation, je mets ma réponse ici aussi.

Pour être précis,

1. Je lisais «Définir les données typées» de SQLContext. Le format de données original est donc DataFrame.

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

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

2. Puis convertissez-le en RDD en utilisant rdd.map () avec le type mutable.WrappedArray.

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

   Résultat:

   (1,Set(1))

En plus des suggestions déjà données, une autre option que j'ai récemment découverte est que vous pouvez déclarer votre classe personnalisée en incluant le trait org.apache.spark.sql.catalyst.DefinedByConstructorParams.

Cela fonctionne si la classe a un constructeur qui utilise des types que ExpressionEncoder peut comprendre, c'est-à-dire des valeurs primitives et des collections standard. Cela peut être utile lorsque vous ne pouvez pas déclarer la classe en tant que classe de cas, mais que vous ne voulez pas utiliser Kryo pour l'encoder chaque fois qu'il est inclus dans un ensemble de données.

Par exemple, je voulais déclarer une classe de cas qui incluait un vecteur Breeze. Le seul encodeur capable de gérer cela serait normalement Kryo. Mais si je déclarais une sous-classe qui étendait le Breeze DenseVector et DefinedByConstructorParams, le ExpressionEncoder comprenait qu'il pouvait être sérialisé comme un tableau de Doubles.

Voici comment je l'ai déclaré:

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]

Maintenant, je peux utiliser SerializableDenseVectordans un ensemble de données (directement ou dans le cadre d'un produit) en utilisant un simple ExpressionEncoder et pas de Kryo. Il fonctionne exactement comme un Breeze DenseVector mais se sérialise comme un Array [Double].