Pourquoi "Impossible de trouver le codeur pour le type stocké dans un jeu de données" lors de la création d'un jeu de données de classe de cas personnalisé?

Question

Spark 2.0 (final) avec Scala 2.11.8. Le code super simple suivant génère l'erreur de compilation Error:(17, 45) Unable to find encoder for type stored in a Dataset. Primitive types (Int, String, etc) and Product types (case classes) are supported by importing spark.implicits._ Support for serializing other types will be added in future releases.

 import org.apache.spark.sql.SparkSession

case class SimpleTuple(id: Int, desc: String)

object DatasetTest {
  val dataList = List(
    SimpleTuple(5, "abc"),
    SimpleTuple(6, "bcd")
  )

  def main(args: Array[String]): Unit = {
    val sparkSession = SparkSession.builder.
      master("local")
      .appName("example")
      .getOrCreate()

    val dataset = sparkSession.createDataset(dataList)
  }
}
 

Answer 1

Spark Datasets besoin Encoders pour le type de données qui est sur le point d'être stockées. Les types courants (atomics, des types de produits) il y a un nombre prédéfini de codeurs disponibles mais vous devez d'abord importer ces de SparkSession.implicits à le faire fonctionner:

val sparkSession: SparkSession = ???
import sparkSession.implicits._
val dataset = sparkSession.createDataset(dataList)

Pour en savoir plus:

• Pour des objets personnalisés qui ne sont pas couverts par le système intégré de codeurs voir Comment stocker des objets personnalisés dans le Dataset?
• Pour Row des objets que vous avez à offrir Encoder explicitement, comme montré dans l'Encodeur d'erreur tout en essayant de carte dataframe ligne à ligne mise à jour

Answer 2

Pour les autres utilisateurs (le vôtre est correct), notez que c'est aussi important que l' case class est définie en dehors de l' object de la portée. Donc:

Échoue:

object DatasetTest {
  case class SimpleTuple(id: Int, desc: String)

  val dataList = List(
    SimpleTuple(5, "abc"),
    SimpleTuple(6, "bcd")
  )

  def main(args: Array[String]): Unit = {
    val sparkSession = SparkSession.builder
      .master("local")
      .appName("example")
      .getOrCreate()
    val dataset = sparkSession.createDataset(dataList)
  }
}

Ajouter le implicites, échoue toujours avec le même message d'erreur:

object DatasetTest {
  case class SimpleTuple(id: Int, desc: String)

  val dataList = List(
    SimpleTuple(5, "abc"),
    SimpleTuple(6, "bcd")
  )

  def main(args: Array[String]): Unit = {
    val sparkSession = SparkSession.builder
      .master("local")
      .appName("example")
      .getOrCreate()

    import sparkSession.implicits._
    val dataset = sparkSession.createDataset(dataList)
  }
}

Œuvres:

case class SimpleTuple(id: Int, desc: String)

object DatasetTest {   
  val dataList = List(
    SimpleTuple(5, "abc"),
    SimpleTuple(6, "bcd")
  )

  def main(args: Array[String]): Unit = {
    val sparkSession = SparkSession.builder
      .master("local")
      .appName("example")
      .getOrCreate()

    import sparkSession.implicits._
    val dataset = sparkSession.createDataset(dataList)
  }
}

Voici le bogue correspondant: https://issues.apache.org/jira/browse/SPARK-13540nous espérons que cela sera corrigé dans la prochaine version de Spark 2.

(Edit: il Semble que la correction est en fait Spark 2.0.0... Donc je ne suis pas sûr de savoir pourquoi cela ne fonctionne toujours).

Answer 3

Je clarifierais avec une réponse à ma propre question, que si l'objectif est de définir un cadre SparkData littéral simple, plutôt que d'utiliser des tuples Scala et une conversion implicite, la voie la plus simple consiste à utiliser l'API Spark directement comme ceci:

   import org.apache.spark.sql._
  import org.apache.spark.sql.types._
  import scala.collection.JavaConverters._

  val simpleSchema = StructType(
    StructField("a", StringType) ::
    StructField("b", IntegerType) ::
    StructField("c", IntegerType) ::
    StructField("d", IntegerType) ::
    StructField("e", IntegerType) :: Nil)

  val data = List(
    Row("001", 1, 0, 3, 4),
    Row("001", 3, 4, 1, 7),
    Row("001", null, 0, 6, 4),
    Row("003", 1, 4, 5, 7),
    Row("003", 5, 4, null, 2),
    Row("003", 4, null, 9, 2),
    Row("003", 2, 3, 0, 1)
  )

  val df = spark.createDataFrame(data.asJava, simpleSchema)