scala generic encoder for spark case class

Question

How can I get this method to compile. Strangely, sparks implicit are already imported.

def loadDsFromHive[T <: Product](tableName: String, spark: SparkSession): Dataset[T] = {
    import spark.implicits._
    spark.sql(s"SELECT * FROM $tableName").as[T]
  }

This is the error:

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.
[error]     spark.sql(s"SELECT * FROM $tableName").as[T]

Ruisdael

What are other types used in Product apart from primitives? – Mariquilla

I want to use a case class and only use primitives. But this snippet above literally will not compile. – Ruisdael

Answer 1

According to the source code for org.apache.spark.sql.SQLImplicits, you need the type class TypeTag for your type, in order for the implicit Encoder to exist:

import scala.reflect.runtime.universe.TypeTag
def loadDsFromHive[T <: Product: TypeTag](tableName: String, spark: SparkSession): Dataset[T] = ...

Libre

In my case T is a class-level type parameter. I've added : TypeTag but I'm still getting that message (I am also importing implicits) – Odine

Needed to add Encoder as well, i.e. [T: Encoder : TypeTag]. – Odine

Adding Encoder as @Odine suggested helps, but it will still generate errors like Unable to find encoder for type ExtendedProduct. One workaround is to have the class T be an inline class, so that you can use the usual import spark.implicits._ which requires a reference to spark as a spark session. E.g., import spark.implicits._<br/> new Product() {<br/> override def ...<br/> }<br/> – Nihhi

def makeDs[T <: Product: ClassTag: Encoder](ds: Dataset[T]): Dataset[(T, String)] = { ds.map { case (t: T) => t -> "hello" } } still doesn't compile. I don't understand. :( – Dessiatine